Spaces:
Sleeping
Sleeping
File size: 5,473 Bytes
ce8f04a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | """Tests for Batch C — snapshot gap audit, regulation schemas, tenant audit log."""
import pytest
from httpx import ASGITransport, AsyncClient
from core.audit.snapshot_gap_audit import run_snapshot_gap_audit
from core.regulation.schema_registry import get_schema, list_schema_ids
from core.search.regulation_engine import regulation_engine
from server import app
SAMPLE_DOCUMENT = """
# Wniosek o dofinansowanie
## Opis projektu i innowacji
Projekt zakłada rozwój innowacyjnego oprogramowania SaaS na poziomie TRL 6.
Komponent B+R obejmuje badania nad modelem predykcyjnym.
## Harmonogram i kamienie milowe
Q1: prototyp, Q2: pilotaż, Q3: wdrożenie komercyjne.
## Budżet i kwalifikowalność kosztów
- Wynagrodzenia personelu B+R: 120 000 PLN
- Oprogramowanie i licencje: 30 000 PLN
- Szkolenia: 10 000 PLN
## Analiza rynku i wdrożenie
Rynek MŚP w Polsce, plan sprzedaży subskrypcji.
Status MŚP: mikroprzedsiębiorstwo zgodnie z 651/2014.
"""
def test_snapshot_gap_audit_detects_complete_smart_document():
rules = regulation_engine.get_structured_rules_for_program("SMART")
report = run_snapshot_gap_audit(
SAMPLE_DOCUMENT,
"SMART",
structured_rules=rules,
company_profile={"size": "mikro", "nip": "5250000000"},
)
assert report.overall_score >= 70
assert not report.has_critical_gaps
assert report.innovation_status.get("adequate") is True
assert report.sme_status.get("eligible_signal") is True
def test_snapshot_gap_audit_flags_missing_sections_and_ineligible_costs():
bad_doc = """
## Budżet
Zakup samochodu osobowego dla prezesa: 80 000 PLN
Marketing i reklama: 50 000 PLN
Status MŚP: mikroprzedsiębiorstwo.
"""
report = run_snapshot_gap_audit(
bad_doc,
"SMART",
company_profile={"size": "duże"},
)
assert report.overall_score < 80
assert any(g.category == "costs" for g in report.gaps)
assert any(g.category == "sections" for g in report.gaps)
assert any(g.code == "SME_STATUS_CONFLICT" for g in report.gaps)
def test_snapshot_gap_audit_rejects_short_document():
report = run_snapshot_gap_audit("krótki", "PARP")
assert report.overall_score == 0
assert report.has_critical_gaps
def test_regulation_schema_registry():
ids = list_schema_ids()
assert "program_rules" in ids
assert "gap_audit_report" in ids
schema = get_schema("gap_audit_report")
assert schema and schema["type"] == "object"
@pytest.mark.asyncio
async def test_api_regulation_schemas(auth_headers):
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
resp = await client.get("/api/regulation/schemas", headers=auth_headers)
assert resp.status_code == 200
data = resp.json()
assert "program_rules" in data["schema_ids"]
assert "schemas" in data
@pytest.mark.asyncio
async def test_api_gap_report_json(auth_headers):
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
resp = await client.post(
"/api/audit/gap-report",
headers=auth_headers,
json={
"program_name": "SMART",
"document_text": SAMPLE_DOCUMENT,
"export_format": "json",
"company_profile": {"size": "mikro"},
},
)
assert resp.status_code == 200
body = resp.json()
assert body["program"] == "SMART"
assert "overall_score" in body
assert "gaps" in body
@pytest.mark.asyncio
async def test_api_audit_logs_after_gap_report(auth_headers):
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
await client.post(
"/api/audit/gap-report",
headers=auth_headers,
json={
"program_name": "PARP",
"document_text": SAMPLE_DOCUMENT,
"export_format": "json",
},
)
logs = await client.get("/api/audit-logs", headers=auth_headers)
assert logs.status_code == 200
items = logs.json()["items"]
assert any(i["action"] == "gap_audit_completed" for i in items)
@pytest.mark.asyncio
async def test_collaborator_add_remove_creates_audit_logs(auth_headers):
from core.projects.models import Project
from core.subscription.db import SessionLocal
db = SessionLocal()
try:
project = Project(
clerk_user_id="test_clerk_id_e2e",
title="Audit Collab Test",
program_type="SMART",
)
db.add(project)
db.commit()
db.refresh(project)
project_id = project.id
finally:
db.close()
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
add = await client.post(
f"/api/audit/collaborators/{project_id}",
headers=auth_headers,
json={"collaborator_user_id": "collab_user_99", "role": "viewer"},
)
assert add.status_code == 200
remove = await client.delete(
f"/api/audit/collaborators/{project_id}/collab_user_99",
headers=auth_headers,
)
assert remove.status_code == 200
logs = await client.get("/api/audit-logs", headers=auth_headers)
actions = [i["action"] for i in logs.json()["items"]]
assert "collaborator_added" in actions
assert "collaborator_removed" in actions
|