""" Grant Pulse (phases A–D): classification, job worker, EUR-Lex batch, firm pulse, API. """ from __future__ import annotations import os from datetime import date, datetime, timedelta, timezone from unittest.mock import patch import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool @pytest.fixture() def db_session(): from core.subscription.db import Base import core.grants.models # noqa: F401 import core.projects.models # noqa: F401 import core.subscription.models # noqa: F401 engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(bind=engine) Session = sessionmaker(bind=engine) session = Session() try: yield session finally: session.close() Base.metadata.drop_all(bind=engine) def _seed_grants(session): from core.grants.models import Grant today = date.today() rows = [ Grant( id="pulse-new-1", source_id="pulse-new-1", name="Nowy SMART cyfryzacja", program="PARP", status="active", source="wyszukiwarka:PARP", deadline=(today + timedelta(days=60)).isoformat(), operator="PARP", description="Cyfryzacja MŚP PKD 62.01", regulation_url="https://www.parp.gov.pl/reg.pdf", official_page_url="https://www.parp.gov.pl/smart", eligible_company_sizes=["mikro", "małe", "średnie"], eligible_regions=["cała Polska"], eligible_pkd=["62.01"], source_credibility_score=0.9, catalog_hidden=False, raw_data={ "first_seen_at": datetime.now(timezone.utc).isoformat(), "fetched_at": datetime.now(timezone.utc).isoformat(), "regulation_grounded": True, "podstawa_prawna": "CELEX:32021R1058", }, ), Grant( id="pulse-close-1", source_id="pulse-close-1", name="Zamykający się nabór OZE", program="NFOŚiGW", status="active", source="wyszukiwarka:NFOSiGW", deadline=(today + timedelta(days=7)).isoformat(), operator="NFOŚiGW", description="OZE dla firm", regulation_url="https://www.gov.pl/nfosigw/reg.pdf", eligible_company_sizes=["mikro", "małe"], eligible_regions=["cała Polska"], eligible_pkd=[], source_credibility_score=0.85, catalog_hidden=False, raw_data={"fetched_at": (datetime.now(timezone.utc) - timedelta(days=40)).isoformat()}, ), Grant( id="pulse-plan-1", source_id="pulse-plan-1", name="Planowany Horizon Europe call", program="Horizon Europe", status="planned", source="wyszukiwarka:KPK", deadline=(today + timedelta(days=120)).isoformat(), operator="EISMEA", description="Digital innovation planned call", regulation_url="", source_credibility_score=0.7, catalog_hidden=False, raw_data={ "link_eurlex": "https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX:32021R0695", }, ), Grant( id="pulse-old-1", source_id="pulse-old-1", name="Stary zamknięty", program="PARP", status="closed", source="wyszukiwarka:PARP", deadline=(today - timedelta(days=30)).isoformat(), operator="PARP", source_credibility_score=0.5, catalog_hidden=False, raw_data={}, ), ] for r in rows: session.add(r) session.commit() return rows # ── Classification ─────────────────────────────────────────────────────────── def test_classify_new_planned_closing(): from core.grants.pulse import classify_pulse_kinds, list_pulse_grants today = date.today() grants = [ { "id": "a", "name": "New", "status": "active", "deadline": (today + timedelta(days=40)).isoformat(), "first_seen_at": today.isoformat(), "regulation_url": "https://www.parp.gov.pl/r.pdf", "source_credibility_score": 0.9, }, { "id": "b", "name": "Closing soon", "status": "active", "deadline": (today + timedelta(days=5)).isoformat(), "source_credibility_score": 0.8, "regulation_url": "https://x.gov.pl/r.pdf", }, { "id": "c", "name": "Planowany nabór", "status": "planned", "deadline": (today + timedelta(days=90)).isoformat(), }, ] kinds_a = classify_pulse_kinds(grants[0], today=today) assert "new" in kinds_a assert "current" in kinds_a assert "closing" in classify_pulse_kinds(grants[1], today=today) assert "planned" in classify_pulse_kinds(grants[2], today=today) for kind in ("new", "planned", "closing", "current"): out = list_pulse_grants(grants, kind=kind, limit=10, today=today) assert out["status"] == "ok" assert isinstance(out["items"], list) # each kind should have at least one for our fixture set if kind != "all": assert out["count"] >= 1, kind for it in out["items"]: assert kind in it["pulse_kinds"] or kind == "current" assert "verification" in it def test_verification_flags_alert_worthy(): from core.grants.pulse import verification_flags today = date.today() good = verification_flags( { "status": "active", "deadline": (today + timedelta(days=20)).isoformat(), "regulation_url": "https://www.parp.gov.pl/r.pdf", "source_credibility_score": 0.9, "regulation_grounded": True, }, today=today, ) assert good["deadline_verified"] assert good["status_verified"] assert good["regulation_verified"] assert good["alert_worthy"] bad = verification_flags( {"status": "closed", "deadline": (today - timedelta(days=1)).isoformat()}, today=today, ) assert not bad["alert_worthy"] # ── Job worker ─────────────────────────────────────────────────────────────── def test_research_job_cycle_offline(db_session): from core.grants.models import ResearchJob from core.grants.job_worker import enqueue_job, process_pending_jobs_cycle from core.grants.pulse import reset_metrics reset_metrics() _seed_grants(db_session) jid = enqueue_job( db_session, "seed_credibility", target_id="pulse-new-1", payload={"action": "test"}, priority=8, ) jid2 = enqueue_job( db_session, "verify_claims", target_id="pulse-close-1", priority=7, ) jid3 = enqueue_job( db_session, "discover_source", payload={"source": "test", "urls": ["https://example.com"]}, priority=3, ) db_session.commit() stats = process_pending_jobs_cycle(db_session, max_jobs=10) assert stats["enabled"] is True assert stats["processed"] >= 3 assert stats["failed"] == 0 for jid in (jid, jid2, jid3): job = db_session.query(ResearchJob).filter(ResearchJob.id == jid).first() assert job is not None assert job.status == "completed" assert job.result is not None def test_eurlex_job_grounds_with_legal_id(db_session): from core.grants.job_worker import enqueue_job, process_one_job from core.grants.models import ResearchJob, Grant _seed_grants(db_session) jid = enqueue_job( db_session, "eurlex_ground", target_id="pulse-new-1", payload={"network": False}, priority=9, ) db_session.commit() job = db_session.query(ResearchJob).filter(ResearchJob.id == jid).first() out = process_one_job(db_session, job) db_session.commit() assert out["status"] == "completed" row = db_session.query(Grant).filter(Grant.source_id == "pulse-new-1").first() raw = row.raw_data or {} assert raw.get("eurlex_grounded") is True assert raw.get("celex_id") # ── EUR-Lex batch ──────────────────────────────────────────────────────────── def test_eurlex_sanitize_rejects_garbage_accepts_celex(): from integrations.eurlex_client import is_valid_eurlex_query, _sanitize_search_query assert is_valid_eurlex_query("32021R1058") assert is_valid_eurlex_query("CELEX:32021R0695") assert _sanitize_search_query("PARP — Harmonogram naborów innowacje MŚP") == "" assert _sanitize_search_query("any HORIZON EUROPE award criteria") == "" assert "32021R1058" in _sanitize_search_query("zgodnie z CELEX 32021R1058") def test_eurlex_batch_skip_without_id_ground_with_id(): from core.grants.eurlex_batch import batch_ground_grants, extract_legal_ids_from_grant grants = [ {"id": "g1", "name": "No legal", "description": "program PARP SMART"}, { "id": "g2", "name": "With CELEX", "podstawa_prawna": "Rozporządzenie CELEX:32021R1058", "description": "fundusze", }, ] assert extract_legal_ids_from_grant(grants[0]) == [] assert extract_legal_ids_from_grant(grants[1]) # mock search — must only be called with legal id called = [] def fake_search(q: str): called.append(q) assert is_valid_like(q) return [{"title": "Reg", "celex": q, "url": f"https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX:{q}"}] def is_valid_like(q: str) -> bool: from integrations.eurlex_client import is_valid_eurlex_query return is_valid_eurlex_query(q) result = batch_ground_grants(grants, search_fn=fake_search, network=False, limit=10) assert result["needs_verification"] >= 1 assert result["grounded"] >= 1 grounded = [g for g in result["grants"] if g.get("eurlex_grounded")] assert grounded assert grounded[0]["celex_id"] # free-text grant never passed to search for q in called: assert "PARP" not in q assert "SMART" not in q or q.startswith("3") def test_eurlex_batch_does_not_call_with_program_title(): from core.grants.eurlex_batch import ground_grant_with_legal_id from integrations.eurlex_client import _sanitize_search_query calls = [] def boom(q): calls.append(q) raise AssertionError("should not be called for garbage") g = ground_grant_with_legal_id( {"name": "PARP SMART"}, "PARP — Harmonogram naborów", search_fn=boom, network=False, ) assert g.get("eurlex_grounded") is False assert calls == [] assert _sanitize_search_query("PARP — Harmonogram naborów") == "" # ── Firm pulse ─────────────────────────────────────────────────────────────── def test_firm_pulse_ranking(db_session): from core.grants.firm_pulse import rank_pulse_for_company from core.grants.completeness import grant_dict_from_row _seed_grants(db_session) grants = [grant_dict_from_row(r) for r in db_session.query(__import__("core.grants.models", fromlist=["Grant"]).Grant).all()] company = { "name": "Demo Sp. z o.o.", "size": "mikro", "voivodeship": "mazowieckie", "region": "mazowieckie", "pkd": ["62.01.Z"], "entity_type": "przedsiębiorca", } out = rank_pulse_for_company(grants, company, kind="current", limit=10, min_score=20) assert out["status"] == "ok" assert "items" in out # ranked items expose match + verification for it in out["items"]: assert "match_score" in it assert "verification" in it assert "alert_worthy" in it # ── API TestClient ─────────────────────────────────────────────────────────── @pytest.fixture() def pulse_client(db_session, monkeypatch): monkeypatch.setenv("ENV", "test") monkeypatch.setenv("ALLOW_DEV_TOKEN", "true") monkeypatch.setenv("ENABLE_LIVE_RESEARCH", "true") monkeypatch.setenv("ENABLE_PULSE_WORKER", "true") monkeypatch.setenv("ENABLE_EURLEX_BATCH", "true") monkeypatch.setenv("ENABLE_EURLEX_NETWORK", "false") _seed_grants(db_session) from core.subscription import auth_utils # ensure dev token works monkeypatch.setenv("ENV", "dev") from fastapi.testclient import TestClient import server def _override_db(): try: yield db_session finally: pass from endpoints.projects import get_db as projects_get_db from endpoints import grants as grants_ep server.app.dependency_overrides[projects_get_db] = _override_db # also override if grants imported get_db by reference try: from endpoints.projects import get_db server.app.dependency_overrides[get_db] = _override_db except Exception: pass client = TestClient(server.app) yield client server.app.dependency_overrides.clear() def test_api_pulse_kinds(pulse_client): headers = {"Authorization": "Bearer dev_test_token"} samples = {} for kind in ("new", "planned", "closing", "current"): r = pulse_client.get(f"/api/grants/pulse?kind={kind}&limit=20", headers=headers) assert r.status_code == 200, (kind, r.text[:300]) data = r.json() assert data.get("status") == "ok" assert "items" in data assert isinstance(data["items"], list) samples[kind] = {"count": data.get("count"), "sample": (data["items"] or [None])[0]} # empty kind should not 500 r = pulse_client.get("/api/grants/pulse?kind=all&limit=5", headers=headers) assert r.status_code == 200 # Real auth gate (conftest autouse mocks verify_token — pop for this assertion) import server as server_mod from core.subscription.middleware import verify_token as real_verify server_mod.app.dependency_overrides.pop(real_verify, None) r2 = pulse_client.get( "/api/grants/pulse?kind=new", headers={"Authorization": "Bearer invalid_token_xyz"}, ) assert r2.status_code in (401, 403), r2.text[:200] # restore mock for subsequent tests async def mock_verify_token(): return {"sub": "test_clerk_id_e2e"} server_mod.app.dependency_overrides[real_verify] = mock_verify_token # samples captured for evidence in scratch when run via verification script assert samples def test_api_firm_pulse_and_worker(pulse_client): headers = {"Authorization": "Bearer dev_test_token"} r = pulse_client.post( "/api/grants/pulse/firm", headers=headers, json={ "description": "Cyfryzacja ERP oprogramowanie MŚP", "kind": "current", "limit": 10, "company_size": "mikro", "voivodeship": "mazowieckie", "pkd": ["62.01.Z"], "min_score": 10, }, ) assert r.status_code == 200, r.text[:400] data = r.json() assert data.get("status") == "ok" assert "items" in data r2 = pulse_client.post( "/api/grants/pulse/worker?max_jobs=5&include_eurlex=true", headers=headers, ) assert r2.status_code == 200, r2.text[:400] body = r2.json() assert body.get("status") == "ok" assert "stages" in body r3 = pulse_client.get("/api/grants/pulse/metrics", headers=headers) assert r3.status_code == 200 m = r3.json() assert "metrics" in m assert "flags" in m def test_law_watchlist_extended(): from core.monitoring.law_watchlist import load_law_watchlist, check_content_hashes wl = load_law_watchlist() assert len(wl) >= 5 # more than original 3 programs = {e["program"] for e in wl} assert "FENG" in programs assert "GBER" in programs or "DE_MINIMIS" in programs first = check_content_hashes({"FENG": "content v1", "NCBR": "other"}) second = check_content_hashes( {"FENG": "content v2 changed", "NCBR": "other"}, previous=first["hashes"], ) assert second["changes"] >= 1 def test_pulse_deadline_backfill_and_past_not_current(): """list_pulse_grants extracts deadline from description and closes past actives.""" from core.grants.pulse import list_pulse_grants, classify_pulse_kinds today = date.today() grants = [ { "id": "text-dl", "name": "Program z terminem w opisie", "status": "active", "deadline": "", "description": "Składanie wniosków do 15.11.2027 w systemie LSI.", "regulation_url": "https://www.parp.gov.pl/reg.pdf", "source_credibility_score": 0.8, "first_seen_at": today.isoformat(), }, { "id": "past-active", "name": "Przeterminowany nadal active", "status": "active", "deadline": (today - timedelta(days=10)).isoformat(), "description": "Stary nabór", "source_credibility_score": 0.5, }, ] out = list_pulse_grants(grants, kind="all", limit=20, today=today) assert out["status"] == "ok" assert "quality" in out assert out["quality"]["with_deadline"] >= 1 assert out["quality"]["deadline_backfilled"] >= 1 by_id = {str(i["id"]): i for i in out["items"]} # text deadline backfilled → visible as current/new, not empty assert "text-dl" in by_id assert by_id["text-dl"]["deadline"] == "2027-11-15" # past deadline must not appear as current past_kinds = classify_pulse_kinds( { "id": "past-active", "status": "closed", "deadline": (today - timedelta(days=10)).isoformat(), }, today=today, ) assert "current" not in past_kinds # item for past either absent from current filter or marked closed current = list_pulse_grants(grants, kind="current", limit=20, today=today) for it in current["items"]: if it["id"] == "past-active": assert it["status"] == "closed" or "current" not in it["pulse_kinds"] def test_backfill_deadline_job_persists(db_session): from core.grants.models import Grant, ResearchJob from core.grants.job_worker import enqueue_job, process_one_job g = Grant( id="bf-1", source_id="bf-1", name="Backfill me", program="PARP", status="active", source="test", deadline="", description="Termin naboru: do 30.09.2027. Portal LSI.", regulation_url="https://www.parp.gov.pl/x.pdf", catalog_hidden=False, raw_data={}, ) db_session.add(g) db_session.commit() jid = enqueue_job(db_session, "backfill_deadline", target_id="bf-1", priority=9) db_session.commit() job = db_session.query(ResearchJob).filter(ResearchJob.id == jid).first() out = process_one_job(db_session, job) db_session.commit() assert out["status"] == "completed" assert out.get("updated") is True row = db_session.query(Grant).filter(Grant.source_id == "bf-1").first() assert row.deadline == "2027-09-30" assert (row.raw_data or {}).get("deadline_source") def test_pulse_cycle_graceful_on_detect_error(db_session, monkeypatch): """Worker cycle must not raise when detect/scrape path fails.""" from core.grants import job_worker monkeypatch.setenv("ENABLE_PULSE_WORKER", "true") monkeypatch.setenv("ENABLE_LIVE_RESEARCH", "true") monkeypatch.setenv("ENABLE_EURLEX_BATCH", "true") monkeypatch.setenv("ENABLE_DEADLINE_BACKFILL", "true") def boom(*a, **k): raise RuntimeError("scrape timeout simulated") monkeypatch.setattr( "core.grants.live_research.detect_grant_changes", boom, raising=False ) # also patch import path used inside run_pulse_cycle import core.grants.live_research as lr monkeypatch.setattr(lr, "detect_grant_changes", boom) monkeypatch.setattr(lr, "is_live_research_enabled", lambda: True) result = job_worker.run_pulse_cycle( db_session, max_jobs=2, include_eurlex_batch=True, network=False ) assert result["status"] == "ok" assert "stages" in result # detect error captured, process continues assert result["stages"].get("detect", {}).get("graceful") is True or "error" in result[ "stages" ].get("detect", {})