93 lines
5.4 KiB
Python
93 lines
5.4 KiB
Python
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()
|