From 64ba8d1c05e6a77f6bd5e30f9d5728184aba272f Mon Sep 17 00:00:00 2001 From: Jacob Craiglow Date: Mon, 13 Jul 2026 16:37:13 -0400 Subject: [PATCH 1/5] Sonar cloud: Regex extracted to method with tests Signed-off-by: Jacob Craiglow --- src/aap_eda/services/project/scm.py | 31 +++++++++-- tests/unit/test_scm.py | 86 +++++++++++++++++++++++++++++ 2 files changed, 111 insertions(+), 6 deletions(-) diff --git a/src/aap_eda/services/project/scm.py b/src/aap_eda/services/project/scm.py index d4688ed2b..88d174e7c 100644 --- a/src/aap_eda/services/project/scm.py +++ b/src/aap_eda/services/project/scm.py @@ -380,12 +380,8 @@ def __call__( ) if match: return match.group(1) - match = re.search( - r'fatal: \[localhost\]: FAILED! => \{.+"msg": (.+)\}', - outputs.getvalue(), - ) - if match: - err_msg = match.group(1) + err_msg = self._extract_error_msg(outputs.getvalue()) + if err_msg: if "Authentication failed" in err_msg: raise ScmAuthenticationError("Authentication failed") if ( @@ -397,6 +393,29 @@ def __call__( raise ScmError(f"{self.ERROR_PREFIX} {err_msg}") raise ScmError(f"{self.ERROR_PREFIX} {outputs.getvalue().strip()}") + @staticmethod + def _extract_error_msg(output): + """Extract msg value from Ansible FAILED output. + + Looks for the fatal error marker, then finds the "msg" + key and extracts its value up to the last closing brace. + Handles msg values that contain } characters. + """ + fatal_marker = "fatal: [localhost]: FAILED! => {" + fatal_idx = output.find(fatal_marker) + if fatal_idx < 0: + return None + block = output[fatal_idx + len(fatal_marker) :] + msg_marker = '"msg": ' + msg_idx = block.find(msg_marker) + if msg_idx < 0: + return None + msg_value = block[msg_idx + len(msg_marker) :] + brace_idx = msg_value.rfind("}") + if brace_idx < 0: + return None + return msg_value[:brace_idx] + def is_refspec_valid(refspec: str, is_branch: bool) -> bool: if is_branch: diff --git a/tests/unit/test_scm.py b/tests/unit/test_scm.py index 7c09c9eec..829057d33 100644 --- a/tests/unit/test_scm.py +++ b/tests/unit/test_scm.py @@ -411,3 +411,89 @@ def test_git_clone_ssh_key(ssh_credential: models.EdaCredential, url: str): # SSH key file must be provided assert "key_file" in extra_vars assert extra_vars["key_file"] # non-empty path + + +################################################################# +# Tests for GitAnsibleRunnerExecutor._extract_error_msg +################################################################# + +_extract = scm.GitAnsibleRunnerExecutor._extract_error_msg + + +def test_extract_error_msg_standard(): + output = ( + "fatal: [localhost]: FAILED! => " + '{"changed": false, "msg": "Authentication failed"}' + ) + assert _extract(output) == '"Authentication failed"' + + +def test_extract_error_msg_multiple_keys(): + output = ( + "fatal: [localhost]: FAILED! => " + '{"changed": false, "rc": 1, ' + '"msg": "Could not clone repo"}' + ) + assert _extract(output) == '"Could not clone repo"' + + +def test_extract_error_msg_only_key(): + output = ( + "fatal: [localhost]: FAILED! => " '{"msg": "could not read Username"}' + ) + assert _extract(output) == '"could not read Username"' + + +def test_extract_error_msg_brace_in_value(): + output = ( + "fatal: [localhost]: FAILED! => " + '{"changed": false, ' + '"msg": "failed with {error: bad}"}' + ) + assert _extract(output) == '"failed with {error: bad}"' + + +def test_extract_error_msg_success_returns_none(): + output = 'ok: [localhost]: SUCCESS => {"changed": true}' + assert _extract(output) is None + + +def test_extract_error_msg_no_msg_key_returns_none(): + output = ( + "fatal: [localhost]: FAILED! => " + '{"changed": false, "error": "something"}' + ) + assert _extract(output) is None + + +def test_extract_error_msg_empty(): + output = "fatal: [localhost]: FAILED! => " '{"changed": false, "msg": ""}' + assert _extract(output) == '""' + + +def test_extract_error_msg_multiline(): + output = ( + "some log output\n" + "fatal: [localhost]: FAILED! => " + '{"msg": "real error"}' + ) + assert _extract(output) == '"real error"' + + +def test_extract_error_msg_auth_failed(): + output = ( + "fatal: [localhost]: FAILED! => " + '{"msg": "Authentication failed for user@host"}' + ) + assert _extract(output) == '"Authentication failed for user@host"' + + +def test_extract_error_msg_username_prompt(): + output = ( + "fatal: [localhost]: FAILED! => " + '{"msg": "could not read Username for ' + "'https://git.example.com'\"}" + ) + assert _extract(output) == ( + '"could not read Username for ' "'https://git.example.com'\"" + ) From ad6b6db9421a160478f012af904579345fd7981a Mon Sep 17 00:00:00 2001 From: Jacob Craiglow Date: Tue, 14 Jul 2026 13:30:02 -0400 Subject: [PATCH 2/5] Sonar cloud: Add tests for scm.py for compliance Signed-off-by: Jacob Craiglow --- tests/unit/test_scm.py | 142 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) diff --git a/tests/unit/test_scm.py b/tests/unit/test_scm.py index 829057d33..ee6309a26 100644 --- a/tests/unit/test_scm.py +++ b/tests/unit/test_scm.py @@ -156,6 +156,44 @@ def spy_get(obj): assert os.environ.get(key) is None +@pytest.mark.django_db +def test_set_proxy_environ_restores_existing_vars( + credential: models.EdaCredential, +): + """Pre-existing proxy env vars are restored after clone.""" + executor = mock.MagicMock() + original_value = "http://original-proxy:8080" + + for key in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"): + os.environ[key] = original_value + + try: + with tempfile.TemporaryDirectory() as dest_path: + scm.ScmRepository.clone( + "https://git.example.com/repo.git", + dest_path, + credential=credential, + proxy="http://override-proxy:3128", + _executor=executor, + ) + + for key in ( + "http_proxy", + "https_proxy", + "HTTP_PROXY", + "HTTPS_PROXY", + ): + assert os.environ.get(key) == original_value + finally: + for key in ( + "http_proxy", + "https_proxy", + "HTTP_PROXY", + "HTTPS_PROXY", + ): + os.environ.pop(key, None) + + @pytest.mark.django_db def test_git_clone_leak_password( credential: models.EdaCredential, @@ -497,3 +535,107 @@ def test_extract_error_msg_username_prompt(): assert _extract(output) == ( '"could not read Username for ' "'https://git.example.com'\"" ) + + +def test_extract_error_msg_no_closing_brace(): + output = 'fatal: [localhost]: FAILED! => {"msg": "truncated output' + assert _extract(output) is None + + +def test_extract_error_msg_fatal_marker_only(): + output = "fatal: [localhost]: FAILED! => {" + assert _extract(output) is None + + +@pytest.mark.django_db +def test_git_clone_gpg_credential( + default_organization: models.Organization, +): + """GPG credential sets up verify_commit and GNUPGHOME.""" + gpg_credential = models.EdaCredential.objects.create( + name="test-gpg-credential", + inputs={"gpg_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n"}, + organization=default_organization, + ) + gpg_credential.refresh_from_db() + executor = mock.MagicMock() + + with mock.patch.object(scm.ScmRepository, "add_gpg_key"): + with tempfile.TemporaryDirectory() as dest_path: + scm.ScmRepository.clone( + "https://git.example.com/repo.git", + dest_path, + gpg_credential=gpg_credential, + _executor=executor, + ) + call_kwargs = executor.call_args + assert call_kwargs[1]["extra_vars"]["verify_commit"] == "true" + assert "GNUPGHOME" in call_kwargs[1]["env_vars"] + + +@pytest.mark.django_db +def test_git_clone_creates_directory(): + """Clone creates the destination directory if it doesn't exist.""" + executor = mock.MagicMock() + + with tempfile.TemporaryDirectory() as parent: + dest_path = os.path.join(parent, "nonexistent") + scm.ScmRepository.clone( + "https://git.example.com/repo.git", + dest_path, + _executor=executor, + ) + assert os.path.isdir(dest_path) + + +@pytest.mark.django_db +def test_git_clone_decrypt_key_file( + default_organization: models.Organization, +): + """Clone calls decrypt_key_file when ssh_key_unlock is provided.""" + credential = models.EdaCredential.objects.create( + name="test-ssh-unlock", + inputs={ + "ssh_key_data": "-----BEGIN OPENSSH PRIVATE KEY-----\n", + "ssh_key_unlock": "passphrase", + }, + organization=default_organization, + ) + credential.refresh_from_db() + executor = mock.MagicMock() + + with mock.patch.object( + scm.ScmRepository, "decrypt_key_file" + ) as mock_decrypt: + with tempfile.TemporaryDirectory() as dest_path: + scm.ScmRepository.clone( + "git@git.example.com:repo.git", + dest_path, + credential=credential, + _executor=executor, + ) + mock_decrypt.assert_called_once() + + +@pytest.mark.django_db +def test_git_clone_gpg_add_key( + default_organization: models.Organization, +): + """Clone calls add_gpg_key when gpg_credential is provided.""" + gpg_credential = models.EdaCredential.objects.create( + name="test-gpg", + inputs={"gpg_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n"}, + organization=default_organization, + ) + gpg_credential.refresh_from_db() + executor = mock.MagicMock() + + with mock.patch.object(scm.ScmRepository, "add_gpg_key") as mock_add_gpg: + with tempfile.TemporaryDirectory() as dest_path: + scm.ScmRepository.clone( + "https://git.example.com/repo.git", + dest_path, + gpg_credential=gpg_credential, + _executor=executor, + ) + mock_add_gpg.assert_called_once() From d317ba1cfa8e042de37d2164dec67e3442bb4187 Mon Sep 17 00:00:00 2001 From: Jacob Craiglow Date: Tue, 14 Jul 2026 14:18:04 -0400 Subject: [PATCH 3/5] Sonar cloud: Add tests for scm.py for compliance Signed-off-by: Jacob Craiglow --- tests/unit/test_scm.py | 140 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 140 insertions(+) diff --git a/tests/unit/test_scm.py b/tests/unit/test_scm.py index ee6309a26..5eaff784b 100644 --- a/tests/unit/test_scm.py +++ b/tests/unit/test_scm.py @@ -639,3 +639,143 @@ def test_git_clone_gpg_add_key( _executor=executor, ) mock_add_gpg.assert_called_once() + + +################################################################# +# Tests for ScmRepository.decrypt_key_file and add_gpg_key +################################################################# + + +def test_decrypt_key_file_success(): + mock_result = mock.Mock() + mock_result.returncode = 0 + + with mock.patch( + "aap_eda.services.project.scm.subprocess.run", + return_value=mock_result, + ) as mock_run: + scm.ScmRepository.decrypt_key_file("/tmp/keyfile", "passphrase") + + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + assert "-P" in args + assert "passphrase" in args + + +def test_decrypt_key_file_failure(): + mock_result = mock.Mock() + mock_result.returncode = 1 + mock_result.stderr = "bad passphrase" + mock_result.stdout = "" + + with mock.patch( + "aap_eda.services.project.scm.subprocess.run", + return_value=mock_result, + ): + with pytest.raises(scm.ScmError) as exc_info: + scm.ScmRepository.decrypt_key_file("/tmp/keyfile", "wrong") + assert "Failed to decrypt" in str(exc_info.value) + + +def test_add_gpg_key_success(): + mock_result = mock.Mock() + mock_result.returncode = 0 + + with mock.patch( + "aap_eda.services.project.scm.subprocess.run", + return_value=mock_result, + ) as mock_run: + scm.ScmRepository.add_gpg_key("/tmp/gpgkey", "/tmp/gnupg") + + mock_run.assert_called_once() + call_kwargs = mock_run.call_args + assert call_kwargs[1]["env"] == {"GNUPGHOME": "/tmp/gnupg"} + + +def test_add_gpg_key_failure(): + mock_result = mock.Mock() + mock_result.returncode = 2 + mock_result.stderr = "no valid OpenPGP data" + mock_result.stdout = "" + + with mock.patch( + "aap_eda.services.project.scm.subprocess.run", + return_value=mock_result, + ): + with pytest.raises(scm.ScmError) as exc_info: + scm.ScmRepository.add_gpg_key("/tmp/gpgkey", "/tmp/gnupg") + assert "Failed to import" in str(exc_info.value) + + +################################################################# +# Tests for GitAnsibleRunnerExecutor.__call__ +################################################################# + + +def _run_executor(runner_rc, runner_stdout): + """Helper to run GitAnsibleRunnerExecutor with mocked runner.""" + executor = scm.GitAnsibleRunnerExecutor() + mock_runner = mock.Mock() + mock_runner.rc = runner_rc + + def fake_run(**kwargs): + # Write to the captured stdout so redirect_stdout picks it up + import sys + + sys.stdout.write(runner_stdout) + return mock_runner + + with mock.patch( + "aap_eda.services.project.scm.ansible_runner.run", + side_effect=fake_run, + ): + return executor( + extra_vars={"project_path": "/tmp"}, + env_vars={}, + ) + + +def test_executor_call_success(): + output = '"msg": "Repository Version abc123def456"' + result = _run_executor(0, output) + assert result == "abc123def456" + + +def test_executor_call_success_no_version(): + with pytest.raises(scm.ScmError) as exc_info: + _run_executor(0, "some output without version") + assert "Project Import Error:" in str(exc_info.value) + + +def test_executor_call_auth_failure(): + output = ( + "fatal: [localhost]: FAILED! => " + '{"changed": false, "msg": "Authentication failed"}' + ) + with pytest.raises(scm.ScmAuthenticationError): + _run_executor(1, output) + + +def test_executor_call_username_prompt(): + output = ( + "fatal: [localhost]: FAILED! => " '{"msg": "could not read Username"}' + ) + with pytest.raises(scm.ScmAuthenticationError) as exc_info: + _run_executor(1, output) + assert "Credentials not provided" in str(exc_info.value) + + +def test_executor_call_generic_error(): + output = ( + "fatal: [localhost]: FAILED! => " '{"msg": "repository not found"}' + ) + with pytest.raises(scm.ScmError) as exc_info: + _run_executor(1, output) + assert "Project Import Error:" in str(exc_info.value) + assert "repository not found" in str(exc_info.value) + + +def test_executor_call_no_fatal_output(): + with pytest.raises(scm.ScmError) as exc_info: + _run_executor(1, "some generic failure output") + assert "Project Import Error:" in str(exc_info.value) From cc5468fd17a3935dff82ceadb542eca91c287d1e Mon Sep 17 00:00:00 2001 From: Jacob Craiglow Date: Thu, 16 Jul 2026 15:59:28 -0400 Subject: [PATCH 4/5] sonar issues: Refactor from regex to utilize ansible runner API Signed-off-by: Jacob Craiglow --- src/aap_eda/services/project/scm.py | 69 ++++++------ tests/unit/test_scm.py | 164 ++++++++++++---------------- 2 files changed, 99 insertions(+), 134 deletions(-) diff --git a/src/aap_eda/services/project/scm.py b/src/aap_eda/services/project/scm.py index 88d174e7c..abaae8ea1 100644 --- a/src/aap_eda/services/project/scm.py +++ b/src/aap_eda/services/project/scm.py @@ -14,10 +14,8 @@ from __future__ import annotations import contextlib -import io import logging import os -import re import shutil import subprocess import tempfile @@ -365,22 +363,20 @@ def __call__( os.makedirs(env_dir) safe_yaml.dump(os.path.join(env_dir, "extravars"), extra_vars) - outputs = io.StringIO() - with contextlib.redirect_stdout(outputs): - runner = ansible_runner.run( - private_data_dir=data_dir, - playbook=PLAYBOOK, - envvars=env_vars, - ) + runner = ansible_runner.run( + private_data_dir=data_dir, + playbook=PLAYBOOK, + envvars=env_vars, + quiet=True, + ) + + events = list(runner.events) if runner.rc == 0: - match = re.search( - r'"msg": "Repository Version ([0-9a-fA-F]+)"', - outputs.getvalue(), - ) - if match: - return match.group(1) - err_msg = self._extract_error_msg(outputs.getvalue()) + git_hash = self._extract_git_hash(events) + if git_hash: + return git_hash + err_msg = self._extract_error_msg(events) if err_msg: if "Authentication failed" in err_msg: raise ScmAuthenticationError("Authentication failed") @@ -391,30 +387,29 @@ def __call__( err_msg = "Credentials not provided or incorrect" raise ScmAuthenticationError(err_msg) raise ScmError(f"{self.ERROR_PREFIX} {err_msg}") - raise ScmError(f"{self.ERROR_PREFIX} {outputs.getvalue().strip()}") + raise ScmError(self.ERROR_PREFIX) @staticmethod - def _extract_error_msg(output): - """Extract msg value from Ansible FAILED output. + def _extract_git_hash(events): + for event in events: + if event.get("event") != "runner_on_ok": + continue + res = event.get("event_data", {}).get("res", {}) + scm_version = res.get("ansible_facts", {}).get("scm_version") + if scm_version: + return scm_version + return None - Looks for the fatal error marker, then finds the "msg" - key and extracts its value up to the last closing brace. - Handles msg values that contain } characters. - """ - fatal_marker = "fatal: [localhost]: FAILED! => {" - fatal_idx = output.find(fatal_marker) - if fatal_idx < 0: - return None - block = output[fatal_idx + len(fatal_marker) :] - msg_marker = '"msg": ' - msg_idx = block.find(msg_marker) - if msg_idx < 0: - return None - msg_value = block[msg_idx + len(msg_marker) :] - brace_idx = msg_value.rfind("}") - if brace_idx < 0: - return None - return msg_value[:brace_idx] + @staticmethod + def _extract_error_msg(events): + for event in events: + if event.get("event") != "runner_on_failed": + continue + res = event.get("event_data", {}).get("res", {}) + msg = res.get("msg") + if msg: + return msg + return None def is_refspec_valid(refspec: str, is_branch: bool) -> bool: diff --git a/tests/unit/test_scm.py b/tests/unit/test_scm.py index 5eaff784b..d9dbaa95a 100644 --- a/tests/unit/test_scm.py +++ b/tests/unit/test_scm.py @@ -452,99 +452,78 @@ def test_git_clone_ssh_key(ssh_credential: models.EdaCredential, url: str): ################################################################# -# Tests for GitAnsibleRunnerExecutor._extract_error_msg +# Tests for GitAnsibleRunnerExecutor._extract_git_hash ################################################################# -_extract = scm.GitAnsibleRunnerExecutor._extract_error_msg - - -def test_extract_error_msg_standard(): - output = ( - "fatal: [localhost]: FAILED! => " - '{"changed": false, "msg": "Authentication failed"}' - ) - assert _extract(output) == '"Authentication failed"' +_extract_hash = scm.GitAnsibleRunnerExecutor._extract_git_hash +_extract_error = scm.GitAnsibleRunnerExecutor._extract_error_msg -def test_extract_error_msg_multiple_keys(): - output = ( - "fatal: [localhost]: FAILED! => " - '{"changed": false, "rc": 1, ' - '"msg": "Could not clone repo"}' - ) - assert _extract(output) == '"Could not clone repo"' +def _ok_event(res=None): + return {"event": "runner_on_ok", "event_data": {"res": res or {}}} -def test_extract_error_msg_only_key(): - output = ( - "fatal: [localhost]: FAILED! => " '{"msg": "could not read Username"}' - ) - assert _extract(output) == '"could not read Username"' +def _failed_event(res=None): + return { + "event": "runner_on_failed", + "event_data": {"res": res or {}}, + } -def test_extract_error_msg_brace_in_value(): - output = ( - "fatal: [localhost]: FAILED! => " - '{"changed": false, ' - '"msg": "failed with {error: bad}"}' - ) - assert _extract(output) == '"failed with {error: bad}"' +def test_extract_git_hash_from_set_fact(): + events = [ + _ok_event({"changed": True}), + _ok_event( + { + "ansible_facts": { + "scm_version": "abc123def456", + }, + } + ), + _ok_event({"msg": "Repository Version abc123def456"}), + ] + assert _extract_hash(events) == "abc123def456" -def test_extract_error_msg_success_returns_none(): - output = 'ok: [localhost]: SUCCESS => {"changed": true}' - assert _extract(output) is None +def test_extract_git_hash_no_matching_event(): + events = [ + _ok_event({"changed": True}), + {"event": "runner_on_start", "event_data": {}}, + ] + assert _extract_hash(events) is None -def test_extract_error_msg_no_msg_key_returns_none(): - output = ( - "fatal: [localhost]: FAILED! => " - '{"changed": false, "error": "something"}' - ) - assert _extract(output) is None +def test_extract_git_hash_empty_events(): + assert _extract_hash([]) is None -def test_extract_error_msg_empty(): - output = "fatal: [localhost]: FAILED! => " '{"changed": false, "msg": ""}' - assert _extract(output) == '""' +################################################################# +# Tests for GitAnsibleRunnerExecutor._extract_error_msg +################################################################# -def test_extract_error_msg_multiline(): - output = ( - "some log output\n" - "fatal: [localhost]: FAILED! => " - '{"msg": "real error"}' - ) - assert _extract(output) == '"real error"' +def test_extract_error_msg_from_failed_event(): + events = [_failed_event({"msg": "Authentication failed"})] + assert _extract_error(events) == "Authentication failed" -def test_extract_error_msg_auth_failed(): - output = ( - "fatal: [localhost]: FAILED! => " - '{"msg": "Authentication failed for user@host"}' - ) - assert _extract(output) == '"Authentication failed for user@host"' +def test_extract_error_msg_no_failure_event(): + events = [_ok_event({"changed": True})] + assert _extract_error(events) is None -def test_extract_error_msg_username_prompt(): - output = ( - "fatal: [localhost]: FAILED! => " - '{"msg": "could not read Username for ' - "'https://git.example.com'\"}" - ) - assert _extract(output) == ( - '"could not read Username for ' "'https://git.example.com'\"" - ) +def test_extract_error_msg_missing_msg_key(): + events = [_failed_event({"changed": False, "rc": 1})] + assert _extract_error(events) is None -def test_extract_error_msg_no_closing_brace(): - output = 'fatal: [localhost]: FAILED! => {"msg": "truncated output' - assert _extract(output) is None +def test_extract_error_msg_empty_msg(): + events = [_failed_event({"msg": ""})] + assert _extract_error(events) is None -def test_extract_error_msg_fatal_marker_only(): - output = "fatal: [localhost]: FAILED! => {" - assert _extract(output) is None +def test_extract_error_msg_empty_events(): + assert _extract_error([]) is None @pytest.mark.django_db @@ -712,70 +691,61 @@ def test_add_gpg_key_failure(): ################################################################# -def _run_executor(runner_rc, runner_stdout): +def _run_executor(runner_rc, events): """Helper to run GitAnsibleRunnerExecutor with mocked runner.""" executor = scm.GitAnsibleRunnerExecutor() mock_runner = mock.Mock() mock_runner.rc = runner_rc - - def fake_run(**kwargs): - # Write to the captured stdout so redirect_stdout picks it up - import sys - - sys.stdout.write(runner_stdout) - return mock_runner + mock_runner.events = events with mock.patch( "aap_eda.services.project.scm.ansible_runner.run", - side_effect=fake_run, + return_value=mock_runner, ): return executor( - extra_vars={"project_path": "/tmp"}, + extra_vars={"project_path": "/mock/path"}, env_vars={}, ) def test_executor_call_success(): - output = '"msg": "Repository Version abc123def456"' - result = _run_executor(0, output) + events = [ + _ok_event({"ansible_facts": {"scm_version": "abc123def456"}}), + ] + result = _run_executor(0, events) assert result == "abc123def456" def test_executor_call_success_no_version(): + events = [_ok_event({"changed": True})] with pytest.raises(scm.ScmError) as exc_info: - _run_executor(0, "some output without version") + _run_executor(0, events) assert "Project Import Error:" in str(exc_info.value) def test_executor_call_auth_failure(): - output = ( - "fatal: [localhost]: FAILED! => " - '{"changed": false, "msg": "Authentication failed"}' - ) + events = [_failed_event({"msg": "Authentication failed"})] with pytest.raises(scm.ScmAuthenticationError): - _run_executor(1, output) + _run_executor(1, events) def test_executor_call_username_prompt(): - output = ( - "fatal: [localhost]: FAILED! => " '{"msg": "could not read Username"}' - ) + events = [_failed_event({"msg": "could not read Username"})] with pytest.raises(scm.ScmAuthenticationError) as exc_info: - _run_executor(1, output) + _run_executor(1, events) assert "Credentials not provided" in str(exc_info.value) def test_executor_call_generic_error(): - output = ( - "fatal: [localhost]: FAILED! => " '{"msg": "repository not found"}' - ) + events = [_failed_event({"msg": "repository not found"})] with pytest.raises(scm.ScmError) as exc_info: - _run_executor(1, output) + _run_executor(1, events) assert "Project Import Error:" in str(exc_info.value) assert "repository not found" in str(exc_info.value) -def test_executor_call_no_fatal_output(): +def test_executor_call_no_error_event(): + events = [_ok_event({"changed": True})] with pytest.raises(scm.ScmError) as exc_info: - _run_executor(1, "some generic failure output") + _run_executor(1, events) assert "Project Import Error:" in str(exc_info.value) From cd4276a614ca7a5b9b514a94237213b74ac99c38 Mon Sep 17 00:00:00 2001 From: Jacob Craiglow Date: Fri, 17 Jul 2026 11:45:06 -0400 Subject: [PATCH 5/5] sonar issues: Refactor from regex to utilize ansible runner API Signed-off-by: Jacob Craiglow --- tests/unit/test_scm.py | 160 +++++++++++++++-------------------------- 1 file changed, 57 insertions(+), 103 deletions(-) diff --git a/tests/unit/test_scm.py b/tests/unit/test_scm.py index d9dbaa95a..a3429c192 100644 --- a/tests/unit/test_scm.py +++ b/tests/unit/test_scm.py @@ -42,7 +42,23 @@ def ssh_credential( ) -> models.EdaCredential: credential = models.EdaCredential.objects.create( name="test-ssh-credential", - inputs={"ssh_key_data": "-----BEGIN OPENSSH PRIVATE KEY-----\n"}, + inputs={ + "ssh_key_data": "-----BEGIN OPENSSH PRIVATE KEY-----\n", + "ssh_key_unlock": "passphrase", + }, + organization=default_organization, + ) + credential.refresh_from_db() + return credential + + +@pytest.fixture +def gpg_credential( + default_organization: models.Organization, +) -> models.EdaCredential: + credential = models.EdaCredential.objects.create( + name="test-gpg-credential", + inputs={"gpg_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n"}, organization=default_organization, ) credential.refresh_from_db() @@ -159,39 +175,31 @@ def spy_get(obj): @pytest.mark.django_db def test_set_proxy_environ_restores_existing_vars( credential: models.EdaCredential, + monkeypatch, ): """Pre-existing proxy env vars are restored after clone.""" executor = mock.MagicMock() original_value = "http://original-proxy:8080" for key in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"): - os.environ[key] = original_value + monkeypatch.setenv(key, original_value) - try: - with tempfile.TemporaryDirectory() as dest_path: - scm.ScmRepository.clone( - "https://git.example.com/repo.git", - dest_path, - credential=credential, - proxy="http://override-proxy:3128", - _executor=executor, - ) + with tempfile.TemporaryDirectory() as dest_path: + scm.ScmRepository.clone( + "https://git.example.com/repo.git", + dest_path, + credential=credential, + proxy="http://override-proxy:3128", + _executor=executor, + ) - for key in ( - "http_proxy", - "https_proxy", - "HTTP_PROXY", - "HTTPS_PROXY", - ): - assert os.environ.get(key) == original_value - finally: - for key in ( - "http_proxy", - "https_proxy", - "HTTP_PROXY", - "HTTPS_PROXY", - ): - os.environ.pop(key, None) + for key in ( + "http_proxy", + "https_proxy", + "HTTP_PROXY", + "HTTPS_PROXY", + ): + assert os.environ.get(key) == original_value @pytest.mark.django_db @@ -431,24 +439,28 @@ def test_is_refspec_valid(ref: str, is_branch: bool, expected: bool): def test_git_clone_ssh_key(ssh_credential: models.EdaCredential, url: str): """Clone with SSH key preserves URL and passes key_file.""" executor = mock.MagicMock() - with tempfile.TemporaryDirectory() as dest_path: - scm.ScmRepository.clone( - url, - dest_path, - credential=ssh_credential, - depth=1, - _executor=executor, - ) - executor.assert_called_once() - call_kwargs = executor.call_args - extra_vars = call_kwargs.kwargs["extra_vars"] + with mock.patch.object( + scm.ScmRepository, "decrypt_key_file" + ) as mock_decrypt: + with tempfile.TemporaryDirectory() as dest_path: + scm.ScmRepository.clone( + url, + dest_path, + credential=ssh_credential, + depth=1, + _executor=executor, + ) + executor.assert_called_once() + call_kwargs = executor.call_args + extra_vars = call_kwargs.kwargs["extra_vars"] - # URL must reach the executor unchanged - assert extra_vars["scm_url"] == url + # URL must reach the executor unchanged + assert extra_vars["scm_url"] == url - # SSH key file must be provided - assert "key_file" in extra_vars - assert extra_vars["key_file"] # non-empty path + # SSH key file must be provided + assert "key_file" in extra_vars + assert extra_vars["key_file"] # non-empty path + mock_decrypt.assert_called_once() ################################################################# @@ -528,18 +540,12 @@ def test_extract_error_msg_empty_events(): @pytest.mark.django_db def test_git_clone_gpg_credential( - default_organization: models.Organization, + gpg_credential: models.EdaCredential, ): """GPG credential sets up verify_commit and GNUPGHOME.""" - gpg_credential = models.EdaCredential.objects.create( - name="test-gpg-credential", - inputs={"gpg_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n"}, - organization=default_organization, - ) - gpg_credential.refresh_from_db() executor = mock.MagicMock() - with mock.patch.object(scm.ScmRepository, "add_gpg_key"): + with mock.patch.object(scm.ScmRepository, "add_gpg_key") as mock_add_gpg: with tempfile.TemporaryDirectory() as dest_path: scm.ScmRepository.clone( "https://git.example.com/repo.git", @@ -550,6 +556,7 @@ def test_git_clone_gpg_credential( call_kwargs = executor.call_args assert call_kwargs[1]["extra_vars"]["verify_commit"] == "true" assert "GNUPGHOME" in call_kwargs[1]["env_vars"] + mock_add_gpg.assert_called_once() @pytest.mark.django_db @@ -567,59 +574,6 @@ def test_git_clone_creates_directory(): assert os.path.isdir(dest_path) -@pytest.mark.django_db -def test_git_clone_decrypt_key_file( - default_organization: models.Organization, -): - """Clone calls decrypt_key_file when ssh_key_unlock is provided.""" - credential = models.EdaCredential.objects.create( - name="test-ssh-unlock", - inputs={ - "ssh_key_data": "-----BEGIN OPENSSH PRIVATE KEY-----\n", - "ssh_key_unlock": "passphrase", - }, - organization=default_organization, - ) - credential.refresh_from_db() - executor = mock.MagicMock() - - with mock.patch.object( - scm.ScmRepository, "decrypt_key_file" - ) as mock_decrypt: - with tempfile.TemporaryDirectory() as dest_path: - scm.ScmRepository.clone( - "git@git.example.com:repo.git", - dest_path, - credential=credential, - _executor=executor, - ) - mock_decrypt.assert_called_once() - - -@pytest.mark.django_db -def test_git_clone_gpg_add_key( - default_organization: models.Organization, -): - """Clone calls add_gpg_key when gpg_credential is provided.""" - gpg_credential = models.EdaCredential.objects.create( - name="test-gpg", - inputs={"gpg_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n"}, - organization=default_organization, - ) - gpg_credential.refresh_from_db() - executor = mock.MagicMock() - - with mock.patch.object(scm.ScmRepository, "add_gpg_key") as mock_add_gpg: - with tempfile.TemporaryDirectory() as dest_path: - scm.ScmRepository.clone( - "https://git.example.com/repo.git", - dest_path, - gpg_credential=gpg_credential, - _executor=executor, - ) - mock_add_gpg.assert_called_once() - - ################################################################# # Tests for ScmRepository.decrypt_key_file and add_gpg_key #################################################################