backend/tests/test_hgis.py

"""Testy Hybrid Grant Intelligence System (bez mocków jednostek rdzenia)."""
from __future__ import annotations

import json
import os
import tempfile
from pathlib import Path

import pytest

from core.grants.completeness import compute_completeness
from core.grants.credibility import score_source_credibility
from core.grants.versioning import detect_field_changes
from core.grants.recommendations import score_grant_for_company
from core.grants.catalog_service import _matches_filter, filter_catalog_rows, resolve_catalog_plan
import importlib.util

_cd_path = Path(__file__).resolve().parents[1] / "rag_pipeline" / "change_detector.py"
_spec = importlib.util.spec_from_file_location("change_detector", _cd_path)
_cd_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_cd_mod)
ChangeDetector = _cd_mod.ChangeDetector

FIXTURE_PATH = Path(__file__).resolve().parent / "fixtures" / "real_grant_sample.json"


def _load_real_grant() -> dict:
    assert FIXTURE_PATH.exists(), f"missing fixture: {FIXTURE_PATH}"
    return json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))


REAL_GRANT = _load_real_grant() if FIXTURE_PATH.exists() else None

SAMPLE_GRANT = {
    "id": "test-001",
    "name": "Fundusze Europejskie dla MŚP 2025",
    "description": "Wsparcie inwestycji MŚP w Mazowieckiem.",
    "status": "active",
    "operator": "PARP",
    "program_year": 2025,
    "eligible_regions": ["mazowieckie", "cała polska"],
    "eligible_company_sizes": ["mikro", "małe"],
    "eligible_beneficiaries": ["mśp", "przedsiębiorca"],
    "eligible_pkd": ["62.01", "62.02"],
    "regulation_url": "https://www.parp.gov.pl/regulamin.pdf",
    "official_page_url": "https://www.parp.gov.pl/",
    "application_url": "https://www.parp.gov.pl/wniosek",
    "deadline": "2026-12-31",
    "budget_max_pln": 500000,
    "instrument_type": "grant",
    "poziom_weryfikacji": "regulamin",
}


def test_real_grant_fixture_loaded():
    assert REAL_GRANT is not None
    assert REAL_GRANT.get("name")
    assert str(REAL_GRANT.get("regulation_url") or "").startswith("http")
    assert REAL_GRANT.get("official_page_url") or REAL_GRANT.get("url")


def test_completeness_high_for_real_grant():
    result = compute_completeness(REAL_GRANT)
    assert result["completeness_score"] >= 70
    assert "regulation_url" in result["fields_present"]


def test_credibility_real_parp_source():
    score = score_source_credibility(REAL_GRANT)
    assert score >= 0.75


def test_recommendation_scoring_real_grant():
    company = {"size": "mikro", "region": "mazowieckie", "pkd": ["62.01"], "entity_type": "startup"}
    rec = score_grant_for_company(REAL_GRANT, company)
    assert rec["relevance_score"] >= 40
    assert len(rec.get("why_it_fits", "")) > 10


def test_versioning_detects_deadline_change():
    old = {**SAMPLE_GRANT, "deadline": "2025-06-30"}
    new = {**SAMPLE_GRANT, "deadline": "2026-12-31"}
    changes = detect_field_changes("test-001", old, new)
    assert any(c["field_name"] == "deadline" for c in changes)


def test_pkd_filter():
    item = {**SAMPLE_GRANT}
    assert _matches_filter(item, {"pkd": "62.01"})
    assert not _matches_filter(item, {"pkd": "01.11"})


def test_change_detector_incremental():
    with tempfile.TemporaryDirectory() as tmp:
        store = Path(tmp) / "hashes.json"
        det = ChangeDetector(store_path=str(store))
        assert det.record("url1", "content A")["is_new"] is True
        assert det.record("url1", "content A")["changed"] is False
        assert det.record("url1", "content B")["changed"] is True


def test_hybrid_search_fts_only_without_pinecone():
    """Bez PINECONE_API_KEY wyszukiwanie działa w trybie fts_only (Pinecone wymaga klucza)."""
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from core.subscription.db import Base, init_models
    from core.grants.models import Grant
    from core.grants.hybrid_search import hybrid_search_catalog

    old_key = os.environ.pop("PINECONE_API_KEY", None)
    try:
        init_models()
        engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
        Base.metadata.create_all(bind=engine)
        Session = sessionmaker(bind=engine)
        db = Session()
        db.add(
            Grant(
                source_id="hyb-1",
                name="Dotacja MŚP innowacje",
                status="active",
                source="test",
                description="Wsparcie dla mikro i małych firm w Mazowieckiem.",
                eligible_company_sizes=["mikro"],
                eligible_regions=["mazowieckie"],
            )
        )
        db.commit()
        results = hybrid_search_catalog(db, query="dotacja MŚP", filters={"status": "active"}, limit=5)
        assert results
        assert results[0].get("search_mode") in ("fts_only", "fts+pinecone")
        db.close()
    finally:
        if old_key:
            os.environ["PINECONE_API_KEY"] = old_key


def test_catalog_hidden_filter():
    items = [
        {**SAMPLE_GRANT, "catalog_hidden": True},
        {**SAMPLE_GRANT, "id": "visible", "catalog_hidden": False},
    ]
    plan = resolve_catalog_plan({})
    out = filter_catalog_rows(items, "", plan.post_filters, plan, hide_stale=False)
    assert len(out) == 1
    assert out[0]["id"] == "visible"


@pytest.mark.skipif(
    os.environ.get("ENABLE_GRANT_RESEARCH_AGENT", "true").lower() == "false",
    reason="agent disabled",
)
def test_grant_research_agent_on_real_grant():
    from agents.grant_research_agent import run_grant_research_agent

    changes = detect_field_changes(
        REAL_GRANT["id"],
        {**REAL_GRANT, "deadline": "2025-01-01"},
        REAL_GRANT,
    )
    company = {"size": "mikro", "region": "mazowieckie", "pkd": ["62.01"]}
    result = run_grant_research_agent(REAL_GRANT, changes=changes, company_profile=company)
    assert result.get("enabled") is True
    assert result.get("change_summary")
    assert result.get("enriched_metadata", {}).get("completeness_score", 0) >= 70
    assert len(result.get("recommendations", [])) >= 1


@pytest.mark.skipif(
    os.environ.get("PYTEST_USE_VERIFY_DB", "0").lower() not in ("1", "true", "yes"),
    reason="requires shared verify_hgis.db",
)
def test_verify_db_imported_catalog():
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from core.subscription.db import init_models
    from core.grants.models import Grant
    from core.grants.regulation_backfill import regulation_completeness_report

    db_url = os.environ.get("DATABASE_URL")
    assert db_url and "verify_hgis" in db_url
    init_models()
    engine = create_engine(db_url, connect_args={"check_same_thread": False})
    Session = sessionmaker(bind=engine)
    db = Session()
    try:
        count = db.query(Grant).count()
        assert count > 1000
        parp = db.query(Grant).filter(Grant.source.like("%PARP%")).first()
        assert parp is not None
        report = regulation_completeness_report(db)
        assert report["with_regulation_url"] > 200
    finally:
        db.close()