mi
88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import copy
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from jsonschema import ValidationError
|
|
|
|
from app.config import validate_config
|
|
from app.db import Base
|
|
|
|
|
|
def seed(artifacts: Path):
|
|
return yaml.safe_load((artifacts / "seed-config.yaml").read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_seed_config_and_artifact_hashes(artifacts: Path) -> None:
|
|
rules, detector, digest = validate_config(seed(artifacts), artifacts)
|
|
assert rules.version == "2026-01-01"
|
|
assert detector.version.startswith("sha256:")
|
|
assert len(digest) == 32
|
|
|
|
|
|
def test_config_cross_field_and_manifest_subset(artifacts: Path) -> None:
|
|
bad = copy.deepcopy(seed(artifacts))
|
|
bad["task"]["heartbeat_sec"] = bad["task"]["lease_sec"]
|
|
with pytest.raises(ValueError):
|
|
validate_config(bad, artifacts)
|
|
bad = copy.deepcopy(seed(artifacts))
|
|
bad["file_policy"]["enabled_mime_types"].append("application/zip")
|
|
with pytest.raises(ValueError):
|
|
validate_config(bad, artifacts)
|
|
|
|
|
|
def test_clamav_signature_age_policy_bounds(artifacts: Path) -> None:
|
|
document = seed(artifacts)
|
|
assert document["clamav"]["max_signature_age_hours"] == 240
|
|
document["clamav"]["max_signature_age_hours"] = 720
|
|
validate_config(document, artifacts)
|
|
document["clamav"]["max_signature_age_hours"] = 721
|
|
with pytest.raises(ValidationError):
|
|
validate_config(document, artifacts)
|
|
|
|
|
|
def test_normative_tables_are_in_service_schema() -> None:
|
|
expected = {
|
|
"safety_requests",
|
|
"safety_tasks",
|
|
"file_verdict_cache",
|
|
"text_rules_cache",
|
|
"link_verdict_cache",
|
|
"safety_audit",
|
|
"config_versions",
|
|
}
|
|
assert expected <= {table.name for table in Base.metadata.tables.values()}
|
|
assert {table.schema for table in Base.metadata.tables.values()} == {"message_safety"}
|
|
|
|
|
|
def test_migration_executes_asyncpg_statements_separately() -> None:
|
|
migration = (
|
|
Path(__file__).parents[1]
|
|
/ "alembic"
|
|
/ "versions"
|
|
/ "0001_message_safety_v2.py"
|
|
)
|
|
tree = ast.parse(migration.read_text(encoding="utf-8"))
|
|
upgrade = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef) and node.name == "upgrade"
|
|
)
|
|
statements = [
|
|
call.args[0].value
|
|
for call in ast.walk(upgrade)
|
|
if isinstance(call, ast.Call)
|
|
and isinstance(call.func, ast.Attribute)
|
|
and call.func.attr == "execute"
|
|
and call.args
|
|
and isinstance(call.args[0], ast.Constant)
|
|
and isinstance(call.args[0].value, str)
|
|
]
|
|
|
|
assert len(statements) == 5
|
|
assert all(re.search(r"\$\$;\s+\S", statement) is None for statement in statements)
|