import json import os import sqlite3 import threading import time import unittest from http.client import HTTPConnection from tempfile import TemporaryDirectory from unittest.mock import patch from app.main import create_server, hash_password class ScopedDiscoveryApiTests(unittest.TestCase): def setUp(self): self.tmp = TemporaryDirectory() for key in ('SEARCH_PROVIDER_URL', 'SEARCH_PROVIDER_ALLOWED_HOSTS', 'SEARCH_PROVIDER_API_KEY'): os.environ.pop(key, None) os.environ['BOOTSTRAP_ADMIN_EMAIL'] = 'discover-owner@example.test' os.environ['BOOTSTRAP_ADMIN_PASSWORD'] = 'password' self.server = create_server('127.0.0.1', 0, self.tmp.name + '/db.sqlite') 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': 'discover-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() cookie = response.getheader('Set-Cookie') if cookie: self.cookie = cookie.split(';', 1)[0] return response.status, json.loads(response.read() or b'{}') def test_direct_criteria_job_crawls_allowlisted_site_and_persists_evidence(self): pages = { 'https://directory.test/': {'status': 200, 'final_url': 'https://directory.test/', 'content_type': 'text/html', 'body': b'

Acme Solar

Acme'}, 'https://acme.test/': {'status': 200, 'final_url': 'https://acme.test/', 'content_type': 'text/html', 'body': b'Acme Solar

Acme Solar

Solar installers

Contact'}, 'https://acme.test/contact': {'status': 200, 'final_url': 'https://acme.test/contact', 'content_type': 'text/html', 'body': b'

Contact Acme

Email

+27 11 555 0100

'}, } def fetch(url, **_): value = pages[url]; return dict(value, redirect_chain=[], elapsed_ms=1, tls=url.startswith('https://'), certificate_status='valid') with patch('app.discovery._fetch', side_effect=fetch), patch('app.discovery.validate_url', side_effect=lambda url, **_: url), patch('app.main.validate_url', side_effect=lambda url, **_: url): status, created = self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['solar'], 'location': 'Cape Town'}, 'seed_urls': ['https://directory.test/'], 'max_pages': 5, 'idempotency_key': 'scope-1'}) self.assertEqual(status, 202); self.assertEqual(created['type'], 'scoped_discovery') for _ in range(50): _, job = self.request('GET', '/api/v1/jobs/' + str(created['id'])) if job['status'] in ('succeeded', 'failed'): break time.sleep(.02) self.assertEqual(job['status'], 'succeeded') status, runs = self.request('GET', '/api/v1/discovery-runs') self.assertEqual(status, 200); self.assertEqual(runs['items'][0]['criteria']['location'], 'Cape Town') self.assertEqual(runs['items'][0]['result_count'], 1) status, businesses = self.request('GET', '/api/v1/businesses') self.assertEqual(status, 200); self.assertEqual(businesses['items'][0]['website_domain'], 'acme.test') detail = self.request('GET', '/api/v1/businesses/' + str(businesses['items'][0]['id']))[1] self.assertTrue(any(x['url'] == 'https://acme.test/contact' for x in detail['evidence'])) self.assertTrue(any(x['source_url'] == 'https://acme.test/contact' and x['provenance'] == 'mailto' for x in detail['contact_extractions'])) self.assertFalse(detail.get('outreach_enabled', False)) def test_requires_bounded_seed_allowlist_and_rejects_ssrf(self): status, body = self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['x']}}) self.assertEqual(status, 503); self.assertEqual(body['error'], 'not_configured') status, body = self.request('POST', '/api/v1/discovery', {'criteria': {}, 'seed_urls': ['http://127.0.0.1/'], 'idempotency_key': 'bad'}) self.assertEqual(status, 400); self.assertEqual(body['error'], 'unsafe_seed_url') def test_criteria_first_search_results_flow_through_existing_job_persistence(self): os.environ['SEARCH_PROVIDER_URL'] = 'https://search.example.test/query' os.environ['SEARCH_PROVIDER_ALLOWED_HOSTS'] = 'search.example.test' pages = {'https://acme.test/': {'status': 200, 'final_url': 'https://acme.test/', 'content_type': 'text/html', 'body': b'Acme Solar

Acme Solar

solar

'}} def fetch(url, **_): value = pages[url]; return dict(value, redirect_chain=[], elapsed_ms=1, tls=True, certificate_status='valid') with patch('app.discovery.search_provider', return_value=['https://acme.test/']) as provider, patch('app.discovery._fetch', side_effect=fetch), patch('app.discovery.validate_url', side_effect=lambda url, **_: url), patch('app.main.validate_url', side_effect=lambda url, **_: url): status, created = self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['solar']}, 'max_candidates': 1, 'idempotency_key': 'criteria-1'}) self.assertEqual(status, 202) for _ in range(50): _, job = self.request('GET', '/api/v1/jobs/' + str(created['id'])) if job['status'] in ('succeeded', 'failed'): break time.sleep(.02) self.assertEqual(job['status'], 'succeeded'); provider.assert_called_once_with({'keywords': ['solar']}, 1) run = self.request('GET', '/api/v1/discovery-runs')[1]['items'][0] self.assertEqual(run['seed_urls'], []); self.assertEqual(run['result']['candidates'][0]['provenance']['mechanism'], 'criteria_search_provider') def test_results_are_tenant_isolated(self): ph, salt = hash_password('other-password') db = sqlite3.connect(self.server.db_path); db.execute("INSERT INTO organizations VALUES ('other-tenant','Other',CURRENT_TIMESTAMP)"); db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ('other-tenant','other@example.test',ph,salt,'owner')); db.commit(); db.close() self.cookie = None; self.request('POST', '/api/v1/auth/login', {'email': 'other@example.test', 'password': 'other-password'}) self.assertEqual(self.request('GET', '/api/v1/discovery-runs')[1]['items'], []) if __name__ == '__main__': unittest.main()