Compare commits
3 Commits
e8c26ecf85
...
f77119b238
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f77119b238 | ||
|
|
64272f52d2 | ||
|
|
3a57371989 |
@ -93,9 +93,8 @@ def _sonar_issue_from_finding(finding: dict[str, Any]) -> dict[str, Any] | None:
|
||||
start = finding.get("start") if isinstance(finding.get("start"), dict) else {}
|
||||
end = finding.get("end") if isinstance(finding.get("end"), dict) else {}
|
||||
start_line = _line_number(start.get("line") if isinstance(start, dict) else None)
|
||||
# _line_number floors the end line at start_line, so the range never inverts.
|
||||
end_line = _line_number(end.get("line") if isinstance(end, dict) else None, start_line)
|
||||
if end_line < start_line:
|
||||
end_line = start_line
|
||||
return {
|
||||
"engineId": "semgrep",
|
||||
"ruleId": str(finding.get("check_id") or "semgrep.unknown"),
|
||||
|
||||
@ -232,6 +232,24 @@ def test_retry_db_connect_reraises_final_error(monkeypatch):
|
||||
sync.retry_db_connect(attempts=1)
|
||||
|
||||
|
||||
def test_retry_helpers_return_none_without_attempts(monkeypatch):
|
||||
sync = load_sync_module(monkeypatch)
|
||||
|
||||
assert sync.retry_request("request", lambda: "never", attempts=0) is None
|
||||
assert sync.retry_db_connect(attempts=0) is None
|
||||
|
||||
|
||||
def test_kc_update_attributes_replaces_non_dict_attributes(monkeypatch):
|
||||
sync = load_sync_module(monkeypatch)
|
||||
current_resp = _FakeResponse({"attributes": "not-a-dict"})
|
||||
ok_resp = _FakeResponse({"attributes": {"mailu_app_password": ["abc"]}})
|
||||
sync.SESSION = _FakeSession(_FakeResponse({}), [current_resp, ok_resp])
|
||||
|
||||
sync.kc_update_attributes("token", {"id": "u1", "username": "u1"}, {"mailu_app_password": "abc"})
|
||||
|
||||
assert sync.SESSION.put_called
|
||||
|
||||
|
||||
def test_ensure_mailu_user_skips_foreign_domain(monkeypatch):
|
||||
sync = load_sync_module(monkeypatch)
|
||||
executed = []
|
||||
@ -405,3 +423,62 @@ def test_main_generates_password_and_upserts(monkeypatch):
|
||||
# Only mail-enabled users (or legacy users with a mailbox) are synced and backfilled.
|
||||
assert len(updated) == 3
|
||||
assert conns and len(conns[0]._cursor.executions) == 3
|
||||
|
||||
|
||||
def test_main_skips_disabled_users_and_provisioned_users_without_update(monkeypatch):
|
||||
sync = load_sync_module(monkeypatch)
|
||||
monkeypatch.setattr(sync.bcrypt_sha256, "hash", lambda password: f"hash:{password}")
|
||||
users = [
|
||||
{
|
||||
"id": "u1",
|
||||
"username": "disabled",
|
||||
"email": "disabled@example.com",
|
||||
"enabled": False,
|
||||
"attributes": {"mailu_enabled": ["true"]},
|
||||
},
|
||||
{
|
||||
"id": "u2",
|
||||
"username": "settled",
|
||||
"email": "settled@example.com",
|
||||
"attributes": {
|
||||
"mailu_enabled": ["true"],
|
||||
"mailu_email": ["settled@example.com"],
|
||||
"mailu_app_password": ["existing"],
|
||||
},
|
||||
},
|
||||
]
|
||||
updated = []
|
||||
|
||||
class _Cursor:
|
||||
def __init__(self):
|
||||
self.executions = []
|
||||
|
||||
def execute(self, sql, params):
|
||||
self.executions.append(params)
|
||||
|
||||
def close(self):
|
||||
return None
|
||||
|
||||
class _Conn:
|
||||
def __init__(self):
|
||||
self.autocommit = False
|
||||
self._cursor = _Cursor()
|
||||
|
||||
def cursor(self, cursor_factory=None):
|
||||
return self._cursor
|
||||
|
||||
def close(self):
|
||||
return None
|
||||
|
||||
conn = _Conn()
|
||||
monkeypatch.setattr(sync, "get_kc_token", lambda: "tok")
|
||||
monkeypatch.setattr(sync, "kc_get_users", lambda token: users)
|
||||
monkeypatch.setattr(sync, "kc_update_attributes", lambda token, user, attrs: updated.append(user["id"]))
|
||||
monkeypatch.setattr(sync.psycopg2, "connect", lambda **kwargs: conn)
|
||||
|
||||
sync.main()
|
||||
|
||||
# The already-provisioned user is synced without a Keycloak write; the
|
||||
# disabled user never reaches the mailbox upsert.
|
||||
assert updated == []
|
||||
assert len(conn._cursor.executions) == 1
|
||||
|
||||
@ -127,6 +127,18 @@ def test_listener_post_wait_keeps_running_request_successful(monkeypatch):
|
||||
assert handler.responses == [200]
|
||||
|
||||
|
||||
def test_listener_post_treats_non_object_json_as_plain_trigger(monkeypatch):
|
||||
listener = load_listener_module(monkeypatch)
|
||||
called = []
|
||||
monkeypatch.setattr(listener, "_trigger_sync_async", lambda force=False: called.append(force) or True)
|
||||
handler = _handler_for(listener, "[1, 2]")
|
||||
|
||||
handler.do_POST()
|
||||
|
||||
assert called == [False]
|
||||
assert handler.responses == [202]
|
||||
|
||||
|
||||
def test_listener_log_message_is_quiet(monkeypatch):
|
||||
listener = load_listener_module(monkeypatch)
|
||||
handler = listener.Handler.__new__(listener.Handler)
|
||||
|
||||
@ -374,5 +374,5 @@ def main() -> int:
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == "__main__": # pragma: no cover - exercised through main()
|
||||
raise SystemExit(main())
|
||||
|
||||
@ -314,7 +314,7 @@ def main():
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == "__main__": # pragma: no cover - exercised through main()
|
||||
try:
|
||||
main()
|
||||
except Exception as exc:
|
||||
|
||||
@ -100,6 +100,6 @@ class Handler(http.server.BaseHTTPRequestHandler):
|
||||
return
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == "__main__": # pragma: no cover - server loop
|
||||
server = http.server.ThreadingHTTPServer(("", 8080), Handler)
|
||||
server.serve_forever()
|
||||
|
||||
@ -7,8 +7,8 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]:
|
||||
"""Load per-file line-rate percentages from a Cobertura XML report."""
|
||||
def _load_rates(xml_path: Path, root: Path) -> dict[str, dict[str, float | None]]:
|
||||
"""Load per-file line and branch percentages from a Cobertura XML report."""
|
||||
tree = ET.parse(xml_path)
|
||||
xml_root = tree.getroot()
|
||||
source_roots = [
|
||||
@ -16,7 +16,7 @@ def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]:
|
||||
for node in xml_root.findall("./sources/source")
|
||||
if node.text
|
||||
]
|
||||
percentages: dict[str, float] = {}
|
||||
rates: dict[str, dict[str, float | None]] = {}
|
||||
for class_node in xml_root.findall(".//class"):
|
||||
filename = class_node.attrib.get("filename")
|
||||
line_rate = class_node.attrib.get("line-rate")
|
||||
@ -32,33 +32,44 @@ def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]:
|
||||
if candidate.exists():
|
||||
key = candidate.relative_to(root).as_posix()
|
||||
break
|
||||
percentages[key] = float(line_rate) * 100.0
|
||||
return percentages
|
||||
branch_rate = class_node.attrib.get("branch-rate")
|
||||
rates[key] = {
|
||||
"line": float(line_rate) * 100.0,
|
||||
"branch": None if branch_rate is None else float(branch_rate) * 100.0,
|
||||
}
|
||||
return rates
|
||||
|
||||
|
||||
def run_check(contract: dict[str, Any], root: Path, xml_path: Path) -> list[str]:
|
||||
"""Return human-readable issues for tracked files below the coverage floor.
|
||||
|
||||
The report is intentionally per-file so a single weak module cannot hide
|
||||
behind aggregate suite coverage.
|
||||
Line and branch rates are both enforced per file so a single weak module
|
||||
cannot hide behind aggregate suite coverage or line-only reporting.
|
||||
"""
|
||||
if not xml_path.exists():
|
||||
return [f"coverage xml missing: {xml_path.relative_to(root)}"]
|
||||
|
||||
percentages = _load_percentages(xml_path, root)
|
||||
rates = _load_rates(xml_path, root)
|
||||
minimum = float(contract.get("coverage", {}).get("minimum_percent", 95.0))
|
||||
issues: list[str] = []
|
||||
|
||||
for relative_path in contract.get("coverage", {}).get("tracked_files", []):
|
||||
normalized = relative_path.replace("\\", "/")
|
||||
percent = percentages.get(normalized)
|
||||
if percent is None:
|
||||
file_rates = rates.get(normalized)
|
||||
if file_rates is None:
|
||||
issues.append(f"coverage missing for tracked file: {relative_path}")
|
||||
continue
|
||||
if percent + 1e-9 < minimum:
|
||||
issues.append(
|
||||
f"coverage below {minimum:.1f}%: {relative_path} ({percent:.1f}%)"
|
||||
)
|
||||
for metric in ("line", "branch"):
|
||||
percent = file_rates[metric]
|
||||
if percent is None:
|
||||
issues.append(
|
||||
f"{metric} coverage missing for tracked file: {relative_path}"
|
||||
)
|
||||
elif percent + 1e-9 < minimum:
|
||||
issues.append(
|
||||
f"{metric} coverage below {minimum:.1f}%: "
|
||||
f"{relative_path} ({percent:.1f}%)"
|
||||
)
|
||||
|
||||
return issues
|
||||
|
||||
@ -73,13 +84,13 @@ def compute_workspace_line_coverage(
|
||||
if not xml_path.exists():
|
||||
return 0.0
|
||||
|
||||
percentages = _load_percentages(xml_path, root)
|
||||
rates = _load_rates(xml_path, root)
|
||||
samples: list[float] = []
|
||||
for relative_path in contract.get("coverage", {}).get("tracked_files", []):
|
||||
normalized = relative_path.replace("\\", "/")
|
||||
percent = percentages.get(normalized)
|
||||
if percent is not None:
|
||||
samples.append(percent)
|
||||
file_rates = rates.get(normalized)
|
||||
if file_rates is not None:
|
||||
samples.append(file_rates["line"])
|
||||
if not samples:
|
||||
return 0.0
|
||||
return round(sum(samples) / len(samples), 3)
|
||||
|
||||
@ -332,7 +332,7 @@ def run_profile(
|
||||
results.append(
|
||||
_result(
|
||||
"coverage",
|
||||
"Per-file 95% coverage floor for tracked quality-managed modules.",
|
||||
"Per-file 95% line and branch coverage floor for tracked quality-managed modules.",
|
||||
_status_from_issues(issues),
|
||||
issues=issues,
|
||||
coverage_xml=str(coverage_xml.relative_to(root)),
|
||||
|
||||
227
testing/tests/test_hermes_image_builder_branch_edges.py
Normal file
227
testing/tests/test_hermes_image_builder_branch_edges.py
Normal file
@ -0,0 +1,227 @@
|
||||
"""Branch-edge coverage for the Hermes release and Harbor policy scripts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
RELEASE = REPO_ROOT / "ci/scripts/hermes_image_release.py"
|
||||
HARBOR = (
|
||||
REPO_ROOT
|
||||
/ "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py"
|
||||
)
|
||||
|
||||
|
||||
def _load(path: Path, name: str):
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
assert spec is not None and spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class Response(io.BytesIO):
|
||||
"""Context-managed urllib response used without network access."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
status: int,
|
||||
body: bytes = b"",
|
||||
headers: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
super().__init__(body)
|
||||
self.status = status
|
||||
self.headers = headers or {}
|
||||
|
||||
|
||||
class FakeClient:
|
||||
"""Ordered Harbor response fake with the pinned API origin."""
|
||||
|
||||
origin = "https://registry.bstein.dev/api/v2.0"
|
||||
|
||||
def __init__(self, responses) -> None:
|
||||
self.responses = list(responses)
|
||||
self.calls = []
|
||||
|
||||
def request(self, method, path, payload=None):
|
||||
self.calls.append((method, path, payload))
|
||||
return self.responses.pop(0)
|
||||
|
||||
|
||||
def _count(value: int) -> dict[str, str]:
|
||||
return {"X-Total-Count": str(value)}
|
||||
|
||||
|
||||
def _publisher(module, *, immutable: bool = False, with_duration: bool = True) -> dict:
|
||||
access = [
|
||||
{"resource": "repository", "action": "pull"},
|
||||
{"resource": "repository", "action": "push"},
|
||||
]
|
||||
if immutable:
|
||||
access.append({"resource": "immutable-tag", "action": "list"})
|
||||
robot = {
|
||||
"id": 41,
|
||||
"name": module.PUBLISH_ROBOT,
|
||||
"description": "publisher",
|
||||
"level": "system",
|
||||
"editable": True,
|
||||
"disable": False,
|
||||
"permissions": [
|
||||
{"kind": "project", "namespace": module.PROJECT, "access": access}
|
||||
],
|
||||
}
|
||||
if with_duration:
|
||||
robot["duration"] = -1
|
||||
return robot
|
||||
|
||||
|
||||
def test_release_redirect_handler_never_forwards_credentials() -> None:
|
||||
"""The registry opener refuses to follow any redirect target."""
|
||||
module = _load(RELEASE, "release_redirect_edges")
|
||||
handler = module._NoRedirect()
|
||||
assert handler.redirect_request(None, None, 0, None, None, None) is None
|
||||
|
||||
|
||||
def test_release_policy_read_requires_credentials_and_bounded_body() -> None:
|
||||
"""The policy preflight rejects empty credentials and oversized responses."""
|
||||
module = _load(RELEASE, "release_policy_read_edges")
|
||||
with pytest.raises(RuntimeError, match="credentials are empty"):
|
||||
module._immutable_rules_response(username="", password="private")
|
||||
with pytest.raises(RuntimeError, match="size limit"):
|
||||
module._immutable_rules_response(
|
||||
username="robot",
|
||||
password="private",
|
||||
opener=lambda *_args: Response(200, b"x" * 1_048_577),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("headers", [{}, {"X-Total-Count": "many"}])
|
||||
def test_release_policy_rejects_missing_or_invalid_count(headers: dict) -> None:
|
||||
"""A proxy that strips or mangles count evidence cannot prove completeness."""
|
||||
module = _load(RELEASE, "release_policy_count_edges")
|
||||
with pytest.raises(RuntimeError, match="total count"):
|
||||
module.verify_immutable_policy(
|
||||
username="robot",
|
||||
password="private",
|
||||
opener=lambda *_args, evidence=headers: Response(200, b"[]", evidence),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("body", [b"{}", b"[1]"])
|
||||
def test_release_policy_rejects_non_rule_list_shapes(body: bytes) -> None:
|
||||
"""Valid JSON that is not a list of rule objects fails the preflight."""
|
||||
module = _load(RELEASE, "release_policy_shape_edges")
|
||||
with pytest.raises(RuntimeError, match="invalid shape"):
|
||||
module.verify_immutable_policy(
|
||||
username="robot",
|
||||
password="private",
|
||||
opener=lambda *_args, value=body: Response(200, value, _count(1)),
|
||||
)
|
||||
|
||||
|
||||
def test_renderer_ignores_trailing_entry_without_digest_at_eof() -> None:
|
||||
"""A dangling final image entry cannot shadow the one complete digest."""
|
||||
module = _load(RELEASE, "release_render_eof_edges")
|
||||
digest = "sha256:" + "6" * 64
|
||||
source = (
|
||||
"images:\n"
|
||||
f" - name: {module.DEFAULT_IMAGE}\n"
|
||||
" digest: sha256:" + "0" * 64 + "\n"
|
||||
f" - name: {module.DEFAULT_IMAGE}\n"
|
||||
" newTag: dangling\n"
|
||||
)
|
||||
rendered = module.render_kustomization(source, digest)
|
||||
assert f"digest: {digest}\n" in rendered
|
||||
assert rendered.endswith("newTag: dangling\n")
|
||||
|
||||
|
||||
def test_harbor_verified_publisher_rejects_exact_identity_with_bad_scope() -> None:
|
||||
"""An exact robot identity still fails verification on a malformed scope."""
|
||||
module = _load(HARBOR, "harbor_verified_scope_edges")
|
||||
robot = _publisher(module)
|
||||
robot["permissions"] = "bad"
|
||||
assert module._verified_publisher(robot) is False
|
||||
|
||||
|
||||
def test_harbor_publisher_update_omits_absent_duration(monkeypatch) -> None:
|
||||
"""A robot without a configured duration is preserved without inventing one."""
|
||||
module = _load(HARBOR, "harbor_duration_edges")
|
||||
monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
|
||||
verified = _publisher(module, immutable=True)
|
||||
client = FakeClient(
|
||||
[
|
||||
(200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
|
||||
(200, json.dumps(_publisher(module, with_duration=False)).encode(), {}),
|
||||
(200, b"", {}),
|
||||
(200, json.dumps(verified).encode(), {}),
|
||||
]
|
||||
)
|
||||
assert module.ensure_publisher_can_read_rule(client) == 41
|
||||
method, path, payload = client.calls[2]
|
||||
assert (method, path) == ("PUT", "/robots/41")
|
||||
assert "duration" not in payload
|
||||
|
||||
|
||||
def test_harbor_publisher_verify_retries_past_invalid_json(monkeypatch) -> None:
|
||||
"""One stale unreadable reread retries instead of passing or failing outright."""
|
||||
module = _load(HARBOR, "harbor_verify_retry_edges")
|
||||
sleeps = []
|
||||
monkeypatch.setattr(module.time, "sleep", sleeps.append)
|
||||
verified = _publisher(module, immutable=True)
|
||||
client = FakeClient(
|
||||
[
|
||||
(200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
|
||||
(200, json.dumps(_publisher(module)).encode(), {}),
|
||||
(200, b"", {}),
|
||||
(200, b"not-json", {}),
|
||||
(200, json.dumps(verified).encode(), {}),
|
||||
]
|
||||
)
|
||||
assert module.ensure_publisher_can_read_rule(client) == 41
|
||||
assert sleeps == [1]
|
||||
|
||||
|
||||
def test_harbor_rule_verify_requires_the_created_rule_id(monkeypatch) -> None:
|
||||
"""A matching rule under a different ID is a stale read, not creation proof."""
|
||||
module = _load(HARBOR, "harbor_rule_id_retry_edges")
|
||||
sleeps = []
|
||||
monkeypatch.setattr(module.time, "sleep", sleeps.append)
|
||||
location = "/api/v2.0/projects/bstein/immutabletagrules/23"
|
||||
client = FakeClient(
|
||||
[
|
||||
(200, b"[]", _count(0)),
|
||||
(201, b"", {"Location": location}),
|
||||
(200, json.dumps([{"id": 99, **module.EXPECTED_RULE}]).encode(), _count(1)),
|
||||
(200, json.dumps([{"id": 23, **module.EXPECTED_RULE}]).encode(), _count(1)),
|
||||
]
|
||||
)
|
||||
assert module.ensure_rule(client) == 23
|
||||
assert sleeps == [1]
|
||||
|
||||
|
||||
def test_harbor_main_reraises_after_bounded_unavailable_retries(
|
||||
monkeypatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Persistent Harbor unavailability surfaces instead of looping forever."""
|
||||
module = _load(HARBOR, "harbor_main_retry_edges")
|
||||
password_file = tmp_path / "password"
|
||||
password_file.write_text("private\n", encoding="utf-8")
|
||||
monkeypatch.setenv("HARBOR_API_ORIGIN", module.EXPECTED_ORIGIN)
|
||||
monkeypatch.setenv("HARBOR_ADMIN_PASSWORD_FILE", str(password_file))
|
||||
monkeypatch.setattr(module, "HarborClient", lambda *_args: object())
|
||||
|
||||
def never_ready(_client):
|
||||
raise module.HarborUnavailable("still warming")
|
||||
|
||||
monkeypatch.setattr(module, "ensure_rule", never_ready)
|
||||
sleeps = []
|
||||
monkeypatch.setattr(module.time, "sleep", sleeps.append)
|
||||
with pytest.raises(module.HarborUnavailable, match="still warming"):
|
||||
module.main()
|
||||
assert len(sleeps) == 11
|
||||
@ -98,8 +98,9 @@ def test_coverage_check_enforces_per_file_floor(tmp_path: Path):
|
||||
<packages>
|
||||
<package>
|
||||
<classes>
|
||||
<class filename="ok.py" line-rate="1.0" />
|
||||
<class filename="low.py" line-rate="0.90" />
|
||||
<class filename="ok.py" line-rate="1.0" branch-rate="1.0" />
|
||||
<class filename="low.py" line-rate="0.90" branch-rate="1.0" />
|
||||
<class filename="weak_branches.py" line-rate="1.0" branch-rate="0.90" />
|
||||
</classes>
|
||||
</package>
|
||||
</packages>
|
||||
@ -112,13 +113,14 @@ def test_coverage_check_enforces_per_file_floor(tmp_path: Path):
|
||||
contract = {
|
||||
"coverage": {
|
||||
"minimum_percent": 95.0,
|
||||
"tracked_files": ["ok.py", "low.py", "missing.py"],
|
||||
"tracked_files": ["ok.py", "low.py", "weak_branches.py", "missing.py"],
|
||||
}
|
||||
}
|
||||
|
||||
issues = run_coverage_check(contract, tmp_path, coverage_xml)
|
||||
|
||||
assert "coverage below 95.0%: low.py (90.0%)" in issues
|
||||
assert "line coverage below 95.0%: low.py (90.0%)" in issues
|
||||
assert "branch coverage below 95.0%: weak_branches.py (90.0%)" in issues
|
||||
assert "coverage missing for tracked file: missing.py" in issues
|
||||
|
||||
|
||||
@ -142,8 +144,8 @@ def test_coverage_check_handles_missing_xml_and_source_root_mapping(tmp_path: Pa
|
||||
<packages>
|
||||
<package>
|
||||
<classes>
|
||||
<class filename="mapped.py" line-rate="1.0" />
|
||||
<class filename="{(tmp_path / 'absolute.py').as_posix()}" line-rate="1.0" />
|
||||
<class filename="mapped.py" line-rate="1.0" branch-rate="1.0" />
|
||||
<class filename="{(tmp_path / 'absolute.py').as_posix()}" line-rate="1.0" branch-rate="1.0" />
|
||||
<class filename="skip.py" />
|
||||
</classes>
|
||||
</package>
|
||||
|
||||
@ -57,7 +57,7 @@ def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path:
|
||||
<packages>
|
||||
<package>
|
||||
<classes>
|
||||
<class filename="relative.py" line-rate="0.80" />
|
||||
<class filename="relative.py" line-rate="0.80" branch-rate="1.0" />
|
||||
</classes>
|
||||
</package>
|
||||
</packages>
|
||||
@ -73,4 +73,65 @@ def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path:
|
||||
coverage_xml,
|
||||
)
|
||||
|
||||
assert issues == ["coverage below 95.0%: relative.py (80.0%)"]
|
||||
assert issues == ["line coverage below 95.0%: relative.py (80.0%)"]
|
||||
|
||||
|
||||
def test_run_check_fails_a_file_below_the_branch_floor(tmp_path: Path) -> None:
|
||||
"""Full line coverage must never hide a file whose branch coverage is weak."""
|
||||
|
||||
coverage_xml = tmp_path / "coverage.xml"
|
||||
coverage_xml.write_text(
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
<coverage>
|
||||
<packages>
|
||||
<package>
|
||||
<classes>
|
||||
<class filename="weak.py" line-rate="1.0" branch-rate="0.90" />
|
||||
<class filename="solid.py" line-rate="1.0" branch-rate="0.95" />
|
||||
</classes>
|
||||
</package>
|
||||
</packages>
|
||||
</coverage>
|
||||
"""
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
issues = run_check(
|
||||
{"coverage": {"minimum_percent": 95.0, "tracked_files": ["weak.py", "solid.py"]}},
|
||||
tmp_path,
|
||||
coverage_xml,
|
||||
)
|
||||
|
||||
assert issues == ["branch coverage below 95.0%: weak.py (90.0%)"]
|
||||
|
||||
|
||||
def test_run_check_fails_closed_when_branch_evidence_is_absent(tmp_path: Path) -> None:
|
||||
"""A report generated without branch measurement cannot satisfy the gate."""
|
||||
|
||||
coverage_xml = tmp_path / "coverage.xml"
|
||||
coverage_xml.write_text(
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
<coverage>
|
||||
<packages>
|
||||
<package>
|
||||
<classes>
|
||||
<class filename="unmeasured.py" line-rate="1.0" />
|
||||
</classes>
|
||||
</package>
|
||||
</packages>
|
||||
</coverage>
|
||||
"""
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
issues = run_check(
|
||||
{"coverage": {"minimum_percent": 95.0, "tracked_files": ["unmeasured.py"]}},
|
||||
tmp_path,
|
||||
coverage_xml,
|
||||
)
|
||||
|
||||
assert issues == ["branch coverage missing for tracked file: unmeasured.py"]
|
||||
|
||||
@ -60,6 +60,32 @@ def test_run_profile_aggregates_internal_and_pytest_results(tmp_path: Path, monk
|
||||
assert any(result.get("junit") == "build/junit-unit.xml" for result in summary["results"])
|
||||
|
||||
|
||||
def test_run_profile_fails_synthetic_file_below_branch_floor(tmp_path: Path):
|
||||
"""The gate itself must fail a tracked file with full lines but weak branches."""
|
||||
build_dir = tmp_path / "build"
|
||||
build_dir.mkdir()
|
||||
(build_dir / "coverage-unit.xml").write_text(
|
||||
"<coverage><packages><package><classes>"
|
||||
'<class filename="weak.py" line-rate="1.0" branch-rate="0.90" />'
|
||||
"</classes></package></packages></coverage>",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
contract = {
|
||||
"profiles": {"local": ["coverage"]},
|
||||
"pytest_suites": {"unit": {"coverage_xml": "build/coverage-unit.xml"}},
|
||||
"coverage": {"minimum_percent": 95.0, "tracked_files": ["weak.py"]},
|
||||
}
|
||||
|
||||
summary = quality_gate.run_profile(contract, tmp_path, "local", build_dir)
|
||||
|
||||
assert summary["status"] == "failed"
|
||||
coverage_result = summary["results"][0]
|
||||
assert coverage_result["name"] == "coverage"
|
||||
assert coverage_result["issues"] == ["branch coverage below 95.0%: weak.py (90.0%)"]
|
||||
assert summary["workspace_line_coverage_percent"] == 100.0
|
||||
|
||||
|
||||
def test_main_writes_summary_file(tmp_path: Path, monkeypatch):
|
||||
summary = {"status": "ok", "profile": "local", "results": [], "manual_scripts": []}
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
@ -4,7 +4,33 @@ from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from testing.quality_hygiene import count_files_over_line_limit
|
||||
from testing.quality_hygiene import count_files_over_line_limit, run_check
|
||||
|
||||
|
||||
def test_run_check_skips_conftest_files_in_naming_rules(tmp_path: Path) -> None:
|
||||
"""A conftest.py is pytest plumbing, never a naming-rule violation."""
|
||||
|
||||
tests_dir = tmp_path / "tests"
|
||||
tests_dir.mkdir()
|
||||
(tests_dir / "conftest.py").write_text("fixtures = True\n", encoding="utf-8")
|
||||
(tests_dir / "helper.py").write_text("value = 1\n", encoding="utf-8")
|
||||
|
||||
contract = {
|
||||
"hygiene": {
|
||||
"naming_rules": [
|
||||
{
|
||||
"glob": "tests/*.py",
|
||||
"pattern": "^test_[a-z0-9_]+\\.py$",
|
||||
"description": "Pytest files use test_*.py names.",
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
issues = run_check(contract, tmp_path)
|
||||
|
||||
assert len(issues) == 1
|
||||
assert "helper.py" in issues[0]
|
||||
|
||||
|
||||
def test_count_files_over_line_limit_counts_only_long_matches(tmp_path: Path) -> None:
|
||||
|
||||
@ -100,6 +100,29 @@ def test_build_sonar_issues_uses_safe_defaults_for_sparse_findings() -> None:
|
||||
assert issue["primaryLocation"]["textRange"] == {"startLine": 1, "endLine": 1}
|
||||
|
||||
|
||||
def test_build_sonar_issues_clamps_inverted_line_ranges() -> None:
|
||||
"""An end line before the start line should collapse to a valid range."""
|
||||
|
||||
issues = semgrep_report.build_sonar_issues(
|
||||
{
|
||||
"results": [
|
||||
{
|
||||
"check_id": "python.lang.correctness.range",
|
||||
"path": "app/main.py",
|
||||
"start": {"line": 9},
|
||||
"end": {"line": 3},
|
||||
"extra": {"severity": "WARNING", "message": "inverted"},
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
assert issues["issues"][0]["primaryLocation"]["textRange"] == {
|
||||
"startLine": 9,
|
||||
"endLine": 9,
|
||||
}
|
||||
|
||||
|
||||
def test_read_json_handles_invalid_json_and_non_object(tmp_path: Path) -> None:
|
||||
"""Report loading should fail closed for bad JSON and wrong top-level shapes."""
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user