"""P2/P3: binary skip, SSL-relaxed domains, flaky source timeouts — pure shipped helpers.""" from __future__ import annotations import asyncio import logging from unittest.mock import AsyncMock, MagicMock, patch import pytest # ── Binary / archive detection ─────────────────────────────────────────────── def test_is_binary_or_archive_url_zip_and_pdf(): from core.document_intel.fetch_resilience import ( is_binary_or_archive_url, path_extension, should_skip_html_extract, ) zip_url = "https://hrpgrants.com.pl/wp-content/uploads/pnr-eaa-regulamin-rekrutacji-031025-1.zip" pdf_url = "http://www.wfosigw.pl/sites/default/files/media/2014OSIGW19-Regulamin.pdf" html_url = "https://www.parp.gov.pl/component/grants/grants/sciezka-smart/" assert is_binary_or_archive_url(zip_url) is True assert path_extension(zip_url) == ".zip" assert should_skip_html_extract(zip_url) is True assert is_binary_or_archive_url(pdf_url) is True assert path_extension(pdf_url) == ".pdf" # PDF still skipped for HTML extract (dedicated PDF pipeline) assert should_skip_html_extract(pdf_url) is True assert is_binary_or_archive_url(html_url) is False assert should_skip_html_extract(html_url) is False def test_content_type_and_magic_bytes_skip(): from core.document_intel.fetch_resilience import ( is_binary_content_type, should_skip_html_extract, ) assert is_binary_content_type("application/zip") is True assert is_binary_content_type("application/pdf; charset=binary") is True assert is_binary_content_type("text/html; charset=utf-8") is False assert should_skip_html_extract("", content_type="application/x-zip-compressed") assert should_skip_html_extract("", body_prefix=b"PK\x03\x04....") assert should_skip_html_extract("", body_prefix=b"%PDF-1.4") def test_html_extract_skips_zip_without_trafilatura(caplog): from core.document_intel.html_extract import html_to_clean_text from core.document_intel.pipeline import extract_from_html, extract_document_from_url zip_url = "https://funduszeue.wzp.pl/wp-content/uploads/2026/06/Regulamin-wyboru-1.0.zip" with caplog.at_level(logging.ERROR): out = html_to_clean_text("x", url=zip_url) pipe = extract_from_html("", url=zip_url) doc = extract_document_from_url(zip_url) assert out.get("skipped") or out.get("extractor") == "skipped_binary" assert out["text"] == "" assert pipe.get("skipped") is True assert pipe["ok"] is False assert doc.get("skipped") is True or doc.get("reason") == "skipped_binary_archive" # No ERROR from Trafilatura empty tree for ZIP err_msgs = [r.message for r in caplog.records if r.levelno >= logging.ERROR] assert not any("empty HTML" in str(m) for m in err_msgs) def test_scrape_url_to_markdown_skips_zip(): import asyncio from core.crawl4ai_client import scrape_url_to_markdown async def _run(): md = await scrape_url_to_markdown( "https://example.com/files/regulamin-wyboru.zip" ) assert md == "" asyncio.run(_run()) # ── SSL-relaxed domains ────────────────────────────────────────────────────── def test_ssl_relaxed_includes_wfosigw_and_zus(): from core.document_intel.fetch_resilience import is_ssl_relaxed_domain from core.crawl4ai_client import is_ssl_relaxed_url assert is_ssl_relaxed_domain("https://www.wfosigw.pl/foo.pdf") is True assert is_ssl_relaxed_domain("http://www.wfosigw.pl/sites/default/files/x.pdf") is True assert is_ssl_relaxed_domain("https://prewencja.zus.pl/x") is True assert is_ssl_relaxed_domain("https://www.parp.gov.pl/x") is False assert is_ssl_relaxed_url("https://wfosigw.pl/a.pdf") is True @pytest.mark.asyncio async def test_download_pdf_ssl_fail_then_relaxed_success(tmp_path): """Shipped download_pdf: SSL error on verify=True → verify=False for allowlist.""" from rag_pipeline import pdf_parser class FakeResp: status_code = 200 content = b"%PDF-1.4 fake content for test " + b"x" * 100 def raise_for_status(self): return None call_state = {"n": 0} class FakeClient: def __init__(self, *a, **kw): self.verify = kw.get("verify", True) async def __aenter__(self): return self async def __aexit__(self, *a): return False async def get(self, url): call_state["n"] += 1 if self.verify is True: raise Exception( "[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: " "unable to get local issuer certificate" ) return FakeResp() with patch.object(pdf_parser, "httpx", create=True): import httpx as real_httpx with patch("httpx.AsyncClient", FakeClient): path = await pdf_parser.download_pdf( "https://www.wfosigw.pl/wp-content/uploads/2019/08/2-Regulamin.pdf" ) assert path is not None assert call_state["n"] >= 2 # strict fail + relaxed success data = open(path, "rb").read() assert data.startswith(b"%PDF") import os os.unlink(path) @pytest.mark.asyncio async def test_download_pdf_soft_fail_returns_none(): from rag_pipeline.pdf_parser import download_pdf class BoomClient: def __init__(self, *a, **kw): pass async def __aenter__(self): return self async def __aexit__(self, *a): return False async def get(self, url): raise ConnectionError("network down") with patch("httpx.AsyncClient", BoomClient): with patch( "core.document_intel.stealth_fetch.stealth_get_async", new=AsyncMock(return_value={"ok": False, "content": b""}), ): path = await download_pdf("https://example.com/missing.pdf") assert path is None # ── Flaky domain / timeout policy ──────────────────────────────────────────── def test_flaky_domain_funduszeeuropejskie_timeout_and_cache(): from core.document_intel.fetch_resilience import ( is_flaky_fetch_domain, fetch_timeout_for_url, prefer_cache_only_for_url, ) url = "https://www.funduszeeuropejskie.gov.pl/strony/regiony/" assert is_flaky_fetch_domain(url) is True assert fetch_timeout_for_url(url, default=28.0) >= 40.0 assert prefer_cache_only_for_url(url) is True # default FLAKY_PREFER_CACHE assert is_flaky_fetch_domain("https://www.parp.gov.pl/") is False assert fetch_timeout_for_url("https://www.parp.gov.pl/", default=28.0) == 28.0 def test_grant_search_timeout_policy_uses_flaky_helper(): """Source timeout selection uses fetch_timeout_for_url for regional/flaky.""" from core.document_intel.fetch_resilience import fetch_timeout_for_url t = fetch_timeout_for_url("https://www.funduszeeuropejskie.gov.pl", default=28.0) assert t > 28.0 # parallel loop must not raise on timeout — structural check of handler text import inspect from core.search.grant_search_service import GrantSearchService src = inspect.getsource(GrantSearchService.get_all_grants) assert "wait_for" in src assert "TimeoutError" in src or "timeout" in src.lower() assert "graceful" in src.lower() or "skipped for stability" in src def test_parse_pdf_from_url_failed_download_no_crash(): import asyncio from rag_pipeline.pdf_parser import parse_pdf_from_url async def _run(): with patch( "rag_pipeline.pdf_parser.download_pdf", new=AsyncMock(return_value=None), ): out = await parse_pdf_from_url("https://www.wfosigw.pl/x.pdf") assert out["text"] == "" assert out["parser"] == "failed_download" asyncio.run(_run())