test(titan-iac): raise quality gate coverage for quality runner
This commit is contained in:
parent
655c26c589
commit
9451bb9c61
@ -365,5 +365,5 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
return 0 if summary["status"] == "ok" else 1
|
return 0 if summary["status"] == "ok" else 1
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__": # pragma: no cover - exercised via CLI execution
|
||||||
raise SystemExit(main())
|
raise SystemExit(main())
|
||||||
|
|||||||
326
testing/tests/test_quality_gate_helpers.py
Normal file
326
testing/tests/test_quality_gate_helpers.py
Normal file
@ -0,0 +1,326 @@
|
|||||||
|
"""Helper-branch coverage for the titan-iac quality gate runner."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import urllib.error
|
||||||
|
from pathlib import Path
|
||||||
|
from types import SimpleNamespace
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from testing import quality_gate
|
||||||
|
|
||||||
|
|
||||||
|
class _JsonResponse:
|
||||||
|
"""Simple context-manager response that returns a JSON payload."""
|
||||||
|
|
||||||
|
def __init__(self, payload: Any):
|
||||||
|
self._payload = payload
|
||||||
|
|
||||||
|
def __enter__(self) -> _JsonResponse:
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, exc_type, exc, tb) -> bool:
|
||||||
|
return False
|
||||||
|
|
||||||
|
def read(self) -> bytes:
|
||||||
|
return json.dumps(self._payload).encode("utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def test_env_flag_truthy_falsey_and_default(monkeypatch):
|
||||||
|
monkeypatch.delenv("QUALITY_FLAG", raising=False)
|
||||||
|
assert quality_gate._env_flag("QUALITY_FLAG", default=True) is True
|
||||||
|
|
||||||
|
monkeypatch.setenv("QUALITY_FLAG", " yes ")
|
||||||
|
assert quality_gate._env_flag("QUALITY_FLAG", default=False) is True
|
||||||
|
|
||||||
|
monkeypatch.setenv("QUALITY_FLAG", "0")
|
||||||
|
assert quality_gate._env_flag("QUALITY_FLAG", default=True) is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_json_report_handles_missing_invalid_and_non_object(tmp_path: Path):
|
||||||
|
payload, issue = quality_gate._load_json_report(tmp_path / "missing.json")
|
||||||
|
assert payload is None
|
||||||
|
assert issue and issue.startswith("report missing:")
|
||||||
|
|
||||||
|
invalid_path = tmp_path / "invalid.json"
|
||||||
|
invalid_path.write_text("{", encoding="utf-8")
|
||||||
|
payload, issue = quality_gate._load_json_report(invalid_path)
|
||||||
|
assert payload is None
|
||||||
|
assert issue and issue.startswith("report invalid JSON:")
|
||||||
|
|
||||||
|
list_path = tmp_path / "list.json"
|
||||||
|
list_path.write_text("[]", encoding="utf-8")
|
||||||
|
payload, issue = quality_gate._load_json_report(list_path)
|
||||||
|
assert payload is None
|
||||||
|
assert issue and issue.startswith("report payload must be an object:")
|
||||||
|
|
||||||
|
object_path = tmp_path / "object.json"
|
||||||
|
object_path.write_text('{"status":"OK"}', encoding="utf-8")
|
||||||
|
payload, issue = quality_gate._load_json_report(object_path)
|
||||||
|
assert payload == {"status": "OK"}
|
||||||
|
assert issue is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_sonarqube_status_extraction_fallbacks():
|
||||||
|
assert quality_gate._sonarqube_gate_status_from_report({"projectStatus": {"status": "OK"}}) == "OK"
|
||||||
|
assert quality_gate._sonarqube_gate_status_from_report({"status": "PASSED"}) == "PASSED"
|
||||||
|
assert quality_gate._sonarqube_gate_status_from_report({"projectStatus": {"state": "NONE"}}) == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_sonarqube_gate_status_happy_path_and_auth(monkeypatch):
|
||||||
|
seen_headers: dict[str, str] = {}
|
||||||
|
|
||||||
|
def fake_urlopen(request, timeout):
|
||||||
|
assert timeout == 5.0
|
||||||
|
seen_headers["authorization"] = request.headers.get("Authorization", "")
|
||||||
|
return _JsonResponse({"projectStatus": {"status": "OK"}})
|
||||||
|
|
||||||
|
monkeypatch.setattr(quality_gate.urllib.request, "urlopen", fake_urlopen)
|
||||||
|
status, issue = quality_gate._fetch_sonarqube_gate_status(
|
||||||
|
"http://sonar.local",
|
||||||
|
"atlas",
|
||||||
|
"token-value",
|
||||||
|
5.0,
|
||||||
|
)
|
||||||
|
assert status == "OK"
|
||||||
|
assert issue is None
|
||||||
|
assert seen_headers["authorization"].startswith("Basic ")
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_sonarqube_gate_status_handles_non_object_and_network_error(monkeypatch):
|
||||||
|
monkeypatch.setattr(quality_gate.urllib.request, "urlopen", lambda *_a, **_k: _JsonResponse(["not", "dict"]))
|
||||||
|
status, issue = quality_gate._fetch_sonarqube_gate_status("http://sonar", "atlas", "", 5.0)
|
||||||
|
assert status == ""
|
||||||
|
assert issue == "sonarqube query returned non-object payload"
|
||||||
|
|
||||||
|
def raise_url_error(*_a, **_k):
|
||||||
|
raise urllib.error.URLError("boom")
|
||||||
|
|
||||||
|
monkeypatch.setattr(quality_gate.urllib.request, "urlopen", raise_url_error)
|
||||||
|
status, issue = quality_gate._fetch_sonarqube_gate_status("http://sonar", "atlas", "", 5.0)
|
||||||
|
assert status == ""
|
||||||
|
assert issue and issue.startswith("sonarqube query failed:")
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_sonarqube_gate_status_missing_status_message(monkeypatch):
|
||||||
|
monkeypatch.setattr(quality_gate.urllib.request, "urlopen", lambda *_a, **_k: _JsonResponse({"projectStatus": {}}))
|
||||||
|
status, issue = quality_gate._fetch_sonarqube_gate_status("http://sonar", "atlas", "", 5.0)
|
||||||
|
assert status == ""
|
||||||
|
assert issue == "sonarqube response missing projectStatus.status"
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_sonarqube_check_uses_report_and_fails_when_status_is_not_ok(tmp_path: Path, monkeypatch):
|
||||||
|
build_dir = tmp_path / "build"
|
||||||
|
build_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
report = build_dir / "sonarqube-quality-gate.json"
|
||||||
|
report.write_text('{"projectStatus":{"status":"ERROR"}}', encoding="utf-8")
|
||||||
|
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_REPORT", str(report))
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_ENFORCE", "1")
|
||||||
|
|
||||||
|
result = quality_gate._run_sonarqube_check(build_dir)
|
||||||
|
assert result["source"] == "report"
|
||||||
|
assert result["gate_status"] == "ERROR"
|
||||||
|
assert result["status"] == "failed"
|
||||||
|
assert any("expected OK" in issue for issue in result["issues"])
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_sonarqube_check_uses_api_when_report_missing(tmp_path: Path, monkeypatch):
|
||||||
|
build_dir = tmp_path / "build"
|
||||||
|
build_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_REPORT", str(build_dir / "missing.json"))
|
||||||
|
monkeypatch.setenv("SONARQUBE_HOST_URL", "http://sonarqube")
|
||||||
|
monkeypatch.setenv("SONARQUBE_PROJECT_KEY", "atlas")
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_ENFORCE", "1")
|
||||||
|
monkeypatch.setattr(quality_gate, "_fetch_sonarqube_gate_status", lambda *_a, **_k: ("OK", None))
|
||||||
|
|
||||||
|
result = quality_gate._run_sonarqube_check(build_dir)
|
||||||
|
assert result["source"] == "api"
|
||||||
|
assert result["status"] == "ok"
|
||||||
|
assert result["gate_status"] == "OK"
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_sonarqube_check_relative_report_missing_status_and_missing_report(tmp_path: Path, monkeypatch):
|
||||||
|
build_dir = tmp_path / "build"
|
||||||
|
build_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
monkeypatch.chdir(tmp_path)
|
||||||
|
|
||||||
|
relative_report = Path("build") / "sonar.json"
|
||||||
|
(tmp_path / relative_report).write_text('{"projectStatus":{}}', encoding="utf-8")
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_REPORT", str(relative_report))
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_ENFORCE", "1")
|
||||||
|
monkeypatch.delenv("SONARQUBE_HOST_URL", raising=False)
|
||||||
|
monkeypatch.delenv("SONARQUBE_PROJECT_KEY", raising=False)
|
||||||
|
missing_status = quality_gate._run_sonarqube_check(build_dir)
|
||||||
|
assert any("missing quality gate status" in issue for issue in missing_status["issues"])
|
||||||
|
assert any("status unavailable" in issue for issue in missing_status["issues"])
|
||||||
|
assert missing_status["report_path"].endswith("build/sonar.json")
|
||||||
|
|
||||||
|
missing_relative = Path("build") / "missing-sonar.json"
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_REPORT", str(missing_relative))
|
||||||
|
missing_report = quality_gate._run_sonarqube_check(build_dir)
|
||||||
|
assert any("report missing:" in issue for issue in missing_report["issues"])
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_sonarqube_check_fallback_api_path_with_query_error(tmp_path: Path, monkeypatch):
|
||||||
|
build_dir = tmp_path / "build"
|
||||||
|
build_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_SONARQUBE_ENFORCE", "1")
|
||||||
|
monkeypatch.setenv("SONARQUBE_HOST_URL", "http://sonarqube")
|
||||||
|
monkeypatch.setenv("SONARQUBE_PROJECT_KEY", "atlas")
|
||||||
|
monkeypatch.setattr(quality_gate, "_load_json_report", lambda _path: (None, None))
|
||||||
|
monkeypatch.setattr(quality_gate, "_fetch_sonarqube_gate_status", lambda *_a, **_k: ("", "api exploded"))
|
||||||
|
|
||||||
|
result = quality_gate._run_sonarqube_check(build_dir)
|
||||||
|
assert result["source"] == "api"
|
||||||
|
assert "api exploded" in result["issues"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_ironbank_status_extraction_and_required_missing_behavior(tmp_path: Path, monkeypatch):
|
||||||
|
assert quality_gate._ironbank_status_from_report({"status": "compliant"}) == ("compliant", None)
|
||||||
|
assert quality_gate._ironbank_status_from_report({"compliant": True}) == ("compliant", True)
|
||||||
|
assert quality_gate._ironbank_status_from_report({"unrelated": "value"}) == ("", None)
|
||||||
|
|
||||||
|
build_dir = tmp_path / "build"
|
||||||
|
build_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_REPORT", str(build_dir / "missing.json"))
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_REQUIRED", "0")
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_ENFORCE", "1")
|
||||||
|
optional_result = quality_gate._run_ironbank_check(build_dir)
|
||||||
|
assert optional_result["status"] == "ok"
|
||||||
|
assert optional_result["required"] is False
|
||||||
|
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_REQUIRED", "1")
|
||||||
|
required_result = quality_gate._run_ironbank_check(build_dir)
|
||||||
|
assert required_result["status"] == "failed"
|
||||||
|
assert any("report missing:" in issue for issue in required_result["issues"])
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_ironbank_check_with_report_compliance_values(tmp_path: Path, monkeypatch):
|
||||||
|
build_dir = tmp_path / "build"
|
||||||
|
build_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
report = build_dir / "ironbank-compliance.json"
|
||||||
|
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_REPORT", str(report))
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_REQUIRED", "1")
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_ENFORCE", "1")
|
||||||
|
|
||||||
|
report.write_text('{"status":"compliant"}', encoding="utf-8")
|
||||||
|
ok_result = quality_gate._run_ironbank_check(build_dir)
|
||||||
|
assert ok_result["status"] == "ok"
|
||||||
|
assert ok_result["source"] == "report"
|
||||||
|
|
||||||
|
report.write_text('{"compliant": false}', encoding="utf-8")
|
||||||
|
bad_result = quality_gate._run_ironbank_check(build_dir)
|
||||||
|
assert bad_result["status"] == "failed"
|
||||||
|
assert any("expected compliant" in issue for issue in bad_result["issues"])
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_ironbank_check_relative_path_and_unavailable_status(tmp_path: Path, monkeypatch):
|
||||||
|
build_dir = tmp_path / "build"
|
||||||
|
build_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
monkeypatch.chdir(tmp_path)
|
||||||
|
|
||||||
|
report_rel = Path("build") / "ironbank.json"
|
||||||
|
(tmp_path / report_rel).write_text("{}", encoding="utf-8")
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_REPORT", str(report_rel))
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_REQUIRED", "1")
|
||||||
|
monkeypatch.setenv("QUALITY_GATE_IRONBANK_ENFORCE", "1")
|
||||||
|
|
||||||
|
result = quality_gate._run_ironbank_check(build_dir)
|
||||||
|
assert result["report_path"].endswith("build/ironbank.json")
|
||||||
|
assert result["status"] == "failed"
|
||||||
|
assert "ironbank compliance status unavailable" in result["issues"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_status_and_result_helpers():
|
||||||
|
assert quality_gate._status_from_issues([]) == "ok"
|
||||||
|
assert quality_gate._status_from_issues(["problem"]) == "failed"
|
||||||
|
|
||||||
|
result = quality_gate._result("docs", "desc", "ok", extra=True)
|
||||||
|
assert result == {"name": "docs", "description": "desc", "status": "ok", "extra": True}
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_ruff_returns_failed_when_subprocess_nonzero(monkeypatch, tmp_path: Path):
|
||||||
|
monkeypatch.setattr(quality_gate.time, "monotonic", lambda: 10.0)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
quality_gate.subprocess,
|
||||||
|
"run",
|
||||||
|
lambda *args, **kwargs: SimpleNamespace(returncode=2),
|
||||||
|
)
|
||||||
|
contract = {"lint_paths": ["testing"]}
|
||||||
|
result = quality_gate._run_ruff(contract, tmp_path)
|
||||||
|
assert result["name"] == "smell"
|
||||||
|
assert result["status"] == "failed"
|
||||||
|
assert result["returncode"] == 2
|
||||||
|
assert result["command"][:3] == [quality_gate.sys.executable, "-m", "ruff"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_pytest_suite_builds_commands_with_and_without_coverage(monkeypatch, tmp_path: Path):
|
||||||
|
suite_with_cov = {
|
||||||
|
"description": "Unit suite",
|
||||||
|
"paths": ["testing/tests"],
|
||||||
|
"junit": "build/junit.xml",
|
||||||
|
"coverage_xml": "build/coverage.xml",
|
||||||
|
"coverage_sources": ["testing"],
|
||||||
|
}
|
||||||
|
suite_without_cov = {
|
||||||
|
"description": "Smoke suite",
|
||||||
|
"paths": ["testing/tests"],
|
||||||
|
"junit": "build/junit-smoke.xml",
|
||||||
|
}
|
||||||
|
|
||||||
|
commands: list[list[str]] = []
|
||||||
|
monkeypatch.setattr(quality_gate.time, "monotonic", lambda: 20.0)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
quality_gate.subprocess,
|
||||||
|
"run",
|
||||||
|
lambda command, cwd, check: commands.append(command) or SimpleNamespace(returncode=0),
|
||||||
|
)
|
||||||
|
|
||||||
|
first = quality_gate._run_pytest_suite(tmp_path, "unit", suite_with_cov)
|
||||||
|
second = quality_gate._run_pytest_suite(tmp_path, "smoke", suite_without_cov)
|
||||||
|
|
||||||
|
assert first["status"] == "ok"
|
||||||
|
assert first["coverage_xml"] == "build/coverage.xml"
|
||||||
|
assert any(item.startswith("--cov=testing") for item in commands[0])
|
||||||
|
assert second["status"] == "ok"
|
||||||
|
assert second["coverage_xml"] is None
|
||||||
|
assert not any(item.startswith("--cov=") for item in commands[1])
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_profile_unknown_profile_and_unknown_check(tmp_path: Path):
|
||||||
|
with pytest.raises(SystemExit, match="unknown profile: jenkins"):
|
||||||
|
quality_gate.run_profile({"profiles": {}}, tmp_path, "jenkins", tmp_path / "build")
|
||||||
|
|
||||||
|
contract = {"profiles": {"local": ["missing"]}, "pytest_suites": {}}
|
||||||
|
with pytest.raises(SystemExit, match="references unknown check: missing"):
|
||||||
|
quality_gate.run_profile(contract, tmp_path, "local", tmp_path / "build")
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_profile_routes_sonarqube_and_ironbank_checks(tmp_path: Path, monkeypatch):
|
||||||
|
contract = {"profiles": {"local": ["sonarqube", "ironbank"]}, "pytest_suites": {}}
|
||||||
|
monkeypatch.setattr(quality_gate, "_run_sonarqube_check", lambda _build_dir: {"name": "sonarqube", "status": "ok"})
|
||||||
|
monkeypatch.setattr(quality_gate, "_run_ironbank_check", lambda _build_dir: {"name": "ironbank", "status": "ok"})
|
||||||
|
|
||||||
|
summary = quality_gate.run_profile(contract, tmp_path, "local", tmp_path / "build")
|
||||||
|
assert summary["status"] == "ok"
|
||||||
|
assert [item["name"] for item in summary["results"]] == ["sonarqube", "ironbank"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_returns_failure_when_profile_reports_failed(tmp_path: Path, monkeypatch):
|
||||||
|
monkeypatch.chdir(tmp_path)
|
||||||
|
monkeypatch.setattr(quality_gate, "load_contract", lambda: {"profiles": {"local": []}, "pytest_suites": {}})
|
||||||
|
monkeypatch.setattr(
|
||||||
|
quality_gate,
|
||||||
|
"run_profile",
|
||||||
|
lambda *_a, **_k: {"status": "failed", "profile": "local", "results": [], "manual_scripts": []},
|
||||||
|
)
|
||||||
|
|
||||||
|
rc = quality_gate.main(["--profile", "local", "--build-dir", "build"])
|
||||||
|
assert rc == 1
|
||||||
|
assert (tmp_path / "build" / "quality-gate-summary.json").exists()
|
||||||
Loading…
x
Reference in New Issue
Block a user