Spaces:
Sleeping
Sleeping
| """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("<html><body>x</body></html>", url=zip_url) | |
| pipe = extract_from_html("<html></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 | |
| 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) | |
| 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()) | |