From 7c832d41c18b5bece6bbb40da284b50703e5dbb5 Mon Sep 17 00:00:00 2001 From: Martin Vrachev Date: Mon, 17 Jun 2024 13:28:46 +0300 Subject: [PATCH] Tests: api-server in test_context and not as arg Signed-off-by: Martin Vrachev --- tests/conftest.py | 18 ++--- tests/unit/cli/admin/test_ceremony.py | 6 +- tests/unit/cli/admin/test_import_artifacts.py | 49 +++++++++----- tests/unit/cli/admin/test_sign.py | 67 +++++++++---------- 4 files changed, 71 insertions(+), 69 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index dfaa8e20..76c7261e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -41,18 +41,15 @@ _PROMPT = "rich.console.Console.input" -def _create_test_context(api_url: Optional[str]) -> Dict[str, Any]: +def _create_test_context() -> Dict[str, Any]: setting_file = os.path.join(TemporaryDirectory().name, "test_settings.yml") test_settings = Dynaconf(settings_files=[setting_file]) - if api_url: - test_settings.SERVER = api_url - return {"settings": test_settings, "config": setting_file} @pytest.fixture def test_context() -> Dict[str, Any]: - return _create_test_context(None) + return _create_test_context() def _create_client() -> CliRunner: @@ -302,8 +299,8 @@ def invoke_command( cmd: Command, inputs: List[str], args: List[str], - std_err_empty: bool = True, test_context: Optional[Context] = None, + std_err_empty: bool = True, ) -> Result: client = _create_client() out_file_name = "out_file.json" @@ -315,15 +312,10 @@ def invoke_command( else: out_args = ["--out", out_file_name] - api_url = None - if "--api-server" in args: - api_server_flag_index = args.index("--api-server") - api_url = args.pop(api_server_flag_index + 1) - args.remove("--api-server") + if not test_context: + test_context = _create_test_context() context = test_context - if test_context is None: - context = _create_test_context(api_url) with client.isolated_filesystem(): result_obj = client.invoke( diff --git a/tests/unit/cli/admin/test_ceremony.py b/tests/unit/cli/admin/test_ceremony.py index 3e4e4295..99e05d58 100644 --- a/tests/unit/cli/admin/test_ceremony.py +++ b/tests/unit/cli/admin/test_ceremony.py @@ -112,6 +112,7 @@ def test_ceremony_api_server( monkeypatch, patch_getpass, patch_utcnow, + test_context, ): fake_task_id = "123ab" fake_send_payload = pretend.call_recorder(lambda **kw: fake_task_id) @@ -119,12 +120,13 @@ def test_ceremony_api_server( fake_task_status = pretend.call_recorder(lambda *a: None) monkeypatch.setattr(ceremony, "task_status", fake_task_status) input_step1, input_step2, input_step3, input_step4 = ceremony_inputs - args = ["--api-server", "http://localhost:80"] + test_context["settings"].SERVER = "http://localhost:80" result = invoke_command( ceremony.ceremony, input_step1 + input_step2 + input_step3 + input_step4, - args, + [], + test_context, ) with open(_PAYLOADS / "ceremony.json") as f: diff --git a/tests/unit/cli/admin/test_import_artifacts.py b/tests/unit/cli/admin/test_import_artifacts.py index 375fd529..35c2e652 100644 --- a/tests/unit/cli/admin/test_import_artifacts.py +++ b/tests/unit/cli/admin/test_import_artifacts.py @@ -257,7 +257,7 @@ def test__get_succinct_roles_failed_parsing(self, monkeypatch): class TestImportArtifactsGroupCLI: - def test_import_artifacts(self): + def test_import_artifacts(self, test_context): # Required to properly mock functions imported inside import_artifacts import sqlalchemy @@ -293,8 +293,6 @@ def test_import_artifacts(self): ) args = [ - "--api-server", - "http://127.0.0.1", "--db-uri", "postgresql://postgres:secret@127.0.0.1:5433", "--csv", @@ -302,7 +300,11 @@ def test_import_artifacts(self): "--csv", "artifacts2of2.csv", ] - result = invoke_command(import_artifacts.import_artifacts, [], args) + test_context["settings"].SERVER = "http://127.0.0.1" + + result = invoke_command( + import_artifacts.import_artifacts, [], args, test_context + ) assert result.exit_code == 0, result.output assert "Finished." in result.output assert import_artifacts.bootstrap_status.calls == [ @@ -343,7 +345,7 @@ def test_import_artifacts_no_api_server_config_no_param(self): assert result.exit_code == 1, result.stderr assert "Requires '--api-server' " in result.stderr - def test_import_artifacts_skip_publish_artifacts(self): + def test_import_artifacts_skip_publish_artifacts(self, test_context): # Required to properly mock functions imported inside import_artifacts import sqlalchemy @@ -379,8 +381,6 @@ def test_import_artifacts_skip_publish_artifacts(self): ) args = [ - "--api-server", - "http://127.0.0.1", "--db-uri", "postgresql://postgres:secret@127.0.0.1:5433", "--csv", @@ -389,7 +389,11 @@ def test_import_artifacts_skip_publish_artifacts(self): "artifacts2of2.csv", "--skip-publish-artifacts", ] - result = invoke_command(import_artifacts.import_artifacts, [], args) + test_context["settings"].SERVER = "http://127.0.0.1" + + result = invoke_command( + import_artifacts.import_artifacts, [], args, test_context + ) assert result.exit_code == 0, result.output assert "Finished." in result.output assert "No artifacts published" in result.output @@ -408,7 +412,7 @@ def test_import_artifacts_skip_publish_artifacts(self): assert import_artifacts.publish_artifacts.calls == [] assert import_artifacts.task_status.calls == [] - def test_import_artifacts_sqlalchemy_import_fails(self): + def test_import_artifacts_sqlalchemy_import_fails(self, test_context): import builtins real_import = builtins.__import__ @@ -423,12 +427,14 @@ def fake_import(name, *args, **kwargs): builtins.__import__ = fake_import - args = ["--api-server", "", "--db-uri", "", "--csv", ""] + args = ["--db-uri", "", "--csv", ""] + test_context["settings"].SERVER = "" with pytest.raises(ModuleNotFoundError) as exc: invoke_command( import_artifacts.import_artifacts, [], args, + test_context, std_err_empty=False, ) @@ -436,15 +442,13 @@ def fake_import(name, *args, **kwargs): builtins.__import__ = real_import assert "pip install repository-service-tuf[sqlalchemy" in str(exc) - def test_import_artifacts_bootstrap_check_failed(self): + def test_import_artifacts_bootstrap_check_failed(self, test_context): import_artifacts.bootstrap_status = pretend.raiser( import_artifacts.click.ClickException("Server ERROR") ) args = [ - "--api-server", - "http://127.0.0.1", "--db-uri", "postgresql://postgres:secret@127.0.0.1:5433", "--csv", @@ -452,22 +456,25 @@ def test_import_artifacts_bootstrap_check_failed(self): "--csv", "artifacts2of2.csv", ] + test_context["settings"].SERVER = "http://127.0.0.1" result = invoke_command( - import_artifacts.import_artifacts, [], args, std_err_empty=False + import_artifacts.import_artifacts, + [], + args, + test_context, + std_err_empty=False, ) assert result.exit_code == 1 assert "Server ERROR" in result.stderr, result.stderr - def test_import_artifacts_without_bootstrap(self): + def test_import_artifacts_without_bootstrap(self, test_context): import_artifacts.bootstrap_status = pretend.call_recorder( lambda *a: {"data": {"bootstrap": False}, "message": "some msg"} ) args = [ - "--api-server", - "http://127.0.0.1", "--db-uri", "postgresql://postgres:secret@127.0.0.1:5433", "--csv", @@ -475,8 +482,14 @@ def test_import_artifacts_without_bootstrap(self): "--csv", "artifacts2of2.csv", ] + test_context["settings"].SERVER = "http://127.0.0.1" + result = invoke_command( - import_artifacts.import_artifacts, [], args, std_err_empty=False + import_artifacts.import_artifacts, + [], + args, + test_context, + std_err_empty=False, ) assert result.exit_code == 1, result.stderr assert ( diff --git a/tests/unit/cli/admin/test_sign.py b/tests/unit/cli/admin/test_sign.py index 5f1133ee..73d7414a 100644 --- a/tests/unit/cli/admin/test_sign.py +++ b/tests/unit/cli/admin/test_sign.py @@ -14,7 +14,7 @@ class TestSign: - def test_sign_with_previous_root(self, patch_getpass): + def test_sign_with_previous_root(self, test_context, patch_getpass): inputs = [ "1", # Please enter signing key index f"{_PEMS / 'JH.ed25519'}", # Please enter path to encrypted private key # noqa @@ -31,9 +31,10 @@ def test_sign_with_previous_root(self, patch_getpass): ) sign.send_payload = pretend.call_recorder(lambda *a: "fake-taskid") sign.task_status = pretend.call_recorder(lambda *a: "OK") - args = ["--api-server", "http://127.0.0.1"] + api_server = "http://127.0.0.1" + test_context["settings"].SERVER = api_server - result = invoke_command(sign.sign, inputs, args) + result = invoke_command(sign.sign, inputs, [], test_context) with open(_PAYLOADS / "sign.json") as f: expected = json.load(f) @@ -44,11 +45,7 @@ def test_sign_with_previous_root(self, patch_getpass): ) assert "Metadata Signed and sent to the API! 🔑" in result.stdout assert sign.request_server.calls == [ - pretend.call( - "http://127.0.0.1", - "api/v1/metadata/sign/", - Methods.GET, - ) + pretend.call(api_server, "api/v1/metadata/sign/", Methods.GET) ] assert sign.send_payload.calls == [ pretend.call( @@ -73,7 +70,7 @@ def test_sign_with_previous_root(self, patch_getpass): ) ] - def test_sign_bootstap_root(self, patch_getpass): + def test_sign_bootstap_root(self, test_context, patch_getpass): inputs = [ "1", # Please enter signing key index f"{_PEMS / 'JH.ed25519'}", # Please enter path to encrypted private key # noqa @@ -94,9 +91,10 @@ def test_sign_bootstap_root(self, patch_getpass): ) sign.send_payload = pretend.call_recorder(lambda *a: "fake-taskid") sign.task_status = pretend.call_recorder(lambda *a: "OK") - args = ["--api-server", "http://127.0.0.1"] + api_server = "http://127.0.0.1" + test_context["settings"].SERVER = api_server - result = invoke_command(sign.sign, inputs, args) + result = invoke_command(sign.sign, inputs, [], test_context) expected = { "keyid": "c6d8bf2e4f48b41ac2ce8eca21415ca8ef68c133b47fc33df03d4070a7e1e9cc", # noqa @@ -107,11 +105,7 @@ def test_sign_bootstap_root(self, patch_getpass): assert result.data["signature"]["keyid"] == expected["keyid"] assert "Metadata Signed and sent to the API! 🔑" in result.stdout assert sign.request_server.calls == [ - pretend.call( - "http://127.0.0.1", - "api/v1/metadata/sign/", - Methods.GET, - ) + pretend.call(api_server, "api/v1/metadata/sign/", Methods.GET) ] assert sign.send_payload.calls == [ pretend.call( @@ -167,7 +161,9 @@ def test_sign_input_option_and_custom_out( assert result.data["signature"]["sig"] == expected["sig"] assert f"Saved result to '{custom_path}'" in result.stdout - def test_sign_with_input_option_and_api_server_set(self, patch_getpass): + def test_sign_with_input_option_and_api_server_set( + self, test_context, patch_getpass + ): inputs = [ "1", # Please enter signing key index f"{_PEMS / 'JH.ed25519'}", # Please enter path to encrypted private key # noqa @@ -175,9 +171,10 @@ def test_sign_with_input_option_and_api_server_set(self, patch_getpass): sign.send_payload = pretend.call_recorder(lambda *a: "fake-taskid") sign.task_status = pretend.call_recorder(lambda *a: "OK") sign_input_path = f"{_PAYLOADS / 'sign_pending_roles.json'}" - api_server = "http://localhost:80" - args = ["--api-server", api_server, "--in", sign_input_path] - result = invoke_command(sign.sign, inputs, args) + test_context["settings"].SERVER = "http://localhost:80" + args = ["--in", sign_input_path] + + result = invoke_command(sign.sign, inputs, args, test_context) expected = { "keyid": "c6d8bf2e4f48b41ac2ce8eca21415ca8ef68c133b47fc33df03d4070a7e1e9cc", # noqa @@ -217,7 +214,9 @@ def test_sign_no_api_server_and_no_input_option(self): class TestSignError: - def test_sign_with_previous_root_but_wrong_version(self, patch_getpass): + def test_sign_with_previous_root_but_wrong_version( + self, test_context, patch_getpass + ): inputs = [ "1", # Please enter signing key index f"{_PEMS / 'JH.ed25519'}", # Please enter path to encrypted private key # noqa @@ -239,22 +238,20 @@ def test_sign_with_previous_root_but_wrong_version(self, patch_getpass): sign.request_server = pretend.call_recorder( lambda *a, **kw: fake_response ) - args = ["--api-server", "http://127.0.0.1"] + api_server = "http://127.0.0.1" + test_context["settings"].SERVER = api_server + test_result = invoke_command( - sign.sign, inputs, args, std_err_empty=False + sign.sign, inputs, [], test_context, std_err_empty=False ) assert test_result.exit_code == 1, test_result.stdout assert "Previous root v1 needed to sign root v2" in test_result.stderr assert sign.request_server.calls == [ - pretend.call( - "http://127.0.0.1", - "api/v1/metadata/sign/", - Methods.GET, - ) + pretend.call(api_server, "api/v1/metadata/sign/", Methods.GET) ] - def test_sign_fully_signed_metadata(self, patch_getpass): + def test_sign_fully_signed_metadata(self, test_context, patch_getpass): inputs = [ "1", # Please enter signing key index f"{_PEMS / 'JH.ed25519'}", # Please enter path to encrypted private key # noqa @@ -276,19 +273,17 @@ def test_sign_fully_signed_metadata(self, patch_getpass): sign.request_server = pretend.call_recorder( lambda *a, **kw: fake_response ) - args = ["--api-server", "http://127.0.0.1"] + api_server = "http://127.0.0.1" + test_context["settings"].SERVER = api_server + test_result = invoke_command( - sign.sign, inputs, args, std_err_empty=False + sign.sign, inputs, [], test_context, std_err_empty=False ) assert test_result.exit_code == 1, test_result.stdout assert "Metadata already fully signed." in test_result.stderr assert sign.request_server.calls == [ - pretend.call( - "http://127.0.0.1", - "api/v1/metadata/sign/", - Methods.GET, - ) + pretend.call(api_server, "api/v1/metadata/sign/", Methods.GET) ]