add conservative domain intelligence

apps/api/tests/test_phase7_domain.py

@@ -0,0 +1,92 @@
+import json
+import os
+import sqlite3
+import threading
+import unittest
+from http.client import HTTPConnection
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
+
+from app.domain_intelligence import (
+    normalize_registrable_domain, resolve_domain, generate_candidate_domains,
+    association_confidence, resolve_mx,
+)
+from app.main import create_server
+
+
+class DomainIntelligenceTests(unittest.TestCase):
+    def test_psl_normalization_and_unknown_suffix(self):
+        self.assertEqual(normalize_registrable_domain('https://WWW.shop.example.co.za/path'), 'example.co.za')
+        self.assertEqual(normalize_registrable_domain('foo.example.com'), 'example.com')
+        self.assertEqual(normalize_registrable_domain('foo.example.invalidtld'), 'unknown')
+
+    def test_candidate_generation_is_bounded_and_safe(self):
+        out = generate_candidate_domains('Acme & Sons (Pty) Ltd', 'Solar Panels', 'Cape Town')
+        self.assertLessEqual(len(out), 20)
+        self.assertTrue(out)
+        for value in out:
+            self.assertRegex(value, r'^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.(?:co\.za|com)$')
+            self.assertNotIn('--', value)
+
+    def test_resolution_distinguishes_nxdomain_timeout_and_error(self):
+        self.assertEqual(resolve_domain('missing.example.com', resolver=lambda *a: (_ for _ in ()).throw(socket_gaierror_name()))['status'], 'nxdomain')
+        self.assertEqual(resolve_domain('slow.example.com', resolver=lambda *a: (_ for _ in ()).throw(TimeoutError()))['status'], 'timeout')
+        self.assertEqual(resolve_domain('bad.example.com', resolver=lambda *a: (_ for _ in ()).throw(OSError('boom')))['status'], 'error')
+
+    def test_local_resolution_and_association_confidence(self):
+        with patch('app.domain_intelligence.socket.getaddrinfo', return_value=[(2, 1, 6, '', ('1.2.3.4', 0))]):
+            result = resolve_domain('example.com')
+        self.assertEqual(result['status'], 'ok')
+        self.assertEqual(result['addresses'], ['1.2.3.4'])
+        self.assertEqual(association_confidence('acme.co.za', 'acme.co.za')['level'], 'high')
+        self.assertEqual(association_confidence('acme.co.za', 'other.co.za')['level'], 'low')
+
+    def test_optional_capabilities_are_not_falsely_empty(self):
+        self.assertEqual(resolve_mx('example.com')['status'], 'not_configured')
+
+
+def socket_gaierror_name():
+    import socket
+    return socket.gaierror(socket.EAI_NONAME)
+
+
+class DomainApiTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory(); self.db_path = self.tmp.name + '/db.sqlite'
+        os.environ['BOOTSTRAP_ADMIN_EMAIL'] = 'domain-owner@example.test'; os.environ['BOOTSTRAP_ADMIN_PASSWORD'] = 'password'
+        self.server = create_server('127.0.0.1', 0, self.db_path); self.thread = threading.Thread(target=self.server.serve_forever, daemon=True); self.thread.start()
+        self.conn = HTTPConnection('127.0.0.1', self.server.server_port, timeout=4); self.cookie = None
+        self.request('POST', '/api/v1/auth/login', {'email':'domain-owner@example.test','password':'password'})
+
+    def tearDown(self):
+        self.server.shutdown(); self.server.server_close(); self.thread.join(2); self.tmp.cleanup()
+
+    def request(self, method, path, payload=None):
+        body = json.dumps(payload).encode() if payload is not None else None; headers = {'Content-Type':'application/json'} if body else {}
+        if self.cookie: headers['Cookie'] = self.cookie
+        self.conn.request(method, path, body, headers); response = self.conn.getresponse(); c = response.getheader('Set-Cookie')
+        if c: self.cookie = c.split(';', 1)[0]
+        return response.status, json.loads(response.read() or b'{}')
+
+    def test_check_cache_candidates_and_no_false_availability(self):
+        status, business = self.request('POST', '/api/v1/businesses', {'name':'Acme Solar','website':'https://example.com','city':'Cape Town','province':'Western Cape'})
+        self.assertEqual(status, 201); bid = business['id']
+        with patch('app.main.resolve_domain', return_value={'status':'unknown','addresses':[],'checked_at':'2026-01-01T00:00:00+00:00','cache_expires_at':'2099-01-01T00:00:00+00:00'}):
+            status, first = self.request('POST', f'/api/v1/businesses/{bid}/domains/check', {'domain':'example.com'})
+            self.assertEqual(status, 200); self.assertIn(first['status'], ('unknown','ok'))
+            status, second = self.request('GET', f'/api/v1/businesses/{bid}/domains/check?domain=example.com')
+        self.assertEqual(status, 200); self.assertTrue(second.get('cache_hit'))
+        status, candidates = self.request('GET', f'/api/v1/businesses/{bid}/domain-candidates')
+        self.assertEqual(status, 200); self.assertTrue(candidates['items'])
+        status, availability = self.request('POST', f'/api/v1/businesses/{bid}/domain-candidates/check-availability', {})
+        self.assertEqual(status, 200); self.assertEqual(availability['status'], 'unknown'); self.assertEqual(availability['reason'], 'not_configured')
+        self.assertEqual(self.request('GET', '/api/v1/domain-checks')[0], 200)
+
+    def test_business_domain_checks_are_tenant_scoped(self):
+        _, business = self.request('POST', '/api/v1/businesses', {'name':'Private'})
+        self.cookie = None
+        self.request('POST', '/api/v1/auth/logout')
+        self.assertEqual(self.request('GET', f'/api/v1/businesses/{business["id"]}/domains/check?domain=example.com')[0], 401)
+
+
+if __name__ == '__main__': unittest.main()