backend/tests/test_ops_fixes.py

"""Tests for HF deployment operational fixes (#2–#7).

#2 DB SSL retry, #3 Crawl4AI blocks/SSL, #4 scheduler leader lock,
#5 bootstrap catalog leader lock, #6 ISAP ELI URLs, #7 PII query sanitization.
"""
from unittest.mock import MagicMock, patch

import pytest
from sqlalchemy.exc import OperationalError


def test_isap_client_uses_eli_text_url():
    from integrations.isap_client import ISAPClient

    client = ISAPClient()
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.headers = {"Content-Type": "application/json"}
    mock_resp.text = (
        '{"ELI":"DU/2018/646","address":"WDU20180000646","displayAddress":"Dz.U. 2018 poz. 646",'
        '"texts":[{"fileName":"text.html","type":"H"},{"fileName":"D20180646Lj.pdf","type":"U"}]}'
    )
    mock_resp.json.return_value = {
        "ELI": "DU/2018/646",
        "address": "WDU20180000646",
        "displayAddress": "Dz.U. 2018 poz. 646",
        "texts": [
            {"fileName": "text.html", "type": "H"},
            {"fileName": "D20180646Lj.pdf", "type": "U"},
        ],
    }
    mock_resp.raise_for_status = MagicMock()

    with patch("integrations.isap_client.requests.get", return_value=mock_resp):
        act = client.fetch_act("WDU", 2018, 646)

    assert act is not None
    assert act["text_url"] == "https://api.sejm.gov.pl/eli/acts/DU/2018/646/text"
    assert "dziennikustaw.gov.pl" not in act["text_url"]
    assert act["text_pdf_url"] == "https://api.sejm.gov.pl/eli/acts/DU/2018/646/D20180646Lj.pdf"


def test_sensitive_data_guard_strips_llm_meta():
    from core.sensitive_data_guard import SensitiveDataGuard

    raw = (
        "Zgodnie z Twoimi instrukcjami, oto zanonimizowany tekst:\n\n"
        "Projekt dotyczy <KNOW_HOW_1> w sektorze OZE."
    )
    fallback = "Projekt dotyczy technologii w sektorze OZE."
    cleaned = SensitiveDataGuard._sanitize_llm_anonymization(raw, fallback)
    assert "Zgodnie z Twoimi" not in cleaned
    assert "<KNOW_HOW_1>" in cleaned


def test_eurlex_sanitize_query_strips_meta():
    from integrations.eurlex_client import _sanitize_search_query

    polluted = (
        "Zgodnie z Twoimi instrukcjami przesyłam tekst.\n"
        "Rozporządzenie UE w sprawie funduszy strukturalnych"
    )
    clean = _sanitize_search_query(polluted)
    assert "Zgodnie z Twoimi" not in clean
    assert "funduszy strukturalnych" in clean


def test_with_db_retry_recovers_from_ssl_disconnect():
    from core.subscription import db as db_module

    calls = {"n": 0}

    def flaky():
        calls["n"] += 1
        if calls["n"] < 2:
            raise OperationalError("stmt", {}, Exception("SSL connection has been closed unexpectedly"))
        return 99

    with patch.object(db_module, "refresh_engine_pool"):
        assert db_module.with_db_retry(flaky, label="test") == 99
    assert calls["n"] == 2


def test_crawl4ai_hard_blocked_and_ssl_relaxed():
    from core.crawl4ai_client import is_hard_blocked_url, _is_ssl_relaxed

    assert is_hard_blocked_url("https://www.parp.gov.pl/component/grants/")
    assert is_hard_blocked_url("https://www.bgk.pl/fundusze-europejskie/")
    assert not is_hard_blocked_url("https://www.ncbr.gov.pl/")
    assert _is_ssl_relaxed("https://prewencja.zus.pl/abc")


def test_scheduler_respects_run_scheduler_false():
    import os
    from core import scheduler

    with patch.dict(os.environ, {"RUN_SCHEDULER": "false"}):
        assert scheduler._acquire_scheduler_leader() is False


def test_bootstrap_respects_run_bootstrap_catalog_false():
    import os
    import grantforge_bootstrap

    with patch.dict(os.environ, {"RUN_BOOTSTRAP_CATALOG": "false"}):
        assert grantforge_bootstrap._acquire_bootstrap_leader() is False