hermes: lift remaining tracked modules to the branch floor

Exercise the mailu sync retry, attribute, and skip branches, the
listener non-object JSON path, and the hygiene conftest skip; drop the
unreachable inverted-range clamp in the semgrep report (the line helper
already floors the end line) and pin that behavior with a test. Exclude
the mailu __main__ guards from measurement.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jenkins 2026-08-17 15:10:17 -03:00
parent 3a57371989
commit 64272f52d2
7 changed files with 142 additions and 5 deletions

View File

@ -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"),

View File

@ -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

View File

@ -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)

View File

@ -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:

View File

@ -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()

View File

@ -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:

View File

@ -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."""