Spaces:
Sleeping
Sleeping
File size: 7,510 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 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 | """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()
|