77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
"""Tests for immutable registry audit refreshes."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from cmdforge.registry.auditing import audit_stale_tools, audit_tool_version
|
|
from cmdforge.registry.db import connect_db, init_db
|
|
|
|
|
|
@pytest.fixture
|
|
def audit_db(tmp_path):
|
|
conn = connect_db(tmp_path / "registry.db")
|
|
init_db(conn)
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT INTO publishers (email, password_hash, slug, display_name)
|
|
VALUES ('audit@example.com', 'x', 'auditor', 'Auditor')
|
|
"""
|
|
)
|
|
publisher_id = cursor.lastrowid
|
|
tool = conn.execute(
|
|
"""
|
|
INSERT INTO tools (
|
|
owner, name, version, config_yaml, publisher_id,
|
|
visibility, moderation_status
|
|
) VALUES ('auditor', 'checked', '1.0.0', ?, ?, 'public', 'approved')
|
|
""",
|
|
[
|
|
"name: checked\nversion: 1.0.0\noutput: stable\n"
|
|
"input_schema: {}\noutput_schema:\n type: string\n",
|
|
publisher_id,
|
|
],
|
|
)
|
|
conn.commit()
|
|
yield conn, tool.lastrowid
|
|
conn.close()
|
|
|
|
|
|
def test_background_audits_append_instead_of_overwriting(audit_db):
|
|
conn, tool_id = audit_db
|
|
first = audit_tool_version(conn, tool_id)
|
|
second = audit_tool_version(conn, tool_id)
|
|
count = conn.execute(
|
|
"SELECT COUNT(*) FROM tool_audits WHERE tool_id = ?", [tool_id]
|
|
).fetchone()[0]
|
|
assert count == 2
|
|
assert first["evidence"]["findings_hash"] == second["evidence"]["findings_hash"]
|
|
assert "generated_tests" in first["evidence"]["findings"]
|
|
assert "dependencies" not in first["evidence"]["checks_run"]
|
|
|
|
|
|
def test_stale_audit_refresh_adds_new_evidence(audit_db):
|
|
conn, tool_id = audit_db
|
|
audit_tool_version(conn, tool_id)
|
|
old = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat()
|
|
conn.execute("UPDATE tool_audits SET evaluated_at = ?", [old])
|
|
conn.commit()
|
|
results = audit_stale_tools(conn, max_age_days=30)
|
|
assert [item["tool_id"] for item in results] == [tool_id]
|
|
assert conn.execute("SELECT COUNT(*) FROM tool_audits").fetchone()[0] == 2
|
|
|
|
|
|
def test_degraded_background_score_is_logged(audit_db):
|
|
conn, tool_id = audit_db
|
|
conn.execute("UPDATE tools SET downloads = 500 WHERE id = ?", [tool_id])
|
|
conn.commit()
|
|
audit_tool_version(conn, tool_id)
|
|
conn.execute("UPDATE tools SET downloads = 0 WHERE id = ?", [tool_id])
|
|
conn.commit()
|
|
result = audit_tool_version(conn, tool_id)
|
|
assert result["quality_change"] < 0
|
|
log = conn.execute(
|
|
"SELECT action FROM audit_log WHERE target_id = ?", [str(tool_id)]
|
|
).fetchone()
|
|
assert log["action"] == "automated_audit_changed"
|