Spaces:
Sleeping
Sleeping
File size: 8,247 Bytes
ce8f04a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | """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
@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())
|