"""Unit tests: catalog URL verify (P0-verify-in-pipeline).""" from __future__ import annotations from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from core.grants.models import Grant from core.subscription.db import Base, init_models @pytest.fixture def db_session(): init_models() engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}) Base.metadata.create_all(bind=engine) Session = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = Session() try: yield db finally: db.close() def _add_grant(db, *, source_id: str, last_verified=None, url=None, status="active", raw=None): row = Grant( source_id=source_id, name=f"Grant {source_id}", program="TEST", status=status, source="test", url=url or f"https://example.gov.pl/{source_id}", deadline="2027-12-31", last_verified=last_verified, operator="PARP", raw_data=raw or {}, ) db.add(row) db.commit() db.refresh(row) return row @pytest.mark.asyncio async def test_always_stamps_last_verified_on_ok(db_session): from core.grants.catalog_freshness import verify_catalog_grants row = _add_grant(db_session, source_id="stamp-ok", last_verified=None) assert row.last_verified is None with patch( "core.grants.catalog_freshness._check_url", new=AsyncMock(return_value={"ok": True, "outdated": False}), ): stats = await verify_catalog_grants(db_session, limit=10) db_session.refresh(row) assert stats["stamped"] == 1 assert stats["still_active"] == 1 assert row.last_verified is not None assert (row.raw_data or {}).get("last_verified") @pytest.mark.asyncio async def test_always_stamps_last_verified_on_fail(db_session): from core.grants.catalog_freshness import verify_catalog_grants row = _add_grant(db_session, source_id="stamp-fail", last_verified=None) with patch( "core.grants.catalog_freshness._check_url", new=AsyncMock(return_value={"ok": False, "reason": "http_5xx"}), ): stats = await verify_catalog_grants(db_session, limit=10) db_session.refresh(row) assert stats["stamped"] == 1 assert stats["url_warnings"] == 1 assert row.last_verified is not None assert row.url_warning == "http_5xx" # 5xx: soft-close only 404/410 — stay active, not hidden assert row.status == "active" assert not row.catalog_hidden @pytest.mark.asyncio async def test_soft_close_only_404_not_5xx(db_session): from core.grants.catalog_freshness import verify_catalog_grants dead = _add_grant(db_session, source_id="dead-404") five = _add_grant(db_session, source_id="soft-5xx") async def _check(url: str): if "dead-404" in url: return {"ok": False, "reason": "http_404"} return {"ok": False, "reason": "http_5xx"} with patch("core.grants.catalog_freshness._check_url", new=AsyncMock(side_effect=_check)): await verify_catalog_grants(db_session, limit=10) db_session.refresh(dead) db_session.refresh(five) assert dead.catalog_hidden is True assert dead.status == "closed" assert five.catalog_hidden is not True assert five.status == "active" assert five.url_warning == "http_5xx" @pytest.mark.asyncio async def test_order_nulls_first_then_oldest(db_session): from core.grants.catalog_freshness import verify_catalog_grants old = datetime.now(timezone.utc) - timedelta(days=30) mid = datetime.now(timezone.utc) - timedelta(days=1) _add_grant(db_session, source_id="already-new", last_verified=mid) _add_grant(db_session, source_id="never", last_verified=None) _add_grant(db_session, source_id="very-old", last_verified=old) checked_ids = [] async def _check(url: str): checked_ids.append(url.rstrip("/").rsplit("/", 1)[-1]) return {"ok": True, "outdated": False} with patch("core.grants.catalog_freshness._check_url", new=AsyncMock(side_effect=_check)): await verify_catalog_grants(db_session, limit=2) # NULLS FIRST, then ASC → never, then very-old (already-new skipped by limit) assert checked_ids == ["never", "very-old"] @pytest.mark.asyncio async def test_consecutive_fail_hide_optional(db_session, monkeypatch): from core.grants.catalog_freshness import verify_catalog_grants monkeypatch.setenv("CATALOG_VERIFY_CONSECUTIVE_FAIL_HIDE", "2") row = _add_grant( db_session, source_id="streak", raw={"url_verify_fail_streak": 1}, ) with patch( "core.grants.catalog_freshness._check_url", new=AsyncMock(return_value={"ok": False, "reason": "check_failed"}), ): stats = await verify_catalog_grants(db_session, limit=5) db_session.refresh(row) assert stats["hidden_consecutive"] == 1 assert row.catalog_hidden is True assert (row.raw_data or {}).get("url_verify_fail_streak") == 2 assert str((row.raw_data or {}).get("catalog_hidden_reason", "")).startswith("consecutive_fail:") def test_normalize_check_fail_reason(): from core.grants.catalog_freshness import _normalize_check_fail_reason assert _normalize_check_fail_reason(TimeoutError("read timed out")) == "timeout" assert _normalize_check_fail_reason(Exception("SSL: CERTIFICATE_VERIFY_FAILED")) == "ssl" assert _normalize_check_fail_reason(Exception("Connection refused")) == "connection" assert _normalize_check_fail_reason(Exception("weird boom")) == "check_failed" @pytest.mark.asyncio async def test_check_url_maps_5xx_and_exceptions(): from core.grants.catalog_freshness import _check_url class _Resp: def __init__(self, code, text=""): self.status_code = code self.text = text with patch("requests.get", return_value=_Resp(503)): out = await _check_url("https://example.gov.pl/x") assert out == {"ok": False, "reason": "http_5xx"} with patch("requests.get", return_value=_Resp(404)): out = await _check_url("https://example.gov.pl/y") assert out == {"ok": False, "reason": "http_404"} with patch("requests.get", side_effect=Exception("boom unknown")): out = await _check_url("https://example.gov.pl/z") assert out == {"ok": False, "reason": "check_failed"} @pytest.mark.asyncio async def test_post_import_calls_verify_after_sanitize(monkeypatch): """Pipeline stage order includes verify_catalog after sanitize_regulation_post.""" from core.grants import post_import_pipeline as pip monkeypatch.setenv("ENABLE_POST_IMPORT_PIPELINE", "true") monkeypatch.setenv("CATALOG_VERIFY_LIMIT", "7") db = MagicMock() call_n = {"sanitize": 0} def sanitize_side(*a, **k): call_n["sanitize"] += 1 return {} async def _html(*a, **k): return {} monkeypatch.setattr(pip, "is_post_import_pipeline_enabled", lambda: True) monkeypatch.setattr(pip, "sanitize_regulation_urls", sanitize_side) monkeypatch.setattr(pip, "backfill_regulation_urls", lambda *a, **k: {}) monkeypatch.setattr(pip, "normalize_catalog_defaults", lambda *a, **k: {}) monkeypatch.setattr(pip, "reparse_eligibility_from_text", lambda *a, **k: {}) monkeypatch.setattr(pip, "_fetch_html_for_discovery", _html) monkeypatch.setattr(pip, "discover_regulation_urls", lambda *a, **k: {}) monkeypatch.setattr(pip, "_refresh_completeness_scores", lambda *a, **k: {}) vmock = AsyncMock(return_value={"checked": 0, "stamped": 0}) with patch( "core.grants.catalog_freshness.verify_catalog_grants", new=vmock, ), patch( "core.grants.regulation_ingest.ingest_regulations_from_catalog", new=AsyncMock(return_value={}), ), patch( "core.grants.live_research.establish_research_baselines", return_value={}, ), patch( "core.grants.postgres_fts.sync_fts_index", return_value={}, ), patch( "core.grants.catalog_semantic_index.index_grants_semantic", return_value={}, ), patch( "core.grants.regulation_backfill.regulation_completeness_report", return_value={}, ): result = await pip.run_post_import_pipeline(db) assert result["enabled"] is True assert "verify_catalog" in result["stages"] assert call_n["sanitize"] >= 2 # pre + post vmock.assert_awaited() assert vmock.await_args.kwargs.get("limit") == 7 stage_keys = list(result["stages"].keys()) assert stage_keys.index("sanitize_regulation_post") < stage_keys.index("verify_catalog")