backend/tests/test_search_catalog.py

"""Integracyjne testy search_catalog na realnym JSON (SQLite in-memory)."""
from __future__ import annotations

import os

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")

from core.subscription.db import Base, init_models  # noqa: E402
from core.grants.wyszukiwarka_import import resolve_wyszukiwarka_path, import_wyszukiwarka_file  # noqa: E402
from core.grants.catalog_service import (  # noqa: E402
    search_catalog,
    _token_in_hay,
    _matches_filter,
    _effective_company_sizes,
)
from core.grants.catalog_filters import is_catalog_visible  # noqa: E402

_catalog_engine = create_engine(
    "sqlite:///:memory:",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
CatalogSession = sessionmaker(autocommit=False, autoflush=False, bind=_catalog_engine)


@pytest.fixture(scope="module")
def db_session():
    init_models()
    Base.metadata.create_all(bind=_catalog_engine)
    db = CatalogSession()
    path = resolve_wyszukiwarka_path()
    assert path is not None
    import_wyszukiwarka_file(path, db=db)
    yield db
    db.close()


def test_token_stem_polish():
    assert _token_in_hay("fundusze", "program fundusz europejski")
    assert _token_in_hay("europejskie", "europejski program współpracy")


def test_search_region_and_status(db_session):
    # require_certainty=False: test filtrów regionu/statusu, nie bramki deadline
    results = search_catalog(
        db_session,
        query="OZE fotowoltaika",
        filters={
            "statuses": ["active", "planned"],
            "region": "mazowieckie",
            "require_certainty": False,
        },
        limit=20,
        require_certainty=False,
    )
    assert len(results) >= 1
    for item in results:
        assert item["status"] in ("active", "planned")
        assert is_catalog_visible(item)
        regions = " ".join(item.get("eligible_regions") or []).lower()
        assert "mazowieckie" in regions or "cała polska" in regions


def test_search_beneficiary_company_size(db_session):
    results = search_catalog(
        db_session,
        query="fundusze europejskie",
        filters={
            "beneficiary": "mśp",
            "company_size": "mikro",
            "require_certainty": False,
        },
        limit=20,
        require_certainty=False,
    )
    assert len(results) >= 1
    for item in results:
        hay = f"{item.get('beneficjenci', '')} {' '.join(item.get('beneficiary_tags') or [])}".lower()
        assert "mśp" in hay or "msp" in hay, f"beneficiary miss: {item.get('name')}"
        sizes = [s.lower() for s in (item.get("eligible_company_sizes") or [])]
        assert sizes, f"no eligible_company_sizes: {item.get('name')}"
        assert any("mikro" in s for s in sizes), f"size miss: {item.get('name')} {sizes}"
        assert _matches_filter(
            item, {"beneficiary": "mśp", "company_size": "mikro"}
        ), f"_matches_filter failed: {item.get('name')}"
    scores = [r["confidence_score"] for r in results]
    assert scores == sorted(scores, reverse=True)


def test_search_closed_with_region(db_session):
    results = search_catalog(
        db_session,
        query="",
        filters={"status": "closed", "region": "mazowieckie"},
        limit=30,
    )
    assert len(results) >= 1, "closed+mazowieckie should return >=1"
    for item in results:
        assert item["status"] == "closed"
        regions = " ".join(item.get("eligible_regions") or []).lower()
        assert "mazowieckie" in regions or "cała polska" in regions, item.get("name")


def test_search_closed_status_not_empty(db_session):
    results = search_catalog(db_session, filters={"status": "closed"}, limit=10)
    assert len(results) >= 1
    assert all(r["status"] == "closed" for r in results)


def test_company_size_filter_no_leakers(db_session):
    """Filtr company_size=mikro nie może przepuszczać rekordów bez dopasowania rozmiaru."""
    results = search_catalog(
        db_session,
        query="",
        filters={"company_size": "mikro", "statuses": ["active", "planned", "closed"]},
        limit=500,
    )
    for item in results:
        effective = _effective_company_sizes(item)
        assert effective, f"no effective sizes: {item.get('name')}"
        assert any("mikro" in s for s in effective), f"size miss: {item.get('name')} {effective}"
        assert _matches_filter(item, {"company_size": "mikro"}), item.get("name")