"""Tests for Phase 3 Batch D — post-award, public API v2, gap-report upload, EUR-Lex generator.""" import datetime from unittest.mock import MagicMock, patch import pytest from httpx import ASGITransport, AsyncClient from agents.generator_agent import DocumentGeneratorAgent from core.notifications.inbox import NOTIFICATIONS from core.post_award.service import scan_upcoming_milestones from core.projects.models import Project, ProjectMilestone from core.subscription.db import SessionLocal from core.subscription.models import User from server import app SAMPLE_DOCUMENT = """ # Wniosek o dofinansowanie Horizon Europe ## Opis projektu i innowacji Projekt zakłada rozwój innowacyjnego oprogramowania SaaS na poziomie TRL 6. ## Budżet i kwalifikowalność kosztów - Wynagrodzenia personelu B+R: 120 000 PLN Status MŚP: mikroprzedsiębiorstwo. """ @pytest.fixture def db_project(): db = SessionLocal() user = db.query(User).filter(User.clerk_id == "test_clerk_id_e2e").first() if not user: user = User(clerk_id="test_clerk_id_e2e", tier="pro") db.add(user) db.commit() project = Project( clerk_user_id="test_clerk_id_e2e", title="Post-Award Test", program_type="HORIZON", ) db.add(project) db.commit() db.refresh(project) yield project db.query(ProjectMilestone).filter(ProjectMilestone.project_id == project.id).delete() db.delete(project) db.commit() db.close() @pytest.mark.asyncio async def test_gap_report_upload_txt(auth_headers): content = SAMPLE_DOCUMENT.encode("utf-8") async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: resp = await client.post( "/api/audit/gap-report/upload", headers=auth_headers, files={"file": ("wniosek.txt", content, "text/plain")}, data={"program_name": "HORIZON", "export_format": "json"}, ) assert resp.status_code == 200 body = resp.json() assert body["program"] == "HORIZON" assert "overall_score" in body @pytest.mark.asyncio async def test_post_award_milestones_crud(auth_headers, db_project): project_id = db_project.id due = (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=7)).isoformat() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: create = await client.post( f"/api/projects/{project_id}/post-award/milestones", headers=auth_headers, json={"title": "Raport M1", "due_date": due, "description": "Pierwszy raport"}, ) assert create.status_code == 200 mid = create.json()["id"] listing = await client.get( f"/api/projects/{project_id}/post-award/milestones", headers=auth_headers, ) assert listing.status_code == 200 assert listing.json()["count"] == 1 patch = await client.patch( f"/api/projects/{project_id}/post-award/milestones/{mid}", headers=auth_headers, json={"status": "completed"}, ) assert patch.status_code == 200 assert patch.json()["status"] == "completed" delete = await client.delete( f"/api/projects/{project_id}/post-award/milestones/{mid}", headers=auth_headers, ) assert delete.status_code == 200 def test_scan_upcoming_milestones_creates_notification(db_project): NOTIFICATIONS.clear() db = SessionLocal() try: due = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=3) row = ProjectMilestone( project_id=db_project.id, clerk_user_id="test_clerk_id_e2e", title="Raport okresowy", due_date=due, status="pending", ) db.add(row) db.commit() created = scan_upcoming_milestones(db) assert created >= 1 assert any(n["type"] == "post_award_deadline" for n in NOTIFICATIONS) finally: db.query(ProjectMilestone).filter(ProjectMilestone.project_id == db_project.id).delete() db.commit() db.close() NOTIFICATIONS.clear() @pytest.mark.asyncio async def test_public_api_v2_key_and_grants(auth_headers): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: issue = await client.post( "/api/v2/developer/keys", headers=auth_headers, json={"label": "test-integration"}, ) assert issue.status_code == 200 raw_key = issue.json()["api_key"] assert raw_key.startswith("gf_") openapi = await client.get("/api/v2/public/openapi.json") assert openapi.status_code == 200 assert openapi.json()["info"]["version"] == "2.0.0" health = await client.get( "/api/v2/public/health", headers={"X-API-Key": raw_key}, ) assert health.status_code == 200 assert health.json()["status"] == "ok" grants = await client.get( "/api/v2/public/grants/nabory", headers={"X-API-Key": raw_key}, params={"limit": 5}, ) assert grants.status_code == 200 assert "nabory" in grants.json() trust = await client.get( "/api/v2/public/trust/summary", headers={"X-API-Key": raw_key}, ) assert trust.status_code == 200 body = trust.json() assert "platform_score" in body assert "level" in body regional = await client.get( "/api/v2/public/regional/programs", headers={"X-API-Key": raw_key}, params={"limit": 10}, ) assert regional.status_code == 200 reg_body = regional.json() assert "programs" in reg_body assert "voivodeships" in reg_body assert "bip_parser_voivodeships" in reg_body paths = openapi.json()["paths"] assert "/trust/summary" in paths assert "/regional/programs" in paths @patch("integrations.eurlex_client.EURLexClient") def test_generator_eurlex_boost_for_horizon(mock_client_cls): mock_client = MagicMock() mock_client.search_legal_acts.return_value = [ { "title": "Horizon Europe Regulation", "celex": "32021R0106", "url": "https://eur-lex.europa.eu/legal-content/PL/TXT/?uri=CELEX:32021R0106", } ] mock_client_cls.return_value = mock_client agent = DocumentGeneratorAgent() state = { "document_type": "Horizon Europe", "external_context": { "program_type": "HORIZON", "program_name": "Horizon Europe", "celex": "32021R0106", }, } boost = agent._build_eurlex_boost(state) assert "[LIVE EUR-LEX" in boost assert "32021R0106" in boost mock_client.search_legal_acts.assert_called() @patch("integrations.eurlex_client.EURLexClient") def test_generator_eurlex_boost_skips_without_legal_id(mock_client_cls): mock_client = MagicMock() mock_client_cls.return_value = mock_client agent = DocumentGeneratorAgent() state = { "document_type": "Horizon Europe", "external_context": {"program_type": "HORIZON", "program_name": "Horizon Europe"}, } boost = agent._build_eurlex_boost(state) assert boost == "" mock_client.search_legal_acts.assert_not_called()