94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from pathlib import Path
|
|
import json
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
|
|
from review_agent.models import RegulatoryRuleVersion
|
|
from review_agent.regulatory_review.services.rule_loader import (
|
|
DEFAULT_RULE_CODE,
|
|
check_rule_version,
|
|
compute_file_sha256,
|
|
load_rule_file,
|
|
)
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_load_rule_file_reads_demo_requirements():
|
|
rule_set = load_rule_file()
|
|
|
|
codes = {item["code"] for item in rule_set["requirements"]}
|
|
assert rule_set["code"] == DEFAULT_RULE_CODE
|
|
assert "product_technical_requirements" in codes
|
|
assert "instructions_for_use" in codes
|
|
assert "registration_test_report" in codes
|
|
assert "clinical_evaluation" in codes
|
|
assert "essential_principles_checklist" in codes
|
|
|
|
|
|
def test_load_rule_file_covers_attachment4_outline():
|
|
rule_set = load_rule_file()
|
|
requirements = rule_set["requirements"]
|
|
outline = json.loads(Path("tests/fixtures/regulatory/attachment4_outline.json").read_text(encoding="utf-8"))
|
|
|
|
for chapter in outline:
|
|
chapter_rule = next(
|
|
item for item in requirements if item["title"] == chapter["title"] and item.get("attachment4_code") == chapter["code"]
|
|
)
|
|
assert chapter_rule["attachment4_code"] == chapter["code"]
|
|
assert chapter_rule["severity"] == "high"
|
|
assert chapter_rule["citation_query"]
|
|
for child in chapter["children"]:
|
|
child_rule = next(
|
|
item
|
|
for item in requirements
|
|
if item["title"] == child and str(item.get("attachment4_code", "")).startswith(f"{chapter['code']}.")
|
|
)
|
|
assert child_rule["rule_id"]
|
|
assert child_rule["file_keywords"]
|
|
assert child_rule["severity"] in {"blocking", "high", "medium"}
|
|
assert child_rule["citation_query"]
|
|
|
|
|
|
def test_compute_file_sha256_changes_when_file_changes(tmp_path):
|
|
path = tmp_path / "rule.yaml"
|
|
path.write_text("code: demo\n", encoding="utf-8")
|
|
first = compute_file_sha256(path)
|
|
path.write_text("code: demo2\n", encoding="utf-8")
|
|
|
|
assert compute_file_sha256(path) != first
|
|
|
|
|
|
def test_check_rule_version_creates_missing_db_record():
|
|
result = check_rule_version(update_missing=True)
|
|
|
|
record = RegulatoryRuleVersion.objects.get(code=DEFAULT_RULE_CODE)
|
|
assert result.status == "created"
|
|
assert result.current_hash == record.yaml_hash
|
|
assert record.rag_collection == "nmpa_ivd_registration_v1"
|
|
|
|
|
|
def test_check_rule_version_reports_hash_mismatch_without_overwriting():
|
|
created = check_rule_version(update_missing=True)
|
|
record = RegulatoryRuleVersion.objects.get(code=DEFAULT_RULE_CODE)
|
|
record.yaml_hash = "stale"
|
|
record.save(update_fields=["yaml_hash"])
|
|
|
|
result = check_rule_version(update_missing=False)
|
|
record.refresh_from_db()
|
|
|
|
assert result.status == "mismatch"
|
|
assert result.database_hash == "stale"
|
|
assert result.current_hash == created.current_hash
|
|
assert record.yaml_hash == "stale"
|
|
|
|
|
|
def test_regulatory_rules_check_command_reports_status(capsys):
|
|
call_command("regulatory_rules_check")
|
|
|
|
captured = capsys.readouterr()
|
|
assert DEFAULT_RULE_CODE in captured.out
|
|
assert "created" in captured.out or "ok" in captured.out
|