diff --git a/pg_backup_api/pg_backup_api/logic/utility_controller.py b/pg_backup_api/pg_backup_api/logic/utility_controller.py index 295c16b..b84c788 100644 --- a/pg_backup_api/pg_backup_api/logic/utility_controller.py +++ b/pg_backup_api/pg_backup_api/logic/utility_controller.py @@ -189,15 +189,15 @@ def servers_operations_post(server_name: str, if op_type == OperationType.RECOVERY: try: - backup_id = request_body["backup_id"] + msg_backup_id = request_body["backup_id"] except KeyError: msg_400 = "Request body is missing ``backup_id``" abort(400, description=msg_400) - backup_id = parse_backup_id(Server(server), backup_id) + backup_id = parse_backup_id(Server(server), msg_backup_id) if not backup_id: - msg_404 = f"Backup '{backup_id}' does not exist" + msg_404 = f"Backup '{msg_backup_id}' does not exist" abort(404, description=msg_404) operation = RecoveryOperation(server_name) diff --git a/pg_backup_api/pg_backup_api/server_operation.py b/pg_backup_api/pg_backup_api/server_operation.py index fb59b1d..b79c143 100644 --- a/pg_backup_api/pg_backup_api/server_operation.py +++ b/pg_backup_api/pg_backup_api/server_operation.py @@ -41,8 +41,6 @@ if TYPE_CHECKING: # pragma: no cover from barman.config import Config as BarmanConfig -load_barman_config() - logging.basicConfig(stream=sys.stdout, level=logging.DEBUG) log = logging.getLogger() @@ -117,6 +115,8 @@ def __init__(self, name: str) -> None: f"No barman config found for '{name}'." ) + load_barman_config() + if TYPE_CHECKING: # pragma: no cover assert isinstance(barman.__config__, BarmanConfig) diff --git a/pg_backup_api/pg_backup_api/tests/test_main.py b/pg_backup_api/pg_backup_api/tests/test_main.py new file mode 100644 index 0000000..9f072f4 --- /dev/null +++ b/pg_backup_api/pg_backup_api/tests/test_main.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +# © Copyright EnterpriseDB UK Limited 2021-2023 +# +# This file is part of Postgres Backup API. +# +# Postgres Backup API is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Postgres Backup API is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Postgres Backup API. If not, see . + +"""Unit tests for the CLI.""" +from textwrap import dedent +from unittest.mock import MagicMock, patch + +import pytest + +from pg_backup_api.__main__ import main + + +_HELP_OUTPUT = { + "pg-backup-api --help": dedent("""\ + usage: pg-backup-api [-h] {serve,status,recovery} ... + + positional arguments: + {serve,status,recovery} + + options: + -h, --help show this help message and exit + + Postgres Backup API by EnterpriseDB (www.enterprisedb.com) +\ + """), + "pg-backup-api serve --help": dedent("""\ + usage: pg-backup-api serve [-h] [--port PORT] + + Start the REST API server. Listen for requests on '127.0.0.1', on the given port. + + options: + -h, --help show this help message and exit + --port PORT Port to bind to. +\ + """), # noqa: E501 + "pg-backup-api status --help": dedent("""\ + usage: pg-backup-api status [-h] [--port PORT] + + Check if the REST API server is up and running + + options: + -h, --help show this help message and exit + --port PORT Port to be checked. +\ + """), # noqa: E501 + "pg-backup-api recovery --help": dedent("""\ + usage: pg-backup-api recovery [-h] --server-name SERVER_NAME --operation-id OPERATION_ID + + Perform a 'barman recover' through the 'pg-backup-api'. Can only be run if a recover operation has been previously registered. + + options: + -h, --help show this help message and exit + --server-name SERVER_NAME + Name of the Barman server to be recovered. + --operation-id OPERATION_ID + ID of the operation in the 'pg-backup-api'. +\ + """), # noqa: E501 +} + +_COMMAND_FUNC = { + "pg-backup-api serve": "serve", + "pg-backup-api status": "status", + "pg-backup-api recovery --server-name SOME_SERVER --operation-id SOME_OP_ID": "recovery_operation", # noqa: E501 +} + + +@pytest.mark.parametrize("command", _HELP_OUTPUT.keys()) +def test_main_helper(command, capsys): + """Test :func:`main`. + + Ensure all the ``--help`` calls print the expected content to the console. + """ + with patch("sys.argv", command.split()), pytest.raises(SystemExit) as exc: + main() + + assert str(exc.value) == "0" + + assert capsys.readouterr().out == _HELP_OUTPUT[command] + + +@pytest.mark.parametrize("command", _COMMAND_FUNC.keys()) +@pytest.mark.parametrize("output", [None, "SOME_OUTPUT"]) +@pytest.mark.parametrize("success", [False, True]) +def test_main_funcs(command, output, success, capsys): + """Test :func:`main`. + + Ensure :func:`main` executes the expected functions, print the expected + messages, and exits with the expected codes. + """ + mock_controller = patch(f"pg_backup_api.__main__.{_COMMAND_FUNC[command]}") + mock_func = mock_controller.start() + + mock_func.return_value = (output, success) + + with patch("sys.argv", command.split()), pytest.raises(SystemExit) as exc: + main() + + mock_controller.stop() + + assert capsys.readouterr().out == (f"{output}\n" if output else "") + assert str(exc.value) == ("0" if success else "-1") + + +@patch("argparse.ArgumentParser.parse_args") +def test_main_with_func(mock_parse_args, capsys): + """Test :func:`main`. + + Ensure :func:`main` calls the function with the expected arguments, if a + command has a function associated with it. + """ + mock_parse_args.return_value.func = MagicMock() + mock_func = mock_parse_args.return_value.func + mock_func.return_value = ("SOME_OUTPUT", True) + + with pytest.raises(SystemExit) as exc: + main() + + capsys.readouterr() # so we don't write to stdout during unit tests + + mock_func.assert_called_once_with(mock_parse_args.return_value) + assert str(exc.value) == "0" + + +@patch("argparse.ArgumentParser.print_help") +@patch("argparse.ArgumentParser.parse_args") +def test_main_without_func(mock_parse_args, mock_print_help, capsys): + """Test :func:`main`. + + Ensure :func:`main` prints a helper if a command has no function associated + with it. + """ + delattr(mock_parse_args.return_value, "func") + + with pytest.raises(SystemExit) as exc: + main() + + capsys.readouterr() # so we don't write to stdout during unit tests + + mock_print_help.assert_called_once_with() + assert str(exc.value) == "0" diff --git a/pg_backup_api/pg_backup_api/tests/test_run.py b/pg_backup_api/pg_backup_api/tests/test_run.py new file mode 100644 index 0000000..255cc35 --- /dev/null +++ b/pg_backup_api/pg_backup_api/tests/test_run.py @@ -0,0 +1,123 @@ +# -*- coding: utf-8 -*- +# © Copyright EnterpriseDB UK Limited 2021-2023 +# +# This file is part of Postgres Backup API. +# +# Postgres Backup API is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Postgres Backup API is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Postgres Backup API. If not, see . + +"""Unit tests for functions used by the CLI.""" + +import argparse +from requests.exceptions import ConnectionError +from unittest.mock import MagicMock, patch, call + +import pytest + +from pg_backup_api.run import serve, status, recovery_operation + + +@pytest.mark.parametrize("port", [7480, 7481]) +@patch("pg_backup_api.run.output") +@patch("pg_backup_api.run.load_barman_config") +@patch("pg_backup_api.run.app") +def test_serve(mock_app, mock_load_config, mock_output, port): + """Test :func:`serve`. + + Ensure :func:`serve` performs the expected calls and return the expected + values. + """ + mock_output.AVAILABLE_WRITERS.__getitem__.return_value = MagicMock() + expected = mock_output.AVAILABLE_WRITERS.__getitem__.return_value + expected.return_value = MagicMock() + + args = argparse.Namespace(port=port) + + assert serve(args) == (mock_app.run.return_value, True) + + mock_load_config.assert_called_once_with() + mock_output.set_output_writer.assert_called_once_with( + expected.return_value, + ) + mock_app.run.assert_called_once_with(host="127.0.0.1", port=port) + + +@pytest.mark.parametrize("port", [7480, 7481]) +@patch("requests.get") +def test_status_ok(mock_request, port): + """Test :func:`status`. + + Ensure the expected ``GET`` request is performed, and that :func:`status` + returns `OK` when the API is available. + """ + args = argparse.Namespace(port=port) + + assert status(args) == ("OK", True) + + mock_request.assert_called_once_with(f"http://127.0.0.1:{port}/status") + + +@pytest.mark.parametrize("port", [7480, 7481]) +@patch("requests.get") +def test_status_failed(mock_request, port): + """Test :func:`status`. + + Ensure the expected ``GET`` request is performed, and that :func:`status` + returns an error message when the API is not available. + """ + args = argparse.Namespace(port=port) + + mock_request.side_effect = ConnectionError("Some Error") + + message = "The Postgres Backup API does not appear to be available." + assert status(args) == (message, False) + + mock_request.assert_called_once_with(f"http://127.0.0.1:{port}/status") + + +@pytest.mark.parametrize("server_name", ["SERVER_1", "SERVER_2"]) +@pytest.mark.parametrize("operation_id", ["OPERATION_1", "OPERATION_2"]) +@pytest.mark.parametrize("rc", [0, 1]) +@patch("pg_backup_api.run.RecoveryOperation") +def test_recovery_operation(mock_rec_op, server_name, operation_id, rc): + """Test :func:`recovery_operation`. + + Ensure the operation is created and executed, and that the expected values + are returned depending on the return code. + """ + args = argparse.Namespace(server_name=server_name, + operation_id=operation_id) + + mock_rec_op.return_value.run.return_value = ("SOME_OUTPUT", rc) + mock_write_output = mock_rec_op.return_value.write_output_file + mock_time_event = mock_rec_op.return_value.time_event_now + mock_read_job = mock_rec_op.return_value.read_job_file + + assert recovery_operation(args) == (mock_write_output.return_value, + rc == 0) + + mock_rec_op.assert_called_once_with(server_name, operation_id) + mock_rec_op.return_value.run.assert_called_once_with() + mock_time_event.assert_called_once_with() + mock_read_job.assert_called_once_with() + + # Make sure the expected content was added to `read_job_file` output before + # writing it to the output file. + assert len(mock_read_job.return_value.__setitem__.mock_calls) == 3 + mock_read_job.return_value.__setitem__.assert_has_calls([ + call('success', rc == 0), + call('end_time', mock_time_event.return_value), + call('output', "SOME_OUTPUT"), + ]) + + mock_write_output.assert_called_once_with(mock_read_job.return_value) diff --git a/pg_backup_api/pg_backup_api/tests/test_server_operation.py b/pg_backup_api/pg_backup_api/tests/test_server_operation.py index fa1eb5c..6d55003 100644 --- a/pg_backup_api/pg_backup_api/tests/test_server_operation.py +++ b/pg_backup_api/pg_backup_api/tests/test_server_operation.py @@ -1,8 +1,28 @@ +# -*- coding: utf-8 -*- +# © Copyright EnterpriseDB UK Limited 2021-2023 +# +# This file is part of Postgres Backup API. +# +# Postgres Backup API is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Postgres Backup API is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Postgres Backup API. If not, see . + +"""Unit tests for the classes related with REST API operations.""" import os import subprocess -from unittest import TestCase from unittest.mock import Mock, MagicMock, call, patch +import pytest + from pg_backup_api.server_operation import ( OperationServer, MalformedContent, @@ -17,287 +37,364 @@ _BARMAN_SERVER = "BARMAR_SERVER" -@patch("pg_backup_api.server_operation.get_server_by_name", Mock()) -@patch("barman.__config__.barman_home", _BARMAN_HOME) -class TestOperationServer(TestCase): - - @patch("pg_backup_api.server_operation.get_server_by_name", Mock()) - @patch("barman.__config__.barman_home", _BARMAN_HOME) - def setUp(self): - with patch.object(OperationServer, "_create_dir"): - self.server = OperationServer(_BARMAN_SERVER) +class TestOperationServer: + """Run tests for :class:`OperationServer`.""" - def test___init__(self): + @pytest.fixture + @patch("pg_backup_api.server_operation.get_server_by_name", Mock()) + @patch("pg_backup_api.server_operation.load_barman_config", Mock()) + @patch.object(OperationServer, "_create_dir", Mock()) + def op_server(self): + """Create a :class:`OperationServer` instance for testing. + + :return: :class:`OperationServer` instance for testing. + """ + with patch("barman.__config__") as mock_config: + mock_config.barman_home = _BARMAN_HOME + return OperationServer(_BARMAN_SERVER) + + def test___init__(self, op_server): + """Test :meth:`OperationServer.__init__`. + + Ensure its attributes are set as expected. + """ # Ensure name is as expected. - self.assertEqual(self.server.name, _BARMAN_SERVER) + assert op_server.name == _BARMAN_SERVER # Ensure "jobs" directory is created in expected path. - self.assertEqual( - self.server.jobs_basedir, - os.path.join(_BARMAN_HOME, _BARMAN_SERVER, "jobs"), - ) + expected = os.path.join(_BARMAN_HOME, _BARMAN_SERVER, "jobs") + assert op_server.jobs_basedir == expected # Ensure "output" directory is created in the expected path. - self.assertEqual( - self.server.output_basedir, - os.path.join(_BARMAN_HOME, _BARMAN_SERVER, "output"), - ) + expected = os.path.join(_BARMAN_HOME, _BARMAN_SERVER, "output") + assert op_server.output_basedir == expected - @patch("os.makedirs") @patch("os.path.isdir") @patch("os.path.exists") - def test__create_dir(self, mock_exists, mock_isdir, mock_makedirs): + def test__create_dir_file_exists(self, mock_exists, mock_isdir, op_server): + """Test :meth:`OperationServer._create_dir`. + + Ensure an exception is raised if the path already exists as a file. + """ dir_path = "/SOME/DIR" - # Ensure an exception is raised if path already exists as a file. mock_exists.return_value = True mock_isdir.return_value = False - with self.assertRaises(NotADirectoryError) as exc: - self.server._create_dir(dir_path) + with pytest.raises(NotADirectoryError) as exc: + op_server._create_dir(dir_path) mock_exists.assert_called_once_with(dir_path) mock_isdir.assert_called_once_with(dir_path) - self.assertEqual( - str(exc.exception), - f"'{dir_path}' exists but it is not a directory", - ) + expected = f"'{dir_path}' exists but it is not a directory" + assert str(exc.value) == expected + + @patch("os.path.isdir") + @patch("os.path.exists") + def test__create_dir_directory_exists(self, mock_exists, mock_isdir, + op_server): + """Test :meth:`OperationServer._create_dir`. - # Ensure no exception occurs if directory already exists. - mock_exists.reset_mock() - mock_isdir.reset_mock() + Ensure no exception occurs if the directory already exists. + """ + dir_path = "/SOME/DIR" mock_exists.return_value = True mock_isdir.return_value = True - self.server._create_dir(dir_path) + op_server._create_dir(dir_path) mock_exists.assert_called_once_with(dir_path) mock_isdir.assert_called_once_with(dir_path) - # Ensure directory is created if missing. - mock_exists.reset_mock() - mock_isdir.reset_mock() + @patch("os.makedirs") + @patch("os.path.isdir") + @patch("os.path.exists") + def test__create_dir_ok(self, mock_exists, mock_isdir, mock_makedirs, + op_server): + """Test :meth:`OperationServer._create_dir`. + + Ensure the directory is created if it's missing. + """ + dir_path = "/SOME/DIR" mock_exists.return_value = False - self.server._create_dir(dir_path) + op_server._create_dir(dir_path) mock_exists.assert_called_once_with(dir_path) mock_isdir.assert_not_called() mock_makedirs.assert_called_once_with(dir_path) - def test__create_jobs_dir(self): - # Ensure "_create_dir" is called. - with patch.object(self.server, "_create_dir") as mock: - self.server._create_jobs_dir() - mock.assert_called_once_with(self.server.jobs_basedir) - - def test__create_output_dir(self): - # Ensure "_create_dir" is called. - with patch.object(self.server, "_create_dir") as mock: - self.server._create_output_dir() - mock.assert_called_once_with(self.server.output_basedir) - - def test_get_job_file_path(self): - # Ensure returns expected file path. + def test__create_jobs_dir(self, op_server): + """Test :meth:`OperationServer._create_jobs_dir`. + + Ensure :meth:`OperationServer._create_dir` is called as expected. + """ + with patch.object(op_server, "_create_dir") as mock_create_dir: + op_server._create_jobs_dir() + mock_create_dir.assert_called_once_with(op_server.jobs_basedir) + + def test__create_output_dir(self, op_server): + """Test :meth:`OperationServer._create_output_dir`. + + Ensure :meth:`OperationServer._create_dir` is called as expected. + """ + with patch.object(op_server, "_create_dir") as mock_create_dir: + op_server._create_output_dir() + mock_create_dir.assert_called_once_with(op_server.output_basedir) + + def test_get_job_file_path(self, op_server): + """Test :meth:`OperationServer.get_job_file_path`. + + Ensure it returns the expected file path. + """ id = "SOME_OP_ID" - self.assertEqual( - self.server.get_job_file_path(id), - os.path.join(self.server.jobs_basedir, f"{id}.json") - ) - - def test_get_output_file_path(self): - # Ensure returns expected file path. + expected = os.path.join(op_server.jobs_basedir, f"{id}.json") + assert op_server.get_job_file_path(id) == expected + + def test_get_output_file_path(self, op_server): + """Test :meth:`OperationServer.get_output_file_path`. + + Ensure it returns the expected file path. + """ id = "SOME_OP_ID" - self.assertEqual( - self.server.get_output_file_path(id), - os.path.join(self.server.output_basedir, f"{id}.json") - ) - - @patch("json.dump") - @patch("builtins.open") + expected = os.path.join(op_server.output_basedir, f"{id}.json") + assert op_server.get_output_file_path(id) == expected + @patch("os.path.exists") - def test__write_file(self, mock_exists, mock_open, mock_dump): + def test__write_file_file_already_exists(self, mock_exists, op_server): + """Test :meth:`OperationServer._write_file`. + + Ensure an exception is raised if the file path already exists. + """ file_path = "/SOME/FILE" file_content = {"SOME": "CONTENT"} - # Ensure exception is raised if file path already exists. mock_exists.return_value = True - with self.assertRaises(FileExistsError) as exc: - self.server._write_file(file_path, file_content) + with pytest.raises(FileExistsError) as exc: + op_server._write_file(file_path, file_content) mock_exists.assert_called_once_with(file_path) - self.assertEqual( - str(exc.exception), - f"File '{file_path}' already exists", - ) + assert str(str(exc.value)) == f"File '{file_path}' already exists" + + @patch("json.dump") + @patch("builtins.open") + @patch("os.path.exists") + def test__write_file_ok(self, mock_exists, mock_open, mock_dump, + op_server): + """Test :meth:`OperationServer._write_file`. + + Ensure the file is created with the expected content. + """ + file_path = "/SOME/FILE" + file_content = {"SOME": "CONTENT"} - # Ensure file is created with expected content. - mock_exists.reset_mock() mock_open.return_value.__enter__.return_value = "SOME_FILE_DESCRIPTOR" mock_exists.return_value = False - self.server._write_file(file_path, file_content) + op_server._write_file(file_path, file_content) mock_open.assert_called_once_with(file_path, "w") mock_dump.assert_called_once_with(file_content, "SOME_FILE_DESCRIPTOR") - def test_write_job_file(self): + @pytest.mark.parametrize("content,missing_keys", [ + ({}, "operation_type, start_time",), + ({"operation_type": "SOME_OPERATION_TYPE"}, "start_time"), + ({"start_time": "SOME_START_TIME"}, "operation_type",), + ]) + def test_write_job_file_content_missing_keys(self, content, missing_keys, + op_server): + """Test :meth:`OperationServer.write_job_file`. + + Ensure an exception is raised if the content is missing keys. + """ id = "SOME_OP_ID" - content = {} - # Ensure exception is raised if content is missing keys -- test 1. - with self.assertRaises(MalformedContent) as exc: - self.server.write_job_file(id, content) + with pytest.raises(MalformedContent) as exc: + op_server.write_job_file(id, content) - self.assertEqual( - str(exc.exception), + expected = ( f"Job file for operation '{id}' is missing required " - f"keys: operation_type, start_time", + f"keys: {missing_keys}" ) + assert str(exc.value) == expected - # Ensure exception is raised if content is missing keys -- test 2. - content["operation_type"] = "SOME_OPERATION_TYPE" + def test_write_job_file_file_already_exists(self, op_server): + """Test :meth:`OperationServer.write_job_file`. - with self.assertRaises(MalformedContent) as exc: - self.server.write_job_file(id, content) + Ensure an exception is raised if the file already exists. + """ + content = { + "operation_type": "SOME_OPERATION_TYPE", + "start_time": "SOME_START_TIME", + } - self.assertEqual( - str(exc.exception), - f"Job file for operation '{id}' is missing required " - f"keys: start_time", - ) + with patch.object(op_server, "_write_file") as mock_write_file: + mock_write_file.side_effect = FileExistsError - # Ensure exception is raised if file already exists. - content["start_time"] = "SOME_START_TIME" + with pytest.raises(FileExistsError) as exc: + op_server.write_job_file(id, content) - with patch.object(self.server, "_write_file") as mock: - mock.side_effect = FileExistsError - - with self.assertRaises(FileExistsError) as exc: - self.server.write_job_file(id, content) + expected = f"Job file for operation '{id}' already exists" + assert str(exc.value) == expected - self.assertEqual( - str(exc.exception), - f"Job file for operation '{id}' already exists", - ) + def test_write_job_file_ok(self, op_server): + """Test :meth:`OperationServer.write_job_file`. - # Ensure file is written if everything is fine. - with patch.object(self.server, "_write_file") as mock: - self.server.write_job_file(id, content) + Ensure the file is written if everything is fine. + """ + content = { + "operation_type": "SOME_OPERATION_TYPE", + "start_time": "SOME_START_TIME", + } - mock.assert_called_once_with( - self.server.get_job_file_path(id), + with patch.object(op_server, "_write_file") as mock_write_file: + op_server.write_job_file(id, content) + + mock_write_file.assert_called_once_with( + op_server.get_job_file_path(id), content, ) - def test_write_output_file(self): + @pytest.mark.parametrize("content,missing_keys", [ + ({}, "end_time, output, success",), + ({"end_time": "SOME_END_TIME"}, "output, success"), + ({"output": "SOME_OUTPUT"}, "end_time, success",), + ({"success": "SOME_SUCCESS"}, "end_time, output",), + ({"end_time": "SOME_END_TIME", "output": "SOME_OUTPUT"}, "success"), + ({"end_time": "SOME_END_TIME", "success": "SOME_SUCCESS"}, "output"), + ({"output": "SOME_OUTPUT", "success": "SOME_SUCCESS"}, "end_time"), + ]) + def test_write_output_file_content_missing_keys(self, content, + missing_keys, op_server): + """Test :meth:`OperationServer.write_output_file`. + + Ensure an exception is raised if the content is missing keys. + """ id = "SOME_OP_ID" - content = {} - # Ensure exception is raised if content is missing keys -- test 1. - with self.assertRaises(MalformedContent) as exc: - self.server.write_output_file(id, content) + with pytest.raises(MalformedContent) as exc: + op_server.write_output_file(id, content) - self.assertEqual( - str(exc.exception), + expected = ( f"Output file for operation '{id}' is missing required " - f"keys: end_time, output, success", + f"keys: {missing_keys}" ) + assert str(exc.value) == expected - # Ensure exception is raised if content is missing keys -- test 2. - content["end_time"] = "SOME_END_TIME" - - with self.assertRaises(MalformedContent) as exc: - self.server.write_output_file(id, content) - - self.assertEqual( - str(exc.exception), - f"Output file for operation '{id}' is missing required " - f"keys: output, success", - ) + def test_write_output_file_file_already_exists(self, op_server): + """Test :meth:`OperationServer.write_output_file`. - # Ensure exception is raised if content is missing keys -- test 3. - content["output"] = "SOME_OUTPUT" + Ensure an exception is raised if the file already exists. + """ + content = { + "end_time": "SOME_END_TIME", + "output": "SOME_OUTPUT", + "success": "SOME_SUCCESS", + } - with self.assertRaises(MalformedContent) as exc: - self.server.write_output_file(id, content) + with patch.object(op_server, "_write_file") as mock_write_file: + mock_write_file.side_effect = FileExistsError - self.assertEqual( - str(exc.exception), - f"Output file for operation '{id}' is missing required " - f"keys: success", - ) + with pytest.raises(FileExistsError) as exc: + op_server.write_output_file(id, content) - # Ensure exception is raised if file already exists. - content["success"] = "SOME_SUCCESS" + expected = f"Output file for operation '{id}' already exists" + assert str(exc.value) == expected - with patch.object(self.server, "_write_file") as mock: - mock.side_effect = FileExistsError - - with self.assertRaises(FileExistsError) as exc: - self.server.write_output_file(id, content) + def test_write_output_file_ok(self, op_server): + """Test :meth:`OperationServer.write_output_file`. - self.assertEqual( - str(exc.exception), - f"Output file for operation '{id}' already exists", - ) + Ensure the file is written if everything is fine. + """ + content = { + "end_time": "SOME_END_TIME", + "output": "SOME_OUTPUT", + "success": "SOME_SUCCESS", + } - # Ensure file is written if everything is fine - with patch.object(self.server, "_write_file") as mock: - self.server.write_output_file(id, content) + with patch.object(op_server, "_write_file") as mock_write_file: + op_server.write_output_file(id, content) - mock.assert_called_once_with( - self.server.get_output_file_path(id), + mock_write_file.assert_called_once_with( + op_server.get_output_file_path(id), content, ) @patch("json.load") @patch("builtins.open") - def test__read_file(self, mock_open, mock_load): + def test__read_file(self, mock_open, mock_load, op_server): + """Test :meth:`OperationServer._read_file`. + + Ensure the file is read and its content is parsed from JSON. + """ file_path = "/SOME/FILE" - # Ensure file is created with expected content. + mock_open.return_value.__enter__.return_value = "SOME_FILE_DESCRIPTOR" - self.server._read_file(file_path) + op_server._read_file(file_path) mock_open.assert_called_once_with(file_path) mock_load.assert_called_once_with("SOME_FILE_DESCRIPTOR") - def test_read_job_file(self): + def test_read_job_file_file_does_not_exist(self, op_server): + """Test :meth:`OperationServer._read_job_file`. + + Ensure an exception is raised if the file does not exist. + """ + id = "SOME_OP_ID" + + with patch.object(op_server, "_read_file") as mock_read_file: + mock_read_file.side_effect = FileNotFoundError + + with pytest.raises(FileNotFoundError) as exc: + op_server.read_job_file(id) + + expected = f"Job file for operation '{id}' does not exist" + assert str(exc.value) == expected + + def test_read_job_file_ok(self, op_server): + """Test :meth:`OperationServer._read_job_file`. + + Ensure its content is retrieved if everything is fine. + """ id = "SOME_OP_ID" content = { "operation_type": "SOME_OPERATION_TYPE", "start_time": "SOME_START_TIME", } - # Ensure exception is raised if file does not exist. - with patch.object(self.server, "_read_file") as mock: - mock.side_effect = FileNotFoundError - - with self.assertRaises(FileNotFoundError) as exc: - self.server.read_job_file(id) + with patch.object(op_server, "_read_file") as mock_read_file: + mock_read_file.return_value = content - self.assertEqual( - str(exc.exception), - f"Job file for operation '{id}' does not exist", - ) + assert op_server.read_job_file(id) == content - # Ensure content is retrieved if everything is fine. - with patch.object(self.server, "_read_file") as mock: - mock.return_value = content + def test_read_output_file_file_does_not_exist(self, op_server): + """Test :meth:`OperationServer._read_output_file`. - self.assertEqual( - self.server.read_job_file(id), - content, - ) + Ensure and exception is raised if the file does not exist. + """ + id = "SOME_OP_ID" + + with patch.object(op_server, "_read_file") as mock_read_file: + mock_read_file.side_effect = FileNotFoundError + + with pytest.raises(FileNotFoundError) as exc: + op_server.read_output_file(id) - def test_read_output_file(self): + expected = f"Output file for operation '{id}' does not exist" + str(exc.value) == expected + + def test_read_output_file_ok(self, op_server): + """Test :meth:`OperationServer._read_output_file`. + + Ensure its content is retrieved if everything is fine. + """ id = "SOME_OP_ID" content = { "success": "SOME_SUCCESS", @@ -305,59 +402,49 @@ def test_read_output_file(self): "output": "SOME_OUTPUT", } - # Ensure exception is raised if file does not exist. - with patch.object(self.server, "_read_file") as mock: - mock.side_effect = FileNotFoundError - - with self.assertRaises(FileNotFoundError) as exc: - self.server.read_output_file(id) + with patch.object(op_server, "_read_file") as mock_read_file: + mock_read_file.return_value = content - self.assertEqual( - str(exc.exception), - f"Output file for operation '{id}' does not exist", - ) + assert op_server.read_output_file(id) == content - # Ensure content is retrieved if everything is fine. - with patch.object(self.server, "_read_file") as mock: - mock.return_value = content - - self.assertEqual( - self.server.read_output_file(id), - content, - ) - - @patch("pg_backup_api.server_operation.OperationServer.read_job_file") @patch("os.listdir") - def test_get_operations_list(self, mock_listdir, mock_read_job_file): - # Ensure it returns an empty list if there are no job files. + def test_get_operations_list_empty_list(self, mock_listdir, op_server): + """Test :meth:`OperationServer.get_operations_list`. + + Ensure it returns an empty list if there are no job files. + """ mock_listdir.return_value = [] - self.assertEqual( - self.server.get_operations_list(), - [], - ) + assert op_server.get_operations_list() == [] - mock_listdir.assert_called_once_with(self.server.jobs_basedir) + mock_listdir.assert_called_once_with(op_server.jobs_basedir) - # Ensure non-JSON files are not considered. - mock_listdir.reset_mock() + @patch("os.listdir") + def test_get_operations_list_ignore_non_json_files(self, mock_listdir, + op_server): + """Test :meth:`OperationServer.get_operations_list`. + Ensure non-JSON files are not considered. + """ mock_listdir.return_value = [ "SOME_OPERATION_1.txt", "SOME_OPERATION_2.xml", "SOME_OPERATION_3.png", ] - self.assertEqual( - self.server.get_operations_list(), - [], - ) + assert op_server.get_operations_list() == [] - mock_listdir.assert_called_once_with(self.server.jobs_basedir) + mock_listdir.assert_called_once_with(op_server.jobs_basedir) - # Ensure expected operations are returned if no filters. - mock_listdir.reset_mock() + @patch("pg_backup_api.server_operation.OperationServer.read_job_file") + @patch("os.listdir") + def test_get_operations_list_with_no_filters(self, mock_listdir, + mock_read_job_file, + op_server): + """Test :meth:`OperationServer.get_operations_list`. + Ensure expected operations are returned if no filters are applied. + """ mock_listdir.return_value = [ "SOME_OPERATION_1.json", "SOME_OPERATION_2.json", @@ -368,29 +455,31 @@ def test_get_operations_list(self, mock_listdir, mock_read_job_file): {"operation_type": "SOME_OPERATION_TYPE_2"}, ] - self.assertEqual( - self.server.get_operations_list(), - [ - { - "id": "SOME_OPERATION_1", - "type": "SOME_OPERATION_TYPE_1", - }, - { - "id": "SOME_OPERATION_2", - "type": "SOME_OPERATION_TYPE_2", - }, - ], - ) + expected = [ + { + "id": "SOME_OPERATION_1", + "type": "SOME_OPERATION_TYPE_1", + }, + { + "id": "SOME_OPERATION_2", + "type": "SOME_OPERATION_TYPE_2", + }, + ] + assert op_server.get_operations_list() == expected mock_read_job_file.assert_has_calls([ call("SOME_OPERATION_1"), call("SOME_OPERATION_2"), ]) - # Ensure expected operations are returned if filtering. - mock_listdir.reset_mock() - mock_read_job_file.reset_mock() + @patch("pg_backup_api.server_operation.OperationServer.read_job_file") + @patch("os.listdir") + def test_get_operations_list_with_filters(self, mock_listdir, + mock_read_job_file, op_server): + """Test :meth:`OperationServer.get_operations_list`. + Ensure expected operations are returned if filters are applied. + """ mock_listdir.return_value = [ "SOME_OPERATION_1.json", "SOME_OPERATION_2.json", @@ -401,15 +490,14 @@ def test_get_operations_list(self, mock_listdir, mock_read_job_file): {"operation_type": "SOME_OPERATION_TYPE_2"}, ] - self.assertEqual( - self.server.get_operations_list(OperationType.RECOVERY), - [ - { - "id": "SOME_OPERATION_1", - "type": "recovery", - }, - ], - ) + expected = [ + { + "id": "SOME_OPERATION_1", + "type": "recovery", + }, + ] + result = op_server.get_operations_list(OperationType.RECOVERY) + assert result == expected mock_read_job_file.assert_has_calls([ call("SOME_OPERATION_1"), @@ -418,240 +506,311 @@ def test_get_operations_list(self, mock_listdir, mock_read_job_file): @patch("pg_backup_api.server_operation.OperationServer.read_output_file") @patch("pg_backup_api.server_operation.OperationServer.read_job_file") - def test_get_operation_status(self, mock_read_job_file, - mock_read_output_file): + def test_get_operation_status_done(self, mock_read_job_file, + mock_read_output_file, op_server): + """Test :meth:`OperationServer.get_operation_status`. + + Ensure it returns ``DONE`` if the output file is successful. + """ id = "SOME_OP_ID" - # Ensure returns DONE if output file is successful. mock_read_output_file.return_value = {"success": True} - self.assertEqual( - self.server.get_operation_status(id), - "DONE", - ) + assert op_server.get_operation_status(id) == "DONE" mock_read_job_file.assert_not_called() mock_read_output_file.assert_called_once_with(id) - # Ensure returns FAILED if output file is not successful. - mock_read_job_file.reset_mock() - mock_read_output_file.reset_mock() + @patch("pg_backup_api.server_operation.OperationServer.read_output_file") + @patch("pg_backup_api.server_operation.OperationServer.read_job_file") + def test_get_operation_status_failed(self, mock_read_job_file, + mock_read_output_file, op_server): + """Test :meth:`OperationServer.get_operation_status`. + + Ensure it returns ``FAILED`` if the output file is not successful. + """ + id = "SOME_OP_ID" mock_read_output_file.return_value = {"success": False} - self.assertEqual( - self.server.get_operation_status(id), - "FAILED", - ) + assert op_server.get_operation_status(id) == "FAILED" mock_read_job_file.assert_not_called() mock_read_output_file.assert_called_once_with(id) - # Ensure returns IN_PROGRESS if job file exists. - mock_read_job_file.reset_mock() - mock_read_output_file.reset_mock() + @patch("pg_backup_api.server_operation.OperationServer.read_output_file") + @patch("pg_backup_api.server_operation.OperationServer.read_job_file") + def test_get_operation_status_in_progress(self, mock_read_job_file, + mock_read_output_file, + op_server): + """Test :meth:`OperationServer.get_operation_status`. + + Ensure it returns ``IN_PROGRESS`` if the job file exists, but no + output file exists. + """ + id = "SOME_OP_ID" mock_read_job_file.return_value = {} mock_read_output_file.side_effect = FileNotFoundError - self.assertEqual( - self.server.get_operation_status(id), - "IN_PROGRESS", - ) + assert op_server.get_operation_status(id) == "IN_PROGRESS" mock_read_job_file.assert_called_once_with(id) mock_read_output_file.assert_called_once_with(id) - # Ensure exception is raised if neither job nor output file exists. - mock_read_job_file.reset_mock() - mock_read_output_file.reset_mock() + @patch("pg_backup_api.server_operation.OperationServer.read_output_file") + @patch("pg_backup_api.server_operation.OperationServer.read_job_file") + def test_get_operation_status_exception(self, mock_read_job_file, + mock_read_output_file, + op_server): + """Test :meth:`OperationServer.get_operation_status`. + + Ensure an exception is raised if neither job nor output file exist. + """ + id = "SOME_OP_ID" mock_read_job_file.side_effect = FileNotFoundError mock_read_output_file.side_effect = FileNotFoundError - with self.assertRaises(OperationNotExists) as exc: - self.server.get_operation_status(id) - + with pytest.raises(OperationNotExists) as exc: + op_server.get_operation_status(id) - self.assertEqual( - str(exc.exception), - f"Operation '{id}' does not exist" - ) + assert str(exc.value) == f"Operation '{id}' does not exist" mock_read_job_file.assert_called_once_with(id) mock_read_output_file.assert_called_once_with(id) @patch("pg_backup_api.server_operation.OperationServer", MagicMock()) -class TestOperation(TestCase): +class TestOperation: + """Run tests for :class:`Operation`.""" + @pytest.fixture @patch("pg_backup_api.server_operation.OperationServer", MagicMock()) - def setUp(self): - self.operation = Operation(_BARMAN_SERVER) + def operation(self): + """Create an :class:`Operation` instance for testing. + + :return: a new instance of :class:`Operation` for testing. + """ + return Operation(_BARMAN_SERVER) + + def test___init___auto_id(self, operation): + """Test :meth:`Operation.__init__`. - def test___init__(self): - # Ensure ID is automatically generated, if no custom one is given. + Ensure the ID is automatically generated, if no custom one is given. + """ id = "AUTO_ID" - with patch.object(Operation, "_generate_id") as mock: - mock.return_value = id + with patch.object(Operation, "_generate_id") as mock_generate_id: + mock_generate_id.return_value = id operation = Operation(_BARMAN_SERVER) - self.assertEqual(operation.id, id) - mock.assert_called_once() + assert operation.id == id + mock_generate_id.assert_called_once() - # Ensure custom ID is considered, if a custom one is given. + def test___init___custom_id(self, operation): + """Test :meth:`Operation.__init__`. + + Ensure a custom ID is considered, if a custom one is given. + """ id = "CUSTOM_OP_ID" - with patch.object(Operation, "_generate_id") as mock: + with patch.object(Operation, "_generate_id") as mock_generate_id: operation = Operation(_BARMAN_SERVER, id) - self.assertEqual(operation.id, id) - mock.assert_not_called() - - def test__generate_id(self): - # Ensure generates ID based on current timestamp. - with patch("pg_backup_api.server_operation.datetime") as mock: - mock.now.return_value = MagicMock() - self.operation._generate_id() - mock.now.return_value.strftime.assert_called_once_with( + assert operation.id == id + mock_generate_id.assert_not_called() + + def test__generate_id(self, operation): + """Test :meth:`Operation._generate_id`. + + Ensure it generates an ID based on current timestamp. + """ + with patch("pg_backup_api.server_operation.datetime") as mock_datetime: + mock_datetime.now.return_value = MagicMock() + operation._generate_id() + mock_datetime.now.return_value.strftime.assert_called_once_with( "%Y%m%dT%H%M%S", ) - def test_time_even_now(self): - # Ensure timestamp is generated in the expected format. - with patch("pg_backup_api.server_operation.datetime") as mock: - mock.now.return_value = MagicMock() - self.operation.time_event_now() - mock.now.return_value.strftime.assert_called_once_with( + def test_time_even_now(self, operation): + """Test :meth:`Operation.time_event_now`. + + Ensure the timestamp is generated in the expected format. + """ + with patch("pg_backup_api.server_operation.datetime") as mock_datetime: + mock_datetime.now.return_value = MagicMock() + operation.time_event_now() + mock_datetime.now.return_value.strftime.assert_called_once_with( "%Y-%m-%dT%H:%M:%S.%f", ) - def test_job_file(self): - # Ensure get_job_file_path is called to satisfy job_file propety. - self.operation.job_file - self.operation.server.get_job_file_path.assert_called_once_with( - self.operation.id, + def test_job_file(self, operation): + """Test :meth:`Operation.job_file`. + + Ensure :meth:`OperationServer.get_job_file_path` is called to satisfy + the property. + """ + operation.job_file + operation.server.get_job_file_path.assert_called_once_with( + operation.id, ) - def test_output_file(self): - # Ensure get_output_file_path is called to satisfy output_file propety. - self.operation.output_file - self.operation.server.get_output_file_path.assert_called_once_with( - self.operation.id, + def test_output_file(self, operation): + """Test :meth:`Operation.output_file`. + + Ensure :meth:`OperationServer.get_output_file_path is called to satisfy + the property. + """ + operation.output_file + operation.server.get_output_file_path.assert_called_once_with( + operation.id, ) - def test_read_job_file(self): - # Ensure OperationServer.read_job_file is called as expected. - self.operation.read_job_file() - self.operation.server.read_job_file.assert_called_once_with( - self.operation.id, + def test_read_job_file(self, operation): + """Test :meth:`Operation.read_job_file`. + + Ensure :meth:`OperationServer.read_job_file` is called as expected. + """ + operation.read_job_file() + operation.server.read_job_file.assert_called_once_with( + operation.id, ) - - def test_read_output_file(self): - # Ensure OperationServer.read_output_file is called as expected. - self.operation.read_output_file() - self.operation.server.read_output_file.assert_called_once_with( - self.operation.id, + + def test_read_output_file(self, operation): + """Test :meth:`Operation.read_output_file`. + + Test :meth:`OperationServer.read_output_file` is called as expected. + """ + operation.read_output_file() + operation.server.read_output_file.assert_called_once_with( + operation.id, ) - - def test_write_job_file(self): - # Ensure OperationServer.write_job_file is called as expected. + + def test_write_job_file(self, operation): + """Test :meth:`Operation.write_jobf_file`. + + Ensure :meth:`OperationServer.write_job_file` is called as expected. + """ content = {"SOME": "CONTENT"} - self.operation.write_job_file(content) - self.operation.server.write_job_file.assert_called_once_with( - self.operation.id, + operation.write_job_file(content) + operation.server.write_job_file.assert_called_once_with( + operation.id, content, ) - - def test_write_output_file(self): - # Ensure OperationServer.write_output_file is called as expected. + + def test_write_output_file(self, operation): + """Test :meth:`Operation.write_output_file`. + + Ensure :meth:`OperationServer.write_output_file` is called as expected. + """ content = {"SOME": "CONTENT"} - self.operation.write_output_file(content) - self.operation.server.write_output_file.assert_called_once_with( - self.operation.id, + operation.write_output_file(content) + operation.server.write_output_file.assert_called_once_with( + operation.id, content, ) - - def test_get_status(self): - # Ensure OperationServer.get_operation_status is called as expected. - self.operation.get_status() - self.operation.server.get_operation_status.assert_called_once_with( - self.operation.id, + + def test_get_status(self, operation): + """Test :meth:`Operation.get_status`. + + Test :meth:`OperationServer.get_operation_status` is called as + expected. + """ + operation.get_status() + operation.server.get_operation_status.assert_called_once_with( + operation.id, ) - def test__run_subprocess(self): - # Ensure expected interactions with subprocess module. + def test__run_subprocess(self, operation): + """Test :meth:`Operation._run_subprocess`. + + Ensure it has the expected interactions with :mod:`subprocess`. + """ cmd = ["SOME", "COMMAND"] stdout = "SOME OUTPUT" return_code = 0 - with patch("subprocess.Popen", MagicMock()) as mock: - mock.return_value.communicate.return_value = (stdout, None) - mock.return_value.returncode = return_code + with patch("subprocess.Popen", MagicMock()) as mock_popen: + mock_popen.return_value.communicate.return_value = (stdout, None) + mock_popen.return_value.returncode = return_code - self.assertEqual( - self.operation._run_subprocess(cmd), - (stdout, return_code), - ) - - mock.assert_called_once_with(cmd, stdout=subprocess.PIPE, - stderr=subprocess.STDOUT) - - def test_run(self): - # Ensure _run_logic is called. - with patch.object(self.operation, "_run_logic") as mock: - self.operation.run() - mock.assert_called_once() - -@patch("pg_backup_api.server_operation.OperationServer", MagicMock()) -class TestRecoveryOperation(TestCase): - - @patch("pg_backup_api.server_operation.OperationServer", MagicMock()) - def setUp(self): - self.operation = RecoveryOperation(_BARMAN_SERVER) - - def test__validate_job_content(self): - content = {} - # Ensure exception is raised if content is missing keys -- test 1. - with self.assertRaises(MalformedContent) as exc: - self.operation._validate_job_content(content) - - self.assertEqual( - str(exc.exception), - "Missing required arguments: backup_id, destination_directory, " - "remote_ssh_command", - ) + assert operation._run_subprocess(cmd) == (stdout, return_code) - # Ensure exception is raised if content is missing keys -- test 2. - content["backup_id"] = "SOME_BACKUP_ID" + mock_popen.assert_called_once_with(cmd, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT) - with self.assertRaises(MalformedContent) as exc: - self.operation._validate_job_content(content) + def test_run(self, operation): + """Test :meth:`Operation.run`. - self.assertEqual( - str(exc.exception), - "Missing required arguments: destination_directory, " - "remote_ssh_command", - ) + Ensure :meth:`Operation._run_logic` is called. + """ + with patch.object(operation, "_run_logic") as mock_run_logic: + operation.run() + mock_run_logic.assert_called_once() - # Ensure exception is raised if content is missing keys -- test 3. - content["destination_directory"] = "SOME_DESTINATION_DIRECTORY" - with self.assertRaises(MalformedContent) as exc: - self.operation._validate_job_content(content) - - self.assertEqual( - str(exc.exception), - "Missing required arguments: remote_ssh_command", - ) +@patch("pg_backup_api.server_operation.OperationServer", MagicMock()) +class TestRecoveryOperation: + """Run tests for :class:`RecoveryOperation`.""" - # Ensure execution is fine if everything is filled. - content["remote_ssh_command"] = "SOME_REMOTE_SSH_COMMAND" - self.operation._validate_job_content(content) + @pytest.fixture + @patch("pg_backup_api.server_operation.OperationServer", MagicMock()) + def operation(self): + """Create a :class:`RecoveryOperation` instance for testing. + + :return: a new instance of :class:`RecoveryOperation` for testing. + """ + return RecoveryOperation(_BARMAN_SERVER) + + @pytest.mark.parametrize("content,missing_keys", [ + ({}, "backup_id, destination_directory, remote_ssh_command",), + ({"backup_id": "SOME_BACKUP_ID"}, + "destination_directory, remote_ssh_command"), + ({"destination_directory": "SOME_DESTINATION_DIRECTORY"}, + "backup_id, remote_ssh_command",), + ({"remote_ssh_command": "SOME_REMOTE_SSH_COMMAND"}, + "backup_id, destination_directory",), + ({"backup_id": "SOME_BACKUP_ID", + "destination_directory": "SOME_DESTINATION_DIRECTORY"}, + "remote_ssh_command"), + ({"backup_id": "SOME_BACKUP_ID", + "remote_ssh_command": "SOME_REMOTE_SSH_COMMAND"}, + "destination_directory"), + ({"destination_directory": "SOME_DESTINATION_DIRECTORY", + "remote_ssh_command": "SOME_REMOTE_SSH_COMMAND"}, + "backup_id"), + ]) + def test__validate_job_content_content_missing_keys(self, content, + missing_keys, + operation): + """Test :meth:`RecoveryOperation._validate_job_content`. + + Ensure and exception is raised if the content is missing keys. + """ + with pytest.raises(MalformedContent) as exc: + operation._validate_job_content(content) + + assert str(exc.value) == f"Missing required arguments: {missing_keys}" + + def test__validate_job_content_ok(self, operation): + """Test :meth:`RecoveryOperation._validate_job_content`. + + Ensure execution is fine if everything is filled as expected. + """ + content = { + "backup_id": "SOME_BACKUP_ID", + "destination_directory": "SOME_DESTINATION_DIRECTORY", + "remote_ssh_command": "SOME_REMOTE_SSH_COMMAND", + } + operation._validate_job_content(content) @patch("pg_backup_api.server_operation.Operation.time_event_now") @patch("pg_backup_api.server_operation.Operation.write_job_file") - def test_write_job_file(self, mock_write_job_file, mock_time_event_now): - # Ensure underlying methods are called as expected. + def test_write_job_file(self, mock_write_job_file, mock_time_event_now, + operation): + """Test :meth:`RecoveryOperation.write_job_file`. + + Ensure the underlying methods are called as expected. + """ content = { "SOME": "CONTENT", } @@ -661,48 +820,50 @@ def test_write_job_file(self, mock_write_job_file, mock_time_event_now): "start_time": "SOME_TIMESTAMP", } - with patch.object(self.operation, "_validate_job_content") as mock: + with patch.object(operation, "_validate_job_content") as mock: mock_time_event_now.return_value = "SOME_TIMESTAMP" - self.operation.write_job_file(content) + operation.write_job_file(content) mock_time_event_now.assert_called_once() mock.assert_called_once_with(extended_content) mock_write_job_file.assert_called_once_with(extended_content) - def test__get_args(self): - # Ensure it returns the correct arguments for 'barman recover'. - with patch.object(self.operation, "read_job_file") as mock: + def test__get_args(self, operation): + """Test :meth:`RecoveryOperation._get_args`. + + Ensure it returns the correct arguments for ``barman recover``. + """ + with patch.object(operation, "read_job_file") as mock: mock.return_value = { "backup_id": "SOME_BACKUP_ID", "destination_directory": "SOME_DESTINATION_DIRECTORY", "remote_ssh_command": "SOME_REMOTE_SSH_COMMAND", } - self.assertEqual( - self.operation._get_args(), - [ - self.operation.server.name, - "SOME_BACKUP_ID", - "SOME_DESTINATION_DIRECTORY", - "--remote-ssh-command", - "SOME_REMOTE_SSH_COMMAND", - ] - ) + expected = [ + operation.server.name, + "SOME_BACKUP_ID", + "SOME_DESTINATION_DIRECTORY", + "--remote-ssh-command", + "SOME_REMOTE_SSH_COMMAND", + ] + assert operation._get_args() == expected @patch("pg_backup_api.server_operation.Operation._run_subprocess") @patch("pg_backup_api.server_operation.RecoveryOperation._get_args") - def test__run_logic(self, mock_get_args, mock_run_subprocess): + def test__run_logic(self, mock_get_args, mock_run_subprocess, operation): + """Test :meth:`RecoveryOperation._run_logic`. + + Ensure the underlying calls occur as expected. + """ arguments = ["SOME", "ARGUMENTS"] output = ("SOME OUTPUT", 0) mock_get_args.return_value = arguments mock_run_subprocess.return_value = output - self.assertEqual( - self.operation._run_logic(), - output, - ) + assert operation._run_logic() == output mock_get_args.assert_called_once() mock_run_subprocess.assert_called_once_with( diff --git a/pg_backup_api/pg_backup_api/tests/test_utility_controller.py b/pg_backup_api/pg_backup_api/tests/test_utility_controller.py new file mode 100644 index 0000000..1aeee3d --- /dev/null +++ b/pg_backup_api/pg_backup_api/tests/test_utility_controller.py @@ -0,0 +1,551 @@ +# -*- coding: utf-8 -*- +# © Copyright EnterpriseDB UK Limited 2021-2023 +# +# This file is part of Postgres Backup API. +# +# Postgres Backup API is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Postgres Backup API is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Postgres Backup API. If not, see . + +"""Unit tests for the REST API endpoints.""" +import json +from unittest.mock import Mock, MagicMock, patch + +import pytest + +from pg_backup_api.server_operation import (OperationServerConfigError, + OperationNotExists, + MalformedContent) + + +_HTTP_METHODS = {"DELETE", "GET", "PATCH", "POST", "PUT", "TRACE"} + + +@patch("pg_backup_api.server_operation.load_barman_config", MagicMock()) +@patch("pg_backup_api.logic.utility_controller.load_barman_config", + MagicMock()) +@patch("barman.__config__", MagicMock()) +class TestUtilityController: + """Run tests for the REST API endpoints.""" + + @pytest.fixture(scope="module") + def client(self): + """Mock :mod:`barman.output` and get a Flask testing client. + + :yield: a Flask testing client. + """ + with patch("pg_backup_api.run.load_barman_config", MagicMock()): + from pg_backup_api.run import app + app.config.update({ + "TESTING": True, + }) + + from barman import output + output.set_output_writer(output.AVAILABLE_WRITERS["json"]()) + + with app.test_client() as test_client: + with app.app_context(): + yield test_client + + def _ensure_http_methods_not_allowed(self, methods, path, client): + """Ensure none among *methods* are allowed when requesting *path*. + + :param methods: a set of methods to be tested. + :param path: the URL path to be requested. + :param client: a Flask testing client. + """ + for method in methods: + response = getattr(client, method.lower())(path) + assert response.status_code == 405 + expected = b"The method is not allowed for the requested URL." + assert expected in response.data + + @patch("pg_backup_api.logic.utility_controller.barman_diagnose", Mock()) + @patch.dict( + "pg_backup_api.logic.utility_controller.output._writer.json_output", + { + '_INFO': ['SOME', 'JSON', 'ENTRIES', '{"global":{"config":{}}}'], + }, + ) + def test_diagnose_ok(self, client): + """Test ``/diagnose`` endpoint. + + Ensure a ``GET`` request returns ``200`` and the expected JSON output. + """ + path = "/diagnose" + + response = client.get(path) + + assert response.status_code == 200 + assert response.data == b'{"global":{"config":{}}}\n' + + def test_diagnose_not_allowed(self, client): + """Test ``/diagnose`` endpoint. + + Ensure all other HTTP request methods return an error. + """ + path = "/diagnose" + self._ensure_http_methods_not_allowed(_HTTP_METHODS - {"GET"}, path, + client) + + def test_status_ok(self, client): + """Test ``/status`` endpoint. + + Ensure a ``GET`` request returns ``200`` and the expected output. + """ + path = "/status" + + response = client.get(path) + + assert response.status_code == 200 + assert response.data == b'"OK"' + + def test_status_not_allowed(self, client): + """Test ``/status`` endpoint. + + Ensure all other HTTP request methods return an error. + """ + path = "/status" + + self._ensure_http_methods_not_allowed(_HTTP_METHODS - {"GET"}, path, + client) + + @pytest.mark.parametrize("status", ["IN_PROGRESS", "DONE", "FAILED"]) + @patch("pg_backup_api.logic.utility_controller.OperationServer") + def test_servers_operation_id_get_ok(self, mock_op_server, status, client): + """Test ``/servers//operations/`` endpoint. + + Ensure a ``GET`` request returns ``200`` and the expected JSON output + according to the status of the operation. + """ + path = "/servers/SOME_SERVER_NAME/operations/SOME_OPERATION_ID" + + mock_op_server.return_value.config = object() + mock_get_status = mock_op_server.return_value.get_operation_status + + mock_get_status.return_value = status + + response = client.get(path) + + mock_op_server.assert_called_once_with("SOME_SERVER_NAME") + mock_get_status.assert_called_once_with("SOME_OPERATION_ID") + + assert response.status_code == 200 + expected = ( + '{"operation_id":"SOME_OPERATION_ID",' + f'"status":"{status}"}}\n' + ).encode() + assert response.data == expected + + @patch("pg_backup_api.logic.utility_controller.OperationServer") + def test_servers_operation_id_get_server_does_not_exist(self, + mock_op_server, + client): + """Test ``/servers//operations/`` endpoint. + + Ensure ``GET`` returns ``404`` if the Barman server doesn't exist. + """ + path = "/servers/SOME_SERVER_NAME/operations/SOME_OPERATION_ID" + + mock_op_server.side_effect = OperationServerConfigError("SOME_ISSUE") + + response = client.get(path) + assert response.status_code == 404 + assert response.data == b'{"error":"404 Not Found: SOME_ISSUE"}\n' + + @patch("pg_backup_api.logic.utility_controller.OperationServer") + def test_servers_operation_id_get_operation_does_not_exist(self, + mock_op_server, + client): + """Test ``/servers//operations/`` endpoint. + + Ensure ``GET`` returns ``404`` if the operation doesn't exist. + """ + path = "/servers/SOME_SERVER_NAME/operations/SOME_OPERATION_ID" + + mock_get_status = mock_op_server.return_value.get_operation_status + mock_op_server.return_value.config = object() + mock_get_status.side_effect = OperationNotExists("NOT_FOUND") + + response = client.get(path) + assert response.status_code == 404 + expected = b'{"error":"404 Not Found: Resource not found"}\n' + assert response.data == expected + + def test_servers_operation_id_get_not_allowed(self, client): + """Test ``/servers//operations/`` endpoint. + + Ensure all other HTTP request methods return an error. + """ + path = "/servers/SOME_SERVER_NAME/operations/SOME_OPERATION_ID" + self._ensure_http_methods_not_allowed(_HTTP_METHODS - {"GET"}, path, + client) + + @patch("pg_backup_api.logic.utility_controller.OperationServer") + def test_server_operation_get_ok(self, mock_op_server, client): + """Test ``/servers//operations`` endpoint. + + Ensure a ``GET`` request returns ``200`` and the expected JSON output. + """ + path = "/servers/SOME_SERVER_NAME/operations" + + mock_op_server.return_value.config = object() + mock_get_ops = mock_op_server.return_value.get_operations_list + mock_get_ops.return_value = [ + { + "id": "SOME_ID_1", + "type": "SOME_TYPE_1", + }, + { + "id": "SOME_ID_2", + "type": "SOME_TYPE_2", + }, + ] + + response = client.get(path) + + mock_op_server.assert_called_once_with("SOME_SERVER_NAME") + mock_get_ops.assert_called_once_with() + + assert response.status_code == 200 + data = json.dumps({"operations": mock_get_ops.return_value}) + data = data.replace(" ", "") + "\n" + expected = data.encode() + assert response.data == expected + + @patch("pg_backup_api.logic.utility_controller.OperationServer") + def test_server_operation_get_server_does_not_exist(self, mock_op_server, + client): + """Test ``/servers//operations`` endpoint. + + Ensure ``GET`` request returns ``404`` if Barman server doesn't exist. + """ + path = "/servers/SOME_SERVER_NAME/operations" + + mock_get_ops = mock_op_server.return_value.get_operations_list + mock_op_server.side_effect = OperationServerConfigError( + "SOME_ISSUE") + + response = client.get(path) + + mock_op_server.assert_called_once_with("SOME_SERVER_NAME") + mock_get_ops.assert_not_called() + + assert response.status_code == 404 + assert response.data == b'{"error":"404 Not Found: SOME_ISSUE"}\n' + + def test_server_operation_post_not_json(self, client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request won't work without data in JSON format. + """ + path = "/servers/SOME_SERVER_NAME/operations" + + response = client.post(path, data={}) + + assert response.status_code == 415 + assert b"Unsupported Media Type" in response.data + + @patch("pg_backup_api.logic.utility_controller.OperationServer", Mock()) + @patch("pg_backup_api.logic.utility_controller.get_server_by_name") + @patch("pg_backup_api.logic.utility_controller.OperationType") + @patch("pg_backup_api.logic.utility_controller.parse_backup_id") + @patch("pg_backup_api.logic.utility_controller.Server") + @patch("pg_backup_api.logic.utility_controller.RecoveryOperation") + @patch("subprocess.Popen") + def test_server_operation_post_empty_json(self, mock_popen, mock_rec_op, + mock_server, mock_parse_id, + mock_op_type, mock_get_server, + client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request returns ``400`` if JSON data is empty. + """ + path = "/servers/SOME_SERVER_NAME/operations" + + response = client.post(path, json={}) + + assert response.status_code == 400 + expected = ( + b"Minimum barman options not met for server " + b"'SOME_SERVER_NAME'" + ) + assert expected in response.data + + mock_get_server.assert_not_called() + mock_op_type.assert_not_called() + mock_parse_id.assert_not_called() + mock_server.assert_not_called() + mock_rec_op.assert_not_called() + mock_popen.assert_not_called() + + @patch("pg_backup_api.logic.utility_controller.OperationServer", Mock()) + @patch("pg_backup_api.logic.utility_controller.get_server_by_name") + @patch("pg_backup_api.logic.utility_controller.OperationType") + @patch("pg_backup_api.logic.utility_controller.parse_backup_id") + @patch("pg_backup_api.logic.utility_controller.Server") + @patch("pg_backup_api.logic.utility_controller.RecoveryOperation") + @patch("subprocess.Popen") + def test_server_operation_post_server_does_not_exist(self, mock_popen, + mock_rec_op, + mock_server, + mock_parse_id, + mock_op_type, + mock_get_server, + client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request returns ``404`` if Barman server doesn't exist. + """ + path = "/servers/SOME_SERVER_NAME/operations" + + mock_get_server.return_value = None + + json_data = { + "type": "recovery", + } + response = client.post(path, json=json_data) + + mock_get_server.assert_called_once_with("SOME_SERVER_NAME") + mock_op_type.assert_not_called() + mock_parse_id.assert_not_called() + mock_server.assert_not_called() + mock_rec_op.assert_not_called() + mock_popen.assert_not_called() + + assert response.status_code == 404 + expected = ( + b'{"error":"404 Not Found: Server ' + b'\'SOME_SERVER_NAME\' does not exist"}\n' + ) + + assert response.data == expected + + @patch("pg_backup_api.logic.utility_controller.OperationServer", Mock()) + @patch("pg_backup_api.logic.utility_controller.get_server_by_name") + @patch("pg_backup_api.logic.utility_controller.OperationType") + @patch("pg_backup_api.logic.utility_controller.parse_backup_id") + @patch("pg_backup_api.logic.utility_controller.Server") + @patch("pg_backup_api.logic.utility_controller.RecoveryOperation") + @patch("subprocess.Popen") + def test_server_operation_post_backup_id_missing(self, mock_popen, + mock_rec_op, mock_server, + mock_parse_id, + mock_op_type, + mock_get_server, client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request returns ``400`` if ``backup_id`` is missing. + """ + path = "/servers/SOME_SERVER_NAME/operations" + json_data = { + "type": "recovery", + } + + mock_op_type.return_value = mock_op_type.RECOVERY + + response = client.post(path, json=json_data) + + mock_get_server.assert_called_once_with("SOME_SERVER_NAME") + mock_op_type.assert_called_once_with("recovery") + mock_parse_id.assert_not_called() + mock_server.assert_not_called() + mock_rec_op.assert_not_called() + mock_popen.assert_not_called() + + assert response.status_code == 400 + assert b"Request body is missing ``backup_id``" in response.data + + @patch("pg_backup_api.logic.utility_controller.OperationServer", Mock()) + @patch("pg_backup_api.logic.utility_controller.get_server_by_name") + @patch("pg_backup_api.logic.utility_controller.OperationType") + @patch("pg_backup_api.logic.utility_controller.parse_backup_id") + @patch("pg_backup_api.logic.utility_controller.Server") + @patch("pg_backup_api.logic.utility_controller.RecoveryOperation") + @patch("subprocess.Popen") + def test_server_operation_post_backup_does_not_exist(self, mock_popen, + mock_rec_op, + mock_server, + mock_parse_id, + mock_op_type, + mock_get_server, + client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request returns ``404`` if backup does not exist. + """ + path = "/servers/SOME_SERVER_NAME/operations" + json_data = { + "type": "recovery", + "backup_id": "SOME_BACKUP_ID", + } + + mock_parse_id.return_value = None + mock_op_type.return_value = mock_op_type.RECOVERY + + response = client.post(path, json=json_data) + + mock_get_server.assert_called_once_with("SOME_SERVER_NAME") + mock_op_type.assert_called_once_with("recovery") + mock_server.assert_called_once_with(mock_get_server.return_value) + mock_parse_id.assert_called_once_with(mock_server.return_value, + "SOME_BACKUP_ID") + mock_rec_op.assert_not_called() + mock_popen.assert_not_called() + + assert response.status_code == 404 + expected = ( + b'{"error":"404 Not Found: Backup ' + b'\'SOME_BACKUP_ID\' does not exist"}\n' + ) + assert response.data == expected + + @patch("pg_backup_api.logic.utility_controller.OperationServer", Mock()) + @patch("pg_backup_api.logic.utility_controller.get_server_by_name") + @patch("pg_backup_api.logic.utility_controller.OperationType") + @patch("pg_backup_api.logic.utility_controller.parse_backup_id") + @patch("pg_backup_api.logic.utility_controller.Server") + @patch("pg_backup_api.logic.utility_controller.RecoveryOperation") + @patch("subprocess.Popen") + def test_server_operation_post_missing_options(self, mock_popen, + mock_rec_op, mock_server, + mock_parse_id, mock_op_type, + mock_get_server, client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request returns ``400`` if any option is missing. + """ + path = "/servers/SOME_SERVER_NAME/operations" + json_data = { + "type": "recovery", + "backup_id": "SOME_BACKUP_ID", + } + + mock_op_type.return_value = mock_op_type.RECOVERY + mock_parse_id.return_value = "SOME_BACKUP_ID" + mock_rec_op.return_value.id = "SOME_OP_ID" + mock_write_job = mock_rec_op.return_value.write_job_file + mock_write_job.side_effect = MalformedContent("SOME_ERROR") + + response = client.post(path, json=json_data) + + mock_get_server.assert_called_once_with("SOME_SERVER_NAME") + mock_op_type.assert_called_once_with("recovery") + mock_server.assert_called_once_with(mock_get_server.return_value) + mock_parse_id.assert_called_once_with(mock_server.return_value, + "SOME_BACKUP_ID") + mock_rec_op.assert_called_once_with("SOME_SERVER_NAME") + mock_write_job.assert_called_once_with(json_data) + mock_popen.assert_not_called() + + assert response.status_code == 400 + expected = b"Make sure all options/arguments are met and try again" + assert expected in response.data + + @patch("pg_backup_api.logic.utility_controller.OperationServer", Mock()) + @patch("pg_backup_api.logic.utility_controller.get_server_by_name") + @patch("pg_backup_api.logic.utility_controller.OperationType") + @patch("pg_backup_api.logic.utility_controller.parse_backup_id") + @patch("pg_backup_api.logic.utility_controller.Server") + @patch("pg_backup_api.logic.utility_controller.RecoveryOperation") + @patch("subprocess.Popen") + def test_server_operation_post_ok(self, mock_popen, mock_rec_op, + mock_server, mock_parse_id, mock_op_type, + mock_get_server, client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request returns ``202`` if everything is ok, and ensure + the subprocess is started. + """ + path = "/servers/SOME_SERVER_NAME/operations" + json_data = { + "type": "recovery", + "backup_id": "SOME_BACKUP_ID", + } + + mock_op_type.return_value = mock_op_type.RECOVERY + mock_parse_id.return_value = "SOME_BACKUP_ID" + mock_rec_op.return_value.id = "SOME_OP_ID" + + response = client.post(path, json=json_data) + + mock_write_job = mock_rec_op.return_value.write_job_file + mock_get_server.assert_called_once_with("SOME_SERVER_NAME") + mock_op_type.assert_called_once_with("recovery") + mock_server.assert_called_once_with(mock_get_server.return_value) + mock_parse_id.assert_called_once_with(mock_server.return_value, + "SOME_BACKUP_ID") + mock_rec_op.assert_called_once_with("SOME_SERVER_NAME") + mock_write_job.assert_called_once_with(json_data) + mock_popen.assert_called_once_with(["pg-backup-api", "recovery", + "--server-name", + "SOME_SERVER_NAME", + "--operation-id", + "SOME_OP_ID"]) + + assert response.status_code == 202 + assert response.data == b'{"operation_id":"SOME_OP_ID"}\n' + + @patch("pg_backup_api.logic.utility_controller.OperationServer", Mock()) + @patch("pg_backup_api.logic.utility_controller.get_server_by_name") + @patch("pg_backup_api.logic.utility_controller.OperationType") + @patch("pg_backup_api.logic.utility_controller.parse_backup_id") + @patch("pg_backup_api.logic.utility_controller.Server") + @patch("pg_backup_api.logic.utility_controller.RecoveryOperation") + @patch("subprocess.Popen") + def test_server_operation_post_ok_type_missing(self, mock_popen, + mock_rec_op, mock_server, + mock_parse_id, mock_op_type, + mock_get_server, client): + """Test ``/servers//operations`` endpoint. + + Ensure ``POST`` request returns ``202`` if everything is ok, and ensure + the subprocess is started, even if ``type`` is absent, in which + case it defaults to ``recovery``. + """ + path = "/servers/SOME_SERVER_NAME/operations" + json_data = { + "backup_id": "SOME_BACKUP_ID", + } + + mock_op_type.return_value = mock_op_type.RECOVERY + mock_parse_id.return_value = "SOME_BACKUP_ID" + mock_rec_op.return_value.id = "SOME_OP_ID" + + response = client.post(path, json=json_data) + + mock_write_job = mock_rec_op.return_value.write_job_file + mock_get_server.assert_called_once_with("SOME_SERVER_NAME") + mock_op_type.assert_called_once_with("recovery") + mock_server.assert_called_once_with(mock_get_server.return_value) + mock_parse_id.assert_called_once_with(mock_server.return_value, + "SOME_BACKUP_ID") + mock_rec_op.assert_called_once_with("SOME_SERVER_NAME") + mock_write_job.assert_called_once_with(json_data) + mock_popen.assert_called_once_with(["pg-backup-api", "recovery", + "--server-name", + "SOME_SERVER_NAME", + "--operation-id", + "SOME_OP_ID"]) + + assert response.status_code == 202 + assert response.data == b'{"operation_id":"SOME_OP_ID"}\n' + + def test_server_operation_not_allowed(self, client): + """Test ``/servers//operations`` endpoint. + + Ensure all other HTTP request methods return an error. + """ + path = "/servers/SOME_SERVER_NAME/operations" + + self._ensure_http_methods_not_allowed(_HTTP_METHODS - {"GET", "POST"}, + path, client) diff --git a/pg_backup_api/pg_backup_api/tests/test_utils.py b/pg_backup_api/pg_backup_api/tests/test_utils.py new file mode 100644 index 0000000..9746edb --- /dev/null +++ b/pg_backup_api/pg_backup_api/tests/test_utils.py @@ -0,0 +1,195 @@ +# -*- coding: utf-8 -*- +# © Copyright EnterpriseDB UK Limited 2021-2023 +# +# This file is part of Postgres Backup API. +# +# Postgres Backup API is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Postgres Backup API is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Postgres Backup API. If not, see . + +"""Unit tests for utilitary functions.""" +from unittest.mock import MagicMock, patch, call + +from barman.infofile import BackupInfo +import pytest + +from pg_backup_api.utils import (create_app, load_barman_config, + setup_logging_for_wsgi_server, + get_server_by_name, parse_backup_id) + + +@patch("pg_backup_api.utils.Flask") +def test_create_app(mock_flask): + """Test :func:`create_app`. + + Ensure the :class:`Flask` object is created as expected. + """ + assert create_app() == mock_flask.return_value + + mock_flask.assert_called_once_with("Postgres Backup API") + + +@patch("pg_backup_api.utils.config.Config") +@patch("barman.__config__") +def test_load_barman_config(mock_global_config, mock_config): + """Test :func:`load_barman_config`. + + Ensure Barman configuration is loaded as expected. + """ + assert load_barman_config() is None + + mock_config.assert_called_once_with("/etc/barman.conf") + mock_global_config == mock_config.return_value + mock_load = mock_config.return_value.load_configuration_files_directory + mock_load.assert_called_once_with() + + +@patch("pg_backup_api.utils.dictConfig") +def test_setup_logging_for_wsgi_server(mock_dict_config): + """Test :func:`setup_logging_for_wsgi_server`. + + Ensure :meth:`logging.config.dictConfig` is called as expected. + """ + assert setup_logging_for_wsgi_server() is None + + log_format = "[%(asctime)s] %(levelname)s:%(module)s: %(message)s" + expected = { + "version": 1, + "formatters": { + "default": { + "format": log_format, + } + }, + "handlers": { + "wsgi": { + "class": "logging.FileHandler", + "filename": "/var/log/barman/barman-api.log", + "formatter": "default", + } + }, + "root": {"level": "INFO", "handlers": ["wsgi"]}, + "disable_existing_loggers": False, + } + mock_dict_config.assert_called_once_with(expected) + + +@patch("barman.__config__") +def test_get_server_by_name_not_found(mock_config): + """Test :func:`get_server_by_name`. + + Ensure ``None`` is returned if the server could not be found. + """ + mock_server_names = mock_config.server_names + mock_server_names.return_value = ["SERVER_1", "SERVER_2", "SERVER_3"] + mock_get_server = mock_config.get_server + + assert get_server_by_name("SERVER_4") is None + + mock_server_names.assert_called_once_with() + mock_get_server.assert_has_calls([ + call("SERVER_1"), + call("SERVER_2"), + call("SERVER_3"), + ]) + + +@patch("barman.__config__") +def test_get_server_by_name_ok(mock_config): + """Test :func:`get_server_by_name`. + + Ensure a server is returned if the server could be found. + """ + mock_server_names = mock_config.server_names + mock_server_names.return_value = ["SERVER_1", "SERVER_2", "SERVER_3"] + mock_get_server = mock_config.get_server + + assert get_server_by_name("SERVER_2") == mock_get_server.return_value + + mock_server_names.assert_called_once_with() + mock_get_server.assert_has_calls([ + call("SERVER_1"), + call("SERVER_2"), + ]) + + +@pytest.mark.parametrize("backup_id", ["latest", "last"]) +def test_parse_backup_id_latest(backup_id): + """Test :func:`parse_backup_id`. + + Ensure :meth:`barman.server.Server.get_last_backup_id()` is called when + backup ID is either ``latest`` or ``last``, then the corresponding backup + is returned. + """ + mock_server = MagicMock() + + expected = mock_server.get_backup.return_value + assert parse_backup_id(mock_server, backup_id) == expected + + mock_server.get_last_backup_id.assert_called_once_with() + mock_server.get_first_backup_id.assert_not_called() + expected = mock_server.get_last_backup_id.return_value + mock_server.get_backup.assert_called_once_with(expected) + + +@pytest.mark.parametrize("backup_id", ["oldest", "first"]) +def test_parse_backup_id_first(backup_id): + """Test :func:`parse_backup_id`. + + Ensure :meth:`barman.server.Server.get_first_backup_id()` is called when + backup ID is either ``oldest`` or ``first``, then the corresponding backup + is returned. + """ + mock_server = MagicMock() + + expected = mock_server.get_backup.return_value + assert parse_backup_id(mock_server, backup_id) == expected + + mock_server.get_last_backup_id.assert_not_called() + mock_server.get_first_backup_id.assert_called_once_with() + expected = mock_server.get_first_backup_id.return_value + mock_server.get_backup.assert_called_once_with(expected) + + +def test_parse_backup_id_last_failed(): + """Test :func:`parse_backup_id`. + + Ensure :meth:`barman.server.Server.get_last_backup_id()` is called when + backup ID is ``last-failed``, then the corresponding backup is returned. + """ + backup_id = "last-failed" + + mock_server = MagicMock() + + expected = mock_server.get_backup.return_value + assert parse_backup_id(mock_server, backup_id) == expected + + mock_server.get_last_backup_id.assert_called_once_with([BackupInfo.FAILED]) + mock_server.get_first_backup_id.assert_not_called() + expected = mock_server.get_last_backup_id.return_value + mock_server.get_backup.assert_called_once_with(expected) + + +def test_parse_backup_id_random(): + """Test :func:`parse_backup_id`. + + Ensure only :meth:`barman.server.Server.get_backup()` is called. + """ + backup_id = "random" + + mock_server = MagicMock() + + expected = mock_server.get_backup.return_value + assert parse_backup_id(mock_server, backup_id) == expected + + mock_server.get_last_backup_id.assert_not_called() + mock_server.get_first_backup_id.assert_not_called() + mock_server.get_backup.assert_called_once_with(backup_id)