104 lines
5.0 KiB
Python
104 lines
5.0 KiB
Python
"""Tests for the bounded post-discovery enrichment pipeline."""
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from app.enrichment import enrich_contacts, enrich_domain, enrich_website
|
|
|
|
|
|
def _page(status=200, html=b"<html><head><title>Acme</title></head><body><h1>Acme</h1></body></html>", redirects=None, elapsed=12, tls=True):
|
|
return {
|
|
"status": status,
|
|
"final_url": "https://acme.test/",
|
|
"redirect_chain": redirects or [],
|
|
"body": html,
|
|
"content_type": "text/html",
|
|
"elapsed_ms": elapsed,
|
|
"tls": tls,
|
|
"certificate_status": "valid" if tls else "not_applicable",
|
|
}
|
|
|
|
|
|
class EnrichmentTests(unittest.TestCase):
|
|
def test_enrich_website_uses_existing_scanner_fetcher(self):
|
|
fetch = lambda url, timeout=5.0, max_bytes=262144: _page()
|
|
with patch("app.enrichment.validate_url", side_effect=lambda url: url), \
|
|
patch("app.enrichment.normalize_registrable_domain", return_value="acme.test"):
|
|
result = enrich_website("https://acme.test/", fetch=fetch)
|
|
self.assertEqual(result["status"], "working")
|
|
self.assertTrue(result["has_working_website"])
|
|
self.assertEqual(result["domain"], "acme.test")
|
|
self.assertIsNotNone(result["response_time_ms"])
|
|
|
|
def test_enrich_website_marks_broken_on_4xx(self):
|
|
fetch = lambda url, **_: _page(status=404)
|
|
with patch("app.enrichment.validate_url", side_effect=lambda url: url):
|
|
result = enrich_website("https://acme.test/", fetch=fetch)
|
|
self.assertEqual(result["status"], "broken")
|
|
self.assertFalse(result["has_working_website"])
|
|
|
|
def test_enrich_website_detects_mobile_viewport(self):
|
|
html = b'<html><head><meta name="viewport" content="width=device-width"></head><body><h1>Acme</h1></body></html>'
|
|
fetch = lambda url, **_: _page(html=html)
|
|
with patch("app.enrichment.validate_url", side_effect=lambda url: url):
|
|
result = enrich_website("https://acme.test/", fetch=fetch)
|
|
self.assertTrue(result["mobile_viewport"])
|
|
|
|
def test_enrich_website_extracts_phone_and_email(self):
|
|
html = b'<html><body><h1>Acme</h1><p>+27 12 345 6789</p><a href="mailto:hello@acme.test">hello@acme.test</a></body></html>'
|
|
fetch = lambda url, **_: _page(html=html)
|
|
with patch("app.enrichment.validate_url", side_effect=lambda url: url):
|
|
result = enrich_website("https://acme.test/", fetch=fetch)
|
|
self.assertEqual(result["visible_phone"], "+27 12 345 6789")
|
|
self.assertEqual(result["visible_email"], "hello@acme.test")
|
|
|
|
def test_enrich_website_no_evidence_is_no_claim(self):
|
|
fetch = lambda url, **_: _page(status=404)
|
|
with patch("app.enrichment.validate_url", side_effect=lambda url: url):
|
|
result = enrich_website("https://acme.test/", fetch=fetch)
|
|
self.assertIsNone(result["visible_phone"])
|
|
self.assertIsNone(result["visible_email"])
|
|
|
|
def test_enrich_website_handles_fetch_failure(self):
|
|
def fetch(url, **_):
|
|
raise ConnectionError("timed out")
|
|
with patch("app.enrichment.validate_url", side_effect=lambda url: url):
|
|
result = enrich_website("https://acme.test/", fetch=fetch)
|
|
self.assertIn("error", result)
|
|
self.assertIsNone(result["has_working_website"])
|
|
|
|
def test_enrich_https_flag(self):
|
|
fetch = lambda url, **_: _page(tls=True)
|
|
with patch("app.enrichment.validate_url", side_effect=lambda url: url):
|
|
result = enrich_website("https://acme.test/", fetch=fetch)
|
|
self.assertTrue(result["https"])
|
|
self.assertTrue(result["ssl_valid"])
|
|
|
|
def test_enrich_contacts_delegates_to_extractor(self):
|
|
html = '<html><body><a href="mailto:hello@acme.test">Email</a><span>+27 12 345 6789</span></body></html>'
|
|
contacts = enrich_contacts(html, "https://acme.test/", max_results=10)
|
|
self.assertTrue(any(c["value"] == "hello@acme.test" for c in contacts))
|
|
self.assertTrue(any(c["kind"] == "phone" for c in contacts))
|
|
self.assertTrue(all(c["source_url"] == "https://acme.test/" for c in contacts))
|
|
|
|
def test_enrich_domain_unknown_for_unsupported_suffix(self):
|
|
result = enrich_domain("localhost")
|
|
self.assertEqual(result["status"], "unknown")
|
|
|
|
def test_enrich_domain_uses_resolution_evidence(self):
|
|
with patch("app.enrichment.normalize_registrable_domain", return_value="acme.co.za"), \
|
|
patch("app.enrichment.resolve_domain", return_value={"status": "ok", "addresses": ["1.2.3.4"]}):
|
|
result = enrich_domain("acme.co.za")
|
|
self.assertTrue(result["resolves"])
|
|
self.assertEqual(result["status"], "registered")
|
|
|
|
def test_enrich_domain_fails_closed(self):
|
|
with patch("app.enrichment.normalize_registrable_domain", return_value="acme.co.za"), \
|
|
patch("app.enrichment.resolve_domain", return_value={"status": "nxdomain"}):
|
|
result = enrich_domain("acme.co.za")
|
|
self.assertFalse(result["resolves"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|