239 lines
17 KiB
Python
239 lines
17 KiB
Python
import json, os, sqlite3, threading, unittest
|
|
from unittest.mock import patch
|
|
from http.client import HTTPConnection
|
|
from tempfile import TemporaryDirectory
|
|
from app.main import ORGANIZATION_ID, connect, create_server
|
|
from app.sources import ApprovedDirectorySource, CsvSource, ManualSource, available_adapters
|
|
|
|
class SourceAdapterTests(unittest.TestCase):
|
|
def test_adapter_catalog_exposes_ready_and_gated_sources(self):
|
|
catalog = {item['source_code']: item for item in available_adapters()}
|
|
for code in ('manual', 'csv', 'public_website', 'ct_logs', 'dns', 'rdap'):
|
|
self.assertTrue(catalog[code]['available'], code)
|
|
for code in ('bing_local', 'permitted_social'):
|
|
self.assertFalse(catalog[code]['available'], code)
|
|
self.assertTrue(catalog[code]['optional'], code)
|
|
for code in ('openstreetmap','wikidata','common_crawl'):
|
|
self.assertTrue(catalog[code]['available'], code)
|
|
self.assertFalse(catalog[code]['optional'], code)
|
|
self.assertTrue(catalog['google_places']['available'])
|
|
self.assertTrue(catalog['google_places']['optional'])
|
|
|
|
def test_csv_adapter_is_deterministic_and_normalizes(self):
|
|
src = CsvSource()
|
|
a = src.discover({'csv': 'Name,Website,Email\n Acme ,https://acme.test,a@acme.test\n'})
|
|
b = src.discover({'csv': 'Name,Website,Email\n Acme ,https://acme.test,a@acme.test\n'})
|
|
self.assertEqual(a.records, b.records)
|
|
self.assertEqual(a.records[0]['name'], 'Acme')
|
|
self.assertEqual(a.records[0]['email'], 'a@acme.test')
|
|
|
|
def test_manual_validation_rejects_secret_fields(self):
|
|
result = ManualSource().validate({'rows': [{'name': 'x', 'api_key': 'secret'}]})
|
|
self.assertFalse(result.valid)
|
|
self.assertIn('secret', result.errors[0].lower())
|
|
|
|
def test_openstreetmap_uses_provider_side_service_tag_query(self):
|
|
class Response:
|
|
def read(self, _): return b'{"elements":[{"tags":{"name":"Cape Plumber","craft":"plumber"}}]}'
|
|
def __enter__(self): return self
|
|
def __exit__(self, *_): return False
|
|
config={'provider':'openstreetmap','query':'plumbers','location':'Cape Town','approved':True,'public_access':True,'terms_accepted':True,'rate_limit':1}
|
|
with patch('app.sources.urlopen',return_value=Response()) as request:
|
|
page=ApprovedDirectorySource().discover(config)
|
|
query=request.call_args.args[0].data.decode()
|
|
self.assertIn('"craft"~"plumber|plumbers",i]',query)
|
|
self.assertIn('"shop"~"plumber|plumbers",i]',query)
|
|
self.assertEqual(page.records[0]['name'],'Cape Plumber')
|
|
|
|
def test_legacy_source_kind_constraint_is_migrated(self):
|
|
with TemporaryDirectory() as tmp:
|
|
path=os.path.join(tmp,'legacy.db')
|
|
legacy=sqlite3.connect(path)
|
|
legacy.execute("CREATE TABLE sources (id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL, name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual')), enabled INTEGER NOT NULL DEFAULT 0, config_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown', consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0, last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', approved INTEGER NOT NULL DEFAULT 0, policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', UNIQUE(organization_id,name))")
|
|
legacy.execute("INSERT INTO sources(organization_id,name,kind) VALUES(?, 'Existing manual','manual')",(ORGANIZATION_ID,)); legacy.commit(); legacy.close()
|
|
db=connect(path)
|
|
self.assertEqual(db.execute("SELECT kind FROM sources WHERE name='Existing manual'").fetchone()[0],'manual')
|
|
db.execute("INSERT INTO sources(organization_id,name,kind,source_code) VALUES(?,?,?,?)",(ORGANIZATION_ID,'OpenStreetMap / Overpass · plumbers','approved_directory','openstreetmap'))
|
|
db.commit(); self.assertEqual(db.execute("SELECT kind FROM sources WHERE source_code='openstreetmap'").fetchone()[0],'approved_directory'); db.close()
|
|
|
|
class SourceApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=TemporaryDirectory(); os.environ['BOOTSTRAP_ADMIN_EMAIL']='owner@example.test'; os.environ['BOOTSTRAP_ADMIN_PASSWORD']='development-password'
|
|
self.server=create_server('127.0.0.1',0,self.tmp.name+'/x.db'); self.thread=threading.Thread(target=self.server.serve_forever,daemon=True); self.thread.start(); self.c=HTTPConnection('127.0.0.1',self.server.server_port); self.cookie=None
|
|
self.req('POST','/api/v1/auth/login',{'email':'owner@example.test','password':'development-password'})
|
|
def tearDown(self): self.server.shutdown(); self.server.server_close(); self.thread.join(2); self.tmp.cleanup()
|
|
def req(self,m,p,x=None,cookie=True):
|
|
body=json.dumps(x).encode() if x is not None else None; h={'Content-Type':'application/json'} if body else {};
|
|
if cookie and self.cookie:h['Cookie']=self.cookie
|
|
self.c.request(m,p,body,h); r=self.c.getresponse(); sc=r.getheader('Set-Cookie');
|
|
if sc:self.cookie=sc.split(';',1)[0]
|
|
raw=r.read(); return r.status,json.loads(raw or b'{}')
|
|
def test_source_lifecycle_ingestion_idempotency_health_and_disabled(self):
|
|
s, source=self.req('POST','/api/v1/sources',{'name':'Import','kind':'manual','config':{}}); self.assertEqual(s,201)
|
|
sid=source['id']; self.assertFalse(source['enabled'])
|
|
self.assertEqual(self.req('PATCH',f'/api/v1/sources/{sid}',{'enabled':True})[0],200)
|
|
payload={'rows':[{'name':'Acme','website':'https://acme.test'}], 'source_url':'file://import.csv','query_context':{'q':'test'}}
|
|
self.assertEqual(self.req('POST',f'/api/v1/sources/{sid}/ingest',payload)[0],201)
|
|
self.assertEqual(self.req('POST',f'/api/v1/sources/{sid}/ingest',payload)[0],200)
|
|
self.assertEqual(len(self.req('GET','/api/v1/source-records')[1]['items']),1)
|
|
self.assertEqual(self.req('POST',f'/api/v1/sources/{sid}/test')[0],200)
|
|
self.assertEqual(self.req('PATCH',f'/api/v1/sources/{sid}',{'enabled':False})[0],200)
|
|
self.assertEqual(self.req('POST',f'/api/v1/sources/{sid}/ingest',payload)[0],409)
|
|
db=sqlite3.connect(self.tmp.name+'/x.db'); self.assertTrue(db.execute("select 1 from audit_log where action='source.disabled'").fetchone()); db.close()
|
|
def test_queries_enqueue_and_records_are_tenant_scoped(self):
|
|
_,source=self.req('POST','/api/v1/sources',{'name':'CSV','kind':'csv','enabled':True,'config':{'csv':'name\nA'}})
|
|
_,q=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'q','query':{}})
|
|
status,job=self.req('POST',f"/api/v1/discovery-queries/{q['id']}/run",{})
|
|
self.assertEqual(status,202); self.assertEqual(job['type'],'source_discovery')
|
|
for _ in range(100):
|
|
_, current=self.req('GET',f"/api/v1/jobs/{job['id']}")
|
|
if current['status'] in ('succeeded','failed'): break
|
|
threading.Event().wait(.01)
|
|
self.assertEqual(current['status'],'succeeded')
|
|
self.assertEqual(len(self.req('GET','/api/v1/businesses')[1]['items']),1)
|
|
self.assertEqual(self.req('GET','/api/v1/source-records')[1]['items'][0]['discovery_query_id'],q['id'])
|
|
self.assertEqual(self.req('GET','/api/v1/source-records?page_size=101')[0],400)
|
|
def test_sources_require_auth(self):
|
|
self.cookie=None; self.assertEqual(self.req('GET','/api/v1/sources',cookie=False)[0],401)
|
|
|
|
def test_unavailable_or_unconfigured_sources_cannot_be_enabled(self):
|
|
status, gated = self.req('POST', '/api/v1/sources', {'name': 'Google', 'kind': 'google_places', 'config': {}})
|
|
self.assertEqual(status, 201)
|
|
self.assertEqual(self.req('PATCH', f"/api/v1/sources/{gated['id']}", {'enabled': True})[0], 409)
|
|
status, website = self.req('POST', '/api/v1/sources', {'name': 'Web', 'kind': 'public_website', 'config': {}})
|
|
self.assertEqual(status, 201)
|
|
self.assertEqual(self.req('PATCH', f"/api/v1/sources/{website['id']}", {'enabled': True})[0], 409)
|
|
|
|
def test_duplicate_source_registration_is_idempotent(self):
|
|
payload={'name':'OpenStreetMap / Overpass · plumbers','kind':'openstreetmap','config':{'provider':'openstreetmap','approved':True,'public_access':True,'terms_accepted':True,'rate_limit':1}}
|
|
status, created=self.req('POST','/api/v1/sources',payload); self.assertEqual(status,201); self.assertTrue(created['created'])
|
|
status, reused=self.req('POST','/api/v1/sources',payload); self.assertEqual(status,200); self.assertFalse(reused['created']); self.assertEqual(reused['id'],created['id'])
|
|
second={**payload,'name':'OpenStreetMap / Overpass · plumbers · Durban'}
|
|
status, other=self.req('POST','/api/v1/sources',second); self.assertEqual(status,201); self.assertTrue(other['created']); self.assertNotEqual(other['id'],created['id'])
|
|
|
|
def test_source_configuration_rejects_discovery_criteria(self):
|
|
forbidden = {'query': 'plumbers', 'category': 'trades', 'city': 'Cape Town', 'location': 'Western Cape'}
|
|
for field, value in forbidden.items():
|
|
with self.subTest(field=field):
|
|
status, body = self.req('POST', '/api/v1/sources', {
|
|
'name': 'Manual ' + field, 'kind': 'manual',
|
|
'config': {'rows': [], field: value},
|
|
})
|
|
self.assertEqual(status, 400)
|
|
self.assertEqual(body['error'], 'source_configuration_contains_criteria')
|
|
|
|
def test_source_worker_passes_query_criteria_and_effective_limits_to_connector(self):
|
|
status, source = self.req('POST', '/api/v1/sources', {
|
|
'name': 'Criteria manual', 'kind': 'manual', 'enabled': True,
|
|
'config': {'rows': [{'name': 'Criteria Acme'}]},
|
|
})
|
|
self.assertEqual(status, 201)
|
|
status, query = self.req('POST', '/api/v1/discovery-queries', {
|
|
'source_id': source['id'], 'name': 'Cape solar',
|
|
'query': {'keywords': ['solar'], 'city': 'Cape Town'},
|
|
'max_records': 7, 'daily_limit': 9,
|
|
})
|
|
self.assertEqual(status, 201)
|
|
observed = []
|
|
from app.sources import ManualSource
|
|
original = ManualSource.discover
|
|
def spy(adapter, config, cursor=None, criteria=None, limits=None):
|
|
observed.append((criteria, limits))
|
|
return original(adapter, config, cursor, criteria=criteria, limits=limits)
|
|
with patch('app.sources.ManualSource.discover', new=spy):
|
|
status, job = self.req('POST', f"/api/v1/discovery-queries/{query['id']}/run", {})
|
|
self.assertEqual(status, 202)
|
|
for _ in range(100):
|
|
_, current = self.req('GET', f"/api/v1/jobs/{job['id']}")
|
|
if current['status'] in ('succeeded', 'failed'):
|
|
break
|
|
threading.Event().wait(.01)
|
|
self.assertEqual(current['status'], 'succeeded')
|
|
self.assertEqual(observed, [({'keywords': ['solar'], 'city': 'Cape Town'}, {'max_records': 7, 'daily_limit': 9, 'per_run_limit': 7})])
|
|
|
|
def test_source_configuration_can_be_saved_before_enablement(self):
|
|
status, source = self.req('POST', '/api/v1/sources', {'name': 'DNS', 'kind': 'dns', 'config': {}})
|
|
self.assertEqual(status, 201)
|
|
status, configured = self.req('PATCH', f"/api/v1/sources/{source['id']}", {'config': {'domains': ['example.co.za']}})
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(self.req('PATCH', f"/api/v1/sources/{source['id']}", {'enabled': True})[0], 200)
|
|
|
|
def test_fresh_schema_accepts_optional_source_kind_fail_closed(self):
|
|
status, source = self.req('POST', '/api/v1/sources', {'name': 'RDAP', 'kind': 'rdap', 'config': {}})
|
|
self.assertEqual(status, 201)
|
|
self.assertEqual(source['kind'], 'rdap')
|
|
self.assertEqual(self.req('POST', f"/api/v1/sources/{source['id']}/test", {})[0], 200)
|
|
status, health = self.req('GET', f"/api/v1/sources/{source['id']}/health")
|
|
self.assertEqual(status, 200)
|
|
self.assertFalse(health['configured'])
|
|
|
|
def test_disabled_selected_source_fails_instead_of_succeeding_with_zero_records(self):
|
|
status, source=self.req('POST','/api/v1/sources',{'name':'Disabled manual source','kind':'manual','config':{'rows':[]}}); self.assertEqual(status,201)
|
|
status, query=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'disabled-source-run','query':{},'selected_adapters':['manual']}); self.assertEqual(status,201)
|
|
status, job=self.req('POST',f"/api/v1/discovery-queries/{query['id']}/run",{}); self.assertEqual(status,202)
|
|
for _ in range(100):
|
|
_, current=self.req('GET',f"/api/v1/jobs/{job['id']}")
|
|
if current['status'] in ('succeeded','failed'): break
|
|
threading.Event().wait(.01)
|
|
self.assertEqual(current['status'],'failed'); self.assertEqual(current['error_code'],'SOURCE_DISABLED')
|
|
|
|
def test_source_discovery_persists_pipeline_and_is_idempotent(self):
|
|
status, source = self.req('POST', '/api/v1/sources', {
|
|
'name': 'Manual leads', 'kind': 'manual', 'enabled': True,
|
|
'config': {'rows': [{'name': 'Acme Solar', 'website': 'https://acme.test',
|
|
'email': 'hello@acme.test', 'phone': '011 555 0100',
|
|
'description': 'solar installers', 'location': 'Cape Town'}]}})
|
|
self.assertEqual(status, 201)
|
|
payload = {'criteria': {'keywords': ['solar']}, 'selected_adapters': ['manual'],
|
|
'idempotency_key': 'source-run-1', 'max_records': 10}
|
|
status, job = self.req('POST', '/api/v1/discovery', payload)
|
|
self.assertEqual(status, 202)
|
|
for _ in range(100):
|
|
_, current = self.req('GET', f"/api/v1/jobs/{job['id']}")
|
|
if current['status'] in ('succeeded', 'failed'): break
|
|
threading.Event().wait(.01)
|
|
self.assertEqual(current['status'], 'succeeded')
|
|
events = self.req('GET', f"/api/v1/jobs/{job['id']}/events")[1]['items']
|
|
event_types = [event['event_type'] for event in events]
|
|
for stage in ('source.started', 'source.raw_persisted', 'source.normalized',
|
|
'business.created', 'enrichment.queued', 'review.queued', 'discovery.completed'):
|
|
self.assertIn(stage, event_types)
|
|
businesses = self.req('GET', '/api/v1/businesses')[1]['items']
|
|
self.assertEqual(len(businesses), 1)
|
|
detail = self.req('GET', f"/api/v1/businesses/{businesses[0]['id']}")[1]
|
|
self.assertTrue(detail['domains']); self.assertTrue(detail['websites']); self.assertTrue(detail['evidence'])
|
|
self.assertTrue(detail['contacts']); self.assertEqual(detail['review_status'], 'pending')
|
|
self.assertEqual(detail['score_version'], 'opportunity-v1')
|
|
self.assertTrue(all({'code', 'name', 'points', 'version'} <= set(item) for item in detail['score_factors']))
|
|
db = sqlite3.connect(self.tmp.name + '/x.db')
|
|
self.assertEqual(db.execute('SELECT processing_status FROM source_records').fetchone()[0], 'processed')
|
|
self.assertEqual(db.execute('SELECT status FROM enrichment_queue').fetchone()[0], 'completed')
|
|
db.close()
|
|
status, second = self.req('POST', '/api/v1/discovery', {**payload, 'idempotency_key': 'source-run-2'})
|
|
self.assertEqual(status, 202)
|
|
for _ in range(100):
|
|
_, current = self.req('GET', f"/api/v1/jobs/{second['id']}")
|
|
if current['status'] in ('succeeded', 'failed'): break
|
|
threading.Event().wait(.01)
|
|
self.assertEqual(current['status'], 'succeeded')
|
|
self.assertEqual(len(self.req('GET', '/api/v1/businesses')[1]['items']), 1)
|
|
self.assertEqual(self.req('GET', '/api/v1/source-records')[1]['items'].__len__(), 1)
|
|
|
|
def test_source_limits_and_run_lifecycle_are_enforced(self):
|
|
status, source = self.req('POST', '/api/v1/sources', {'name': 'Limited', 'kind': 'manual', 'enabled': True,
|
|
'config': {'rows': [{'name': 'A'}, {'name': 'B'}]}, 'quota': {'daily_limit': 1, 'per_run_limit': 1}})
|
|
self.assertEqual(status, 201)
|
|
status, job = self.req('POST', '/api/v1/discovery', {'criteria': {}, 'selected_adapters': ['manual'],
|
|
'idempotency_key': 'limited-1', 'max_records': 10, 'daily_limit': 1})
|
|
self.assertEqual(status, 202)
|
|
runs = self.req('GET', '/api/v1/discovery-runs')[1]['items']; rid = runs[0]['id']
|
|
self.assertEqual(self.req('POST', f'/api/v1/discovery-runs/{rid}/pause', {})[0], 200)
|
|
self.assertEqual(self.req('POST', f'/api/v1/discovery-runs/{rid}/resume', {})[0], 200)
|
|
self.assertEqual(self.req('POST', f'/api/v1/discovery-runs/{rid}/cancel', {})[0], 200)
|
|
db = sqlite3.connect(self.tmp.name + '/x.db')
|
|
self.assertEqual(db.execute("SELECT lifecycle FROM discovery_runs WHERE id=?", (rid,)).fetchone()[0], 'cancelled')
|
|
self.assertIn(db.execute("SELECT status FROM jobs WHERE id=?", (job['id'],)).fetchone()[0], ('cancelled', 'succeeded'))
|
|
db.close()
|
|
|
|
if __name__=='__main__': unittest.main()
|