"""Unit tests for the NSCT crawler module.""" from __future__ import annotations import hashlib import os import socket import time import pytest from nsct.crawler.extraction import extract_main_content from nsct.crawler.fetcher import AsyncFetcher, FetchResult, FetchStatus from nsct.crawler.manager import CrawlerManager from nsct.crawler.normalize import NormalizedDocument from nsct.crawler.policy import ( CrawlerPolicyError, InvalidURLError, SSRFValidationError, _is_local_host, validate_url, ) # --------------------------------------------------------------------------- # SSRF-Schutz: localhost, private IPs blockiert # --------------------------------------------------------------------------- def test_ssrf_block_localhost() -> None: """localhost access must be blocked by default.""" with pytest.raises(SSRFValidationError, match="localhost"): validate_url("http://localhost/test") def test_ssrf_block_127() -> None: """127.0.0.1 must be blocked.""" with pytest.raises(SSRFValidationError): validate_url("http://127.0.0.1/test") def test_ssrf_block_private_10() -> None: """10.x.x.x private range must be blocked.""" with pytest.raises(SSRFValidationError): validate_url("http://10.0.0.1/internal") def test_ssrf_block_private_172() -> None: """172.16-31.x.x private range must be blocked.""" with pytest.raises(SSRFValidationError): validate_url("http://172.16.0.1/internal") def test_ssrf_block_private_192() -> None: """192.168.x.x private range must be blocked.""" with pytest.raises(SSRFValidationError): validate_url("http://192.168.1.1/internal") def test_ssrf_block_metadata() -> None: """AWS/GCP metadata endpoints must be blocked.""" with pytest.raises(SSRFValidationError): validate_url("http://169.254.169.254/latest/meta-data/") def test_ssrf_block_file_scheme() -> None: """file:// scheme must be blocked.""" with pytest.raises(InvalidURLError): validate_url("file:///etc/passwd") def test_ssrf_block_ftp_scheme() -> None: """ftp:// scheme must be blocked.""" with pytest.raises(InvalidURLError): validate_url("ftp://ftp.example.com/file") def test_ssrf_safety_https_public() -> None: """Public HTTPS URLs should pass validation.""" validate_url("https://example.com/page") def test_ssrf_safety_http_public() -> None: """Public HTTP URLs should pass validation.""" validate_url("http://example.com/page") # --------------------------------------------------------------------------- # robots.txt Policy (structural test) # --------------------------------------------------------------------------- def test_validate_url_empty_raises() -> None: """Empty string should raise.""" with pytest.raises(InvalidURLError): validate_url("") def test_validate_url_none_raises() -> None: """None should raise.""" with pytest.raises(InvalidURLError): validate_url("") # type: ignore def test_validate_url_no_hostname() -> None: """URL without hostname should raise.""" with pytest.raises(InvalidURLError): validate_url("https:///path") def test_validate_url_data_scheme() -> None: """data:// scheme must be blocked.""" with pytest.raises(InvalidURLError): validate_url("data:text/html,

hello

") # --------------------------------------------------------------------------- # Content Extraction (HTML -> Text) # --------------------------------------------------------------------------- def test_extract_main_content_basic() -> None: """Basic HTML extraction must return readable text.""" html = "

Hello world

" text = extract_main_content(html) assert "Hello world" in text def test_extract_main_content_strips_nav() -> None: """Navigation elements should be minimised.""" html = ( "" "" "

Main content here

" "" "" ) text = extract_main_content(html) assert "Main content here" in text def test_extract_main_content_empty() -> None: """Empty input must return empty string.""" assert extract_main_content("") == "" def test_extract_main_content_plain_text() -> None: """Plain text (no HTML) should pass through.""" text = extract_main_content("Just plain text\nNo HTML here", url="plain://test") assert "plain text" in text def test_extract_main_content_script_removed() -> None: """Script tags should be stripped.""" html = "

Clean

" text = extract_main_content(html) assert "evil" not in text.lower() or "clean" in text.lower() # --------------------------------------------------------------------------- # URL Dedup (structural check — manager does this) # --------------------------------------------------------------------------- def test_url_dedup_no_duplicate() -> None: """URL deduplication should not produce duplicate entries.""" urls = ["https://a.com", "https://b.com", "https://a.com"] seen: set[str] = [] for u in urls: if u not in seen: seen.append(u) assert len(seen) == 2 # --------------------------------------------------------------------------- # Error Handling (invalid URLs, timeouts) # --------------------------------------------------------------------------- async def test_fetcher_configures_all_httpx_timeout_phases() -> None: """httpx requires a pool timeout when the other phases are explicit.""" fetcher = AsyncFetcher() assert fetcher._client.timeout.pool == 10.0 await fetcher.close() async def test_fetcher_blocked_url() -> None: """Fetcher must return BLOCKED for SSRF-blocked URLs.""" fetcher = AsyncFetcher() result = await fetcher.fetch("http://localhost/test") assert result.status == FetchStatus.BLOCKED assert result.error is not None await fetcher.close() async def test_fetcher_invalid_url() -> None: """Fetcher must return BLOCKED for invalid URLs.""" fetcher = AsyncFetcher() result = await fetcher.fetch("") assert result.status == FetchStatus.BLOCKED await fetcher.close() async def test_fetcher_unreachable_domain() -> None: """Fetcher handles DNS failure gracefully.""" fetcher = AsyncFetcher() result = await fetcher.fetch("http://this-domain-definitely-does-not-exist-12345.com/") assert result.status in (FetchStatus.ERROR, FetchStatus.TIMEOUT) assert result.error is not None await fetcher.close() def test_error_document_has_hash_for_empty_content() -> None: """A failed fetch must still yield a valid normalized document.""" doc = CrawlerManager._error_doc("https://example.com/unavailable", "connection refused") assert doc.text == "" assert doc.metadata["error"] == "connection refused" assert doc.content_hash == hashlib.sha256(b"").hexdigest() # --------------------------------------------------------------------------- # NormalizedDocument has all required fields # --------------------------------------------------------------------------- def test_normalized_document_required_fields() -> None: """NormalizedDocument must have all required fields.""" doc = NormalizedDocument.from_text( url="https://example.com", text="Hello world", title="Test", extraction_tool="test_tool", ) assert doc.url == "https://example.com" assert doc.title == "Test" assert doc.text == "Hello world" assert doc.links == [] assert doc.content_hash != "" assert doc.extraction_tool == "test_tool" assert doc.word_count == 2 assert isinstance(doc.extracted_at, type(doc.extracted_at)) def test_normalized_document_is_valid() -> None: """is_valid must be True for valid docs.""" doc = NormalizedDocument.from_text(url="https://e.com", text="content") assert doc.is_valid is True def test_normalized_document_is_valid_false() -> None: """is_valid must be False when text is empty.""" doc = NormalizedDocument.from_text(url="https://e.com", text="") assert doc.is_valid is False # --------------------------------------------------------------------------- # Content Hash (deterministic) # --------------------------------------------------------------------------- def test_content_hash_deterministic() -> None: """Same text must produce same hash.""" text = "Hello world, deterministic content." h1 = hashlib.sha256(text.encode("utf-8")).hexdigest() h2 = hashlib.sha256(text.encode("utf-8")).hexdigest() assert h1 == h2 def test_content_hash_differs_for_different_content() -> None: """Different text must produce different hashes.""" t1 = "Hello world" t2 = "Goodbye world" h1 = hashlib.sha256(t1.encode("utf-8")).hexdigest() h2 = hashlib.sha256(t2.encode("utf-8")).hexdigest() assert h1 != h2 def test_normalized_document_hash_matches() -> None: """NormalizedDocument.from_text must compute correct hash.""" doc = NormalizedDocument.from_text( url="https://example.com", text="test content for hash", ) expected = hashlib.sha256(b"test content for hash").hexdigest() assert doc.content_hash == expected