diff --git a/apps/api/app/main.py b/apps/api/app/main.py index af1227d..f91aee4 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -78,7 +78,7 @@ def connect(db_path: str) -> sqlite3.Connection: "sources": (("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 '{}'")), "discovery_queries": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'")), "discovery_runs": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'"), ("paused_at", "TEXT")), - "source_records": (("discovery_run_id", "INTEGER"), ("normalized_key", "TEXT NOT NULL DEFAULT ''"), ("provenance_json", "TEXT NOT NULL DEFAULT '{}'")), + "source_records": (("discovery_run_id", "INTEGER"), ("normalized_key", "TEXT NOT NULL DEFAULT ''"), ("provenance_json", "TEXT NOT NULL DEFAULT '{}'"), ("response_metadata_json", "TEXT NOT NULL DEFAULT '{}'")), }.items(): existing = {r[1] for r in db.execute(f"PRAGMA table_info({table})")} for col, definition in additions: @@ -1322,7 +1322,7 @@ class ApiHandler(BaseHTTPRequestHandler): rows=db.execute("SELECT * FROM source_records WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?",(org,limit+1,offset)).fetchall(); out=[] for r in rows[:limit]: x=row_json(r) - for k in ('raw_json','normalized_json','query_context_json','cursor_json','rate_policy_json'): + for k in ('raw_json','normalized_json','query_context_json','response_metadata_json','cursor_json','rate_policy_json'): try:x[k]=json.loads(x[k]) except (ValueError,TypeError):pass out.append(x) @@ -1334,7 +1334,9 @@ class ApiHandler(BaseHTTPRequestHandler): if contains_secret(config):return self.send_json(400,{"error":"secret_not_permitted"}) try: validation=adapter_for(kind).validate_config(config) - if (config or optional) and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors}) + # Registration may precede the actual local payload or optional provider + # approval. Discovery/ingest still validates the effective configuration. + if config and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors}) cur=db.execute("INSERT INTO sources(organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json) VALUES(?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],name,kind,kind,str(payload.get('display_name') or adapter_for(kind).display_name),int(bool(payload.get('enabled',False))),int(bool(payload.get('approved',config.get('approved',False)))),json.dumps(config,sort_keys=True),json.dumps(payload.get('policy',{}),sort_keys=True),json.dumps(payload.get('quota',{}),sort_keys=True))) except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_source"}) self.audit(db,user,'source.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE id=?",(cur.lastrowid,)).fetchone())) @@ -1355,8 +1357,14 @@ class ApiHandler(BaseHTTPRequestHandler): except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_query"}) self.audit(db,user,'discovery_query.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT * FROM discovery_queries WHERE id=?",(cur.lastrowid,)).fetchone())) def run_query(self,qid,db,user): - if not db.execute("SELECT id FROM discovery_queries WHERE id=? AND organization_id=?",(qid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"}) - return self.create_job({"type":"source_discovery","_accepted":True,"payload":{"discovery_query_id":qid},"idempotency_key":f"discovery-query-{qid}-{int(time.time())}"},db,user) + query=db.execute("SELECT * FROM discovery_queries WHERE id=? AND organization_id=?",(qid,user["organization_id"])).fetchone() + if not query:return self.send_json(404,{"error":"not_found"}) + key=f"discovery-query-{qid}-{int(time.time())}" + result=self.create_job({"type":"source_discovery","_accepted":True,"payload":{"discovery_query_id":qid,"selected_adapters":json.loads(query["selected_adapters_json"] or "[]"),"max_records":query["max_records"],"daily_limit":query["daily_limit"]},"idempotency_key":key,"_defer_wakeup":True},db,user) + job=db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?",(user["organization_id"],key)).fetchone() + if not db.execute("SELECT id FROM discovery_runs WHERE organization_id=? AND job_id=?",(user["organization_id"],job["id"])).fetchone(): + db.execute("INSERT INTO discovery_runs(organization_id,job_id,selected_adapters_json,location,category,max_records,daily_limit,schedule,dry_run,lifecycle,criteria_json,seed_urls_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",(user["organization_id"],job["id"],query["selected_adapters_json"],query["location"],query["category"],query["max_records"],query["daily_limit"],query["schedule"],query["dry_run"],"queued",query["query_json"],"[]")); db.commit() + getattr(self.server,"job_wakeup",threading.Event()).set(); return result def test_source(self,sid,db,user): source=db.execute("SELECT * FROM sources WHERE id=? AND organization_id=?",(sid,user['organization_id'])).fetchone() if not source:return self.send_json(404,{"error":"not_found"}) @@ -1486,19 +1494,66 @@ def _run_scoped_discovery(db, job, handler): def _run_source_discovery(db, job, handler): - payload=json.loads(job["payload"] or "{}"); org=job["organization_id"]; run=db.execute("SELECT * FROM discovery_runs WHERE organization_id=? AND job_id=?",(org,job["id"])).fetchone(); selected=payload.get("selected_adapters") or [] - sources=db.execute("SELECT * FROM sources WHERE organization_id=? AND enabled=1 AND (source_code IN ("+(','.join('?'*len(selected)) or "NULL")+") OR kind IN ("+(','.join('?'*len(selected)) or "NULL")+"))",[org]+list(selected)+list(selected)).fetchall() if selected else [] - total=0 + payload=json.loads(job["payload"] or "{}"); org=job["organization_id"] + run=db.execute("SELECT * FROM discovery_runs WHERE organization_id=? AND job_id=?",(org,job["id"])).fetchone() + selected=payload.get("selected_adapters") or [] + query=None + if payload.get("discovery_query_id"): + query=db.execute("SELECT * FROM discovery_queries WHERE id=? AND organization_id=?",(payload["discovery_query_id"],org)).fetchone() + if query: + selected=json.loads(query["selected_adapters_json"] or "[]") + if not selected: + linked=db.execute("SELECT kind FROM sources WHERE id=? AND organization_id=?",(query["source_id"],org)).fetchone() + selected=[linked["kind"]] if linked else [] + placeholders=','.join('?'*len(selected)) or "NULL" + sources=db.execute("SELECT * FROM sources WHERE organization_id=? AND enabled=1 AND circuit_open=0 AND (source_code IN ("+placeholders+") OR kind IN ("+placeholders+"))",[org]+list(selected)+list(selected)).fetchall() if selected else [] + total=0; max_records=max(0,int(payload.get("max_records", query["max_records"] if query else 100))) for source in sources: - try: config=json.loads(source["config_json"] or "{}"); page=adapter_for(source["kind"]).discover(config) + handler.add_job_event(db,job["id"],org,"source.started",f"Starting {source['display_name'] or source['kind']}",5) + try: + config=json.loads(source["config_json"] or "{}") + if query and query["source_id"]==source["id"]: + config.update(json.loads(query["query_json"] or "{}")) + quota=json.loads(source["quota_json"] or "{}") + daily_limit=int(quota.get("daily_limit", payload.get("daily_limit", 100000))) + per_run_limit=int(quota.get("per_run_limit", max_records)) + used_today=db.execute("SELECT COUNT(*) FROM source_records WHERE organization_id=? AND source_id=? AND date(created_at)=date('now')",(org,source["id"])).fetchone()[0] + if used_today >= daily_limit or per_run_limit <= 0: + handler.add_job_event(db,job["id"],org,"source.quota_exceeded",f"Quota reached for {source['kind']}",10,"SOURCE_QUOTA_EXCEEDED"); continue + page=adapter_for(source["kind"]).discover(config) except Exception as exc: - handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['kind']} unavailable",0,"SOURCE_NOT_CONFIGURED"); continue - for record in page.records[:max(0,int(payload.get("max_records",100))-total)]: - raw=json.dumps(record,sort_keys=True,separators=(",",":")); digest=hashlib.sha256(raw.encode()).hexdigest(); norm=json.dumps(normalize_record(record),sort_keys=True); nkey=hashlib.sha256(norm.encode()).hexdigest() + failures=int(source["consecutive_failures"])+1 + db.execute("UPDATE sources SET health_status='unhealthy',consecutive_failures=?,circuit_open=?,last_failure_at=CURRENT_TIMESTAMP,last_error=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(failures,int(circuit_is_open(failures)),str(exc)[:300],source["id"])) + handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['kind']} unavailable",10,"SOURCE_NOT_CONFIGURED"); continue + limit=min(max_records-total, per_run_limit, max(0,daily_limit-used_today)) + for record in page.records[:limit]: + raw=json.dumps(record,sort_keys=True,separators=(",",":")); digest=hashlib.sha256(raw.encode()).hexdigest(); normalized=normalize_business(normalize_record(record)); norm=json.dumps(normalized,sort_keys=True); nkey=hashlib.sha256(norm.encode()).hexdigest() + handler.add_job_event(db,job["id"],org,"source.raw_persisted",f"Persisting {source['kind']} record",20) try: - cur=db.execute("INSERT INTO source_records(organization_id,source_id,discovery_run_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json) VALUES(?,?,?,?,?,?,?,?,?)",(org,source["id"],run["id"] if run else None,digest,raw,norm,nkey,str(config.get("source_url","")),json.dumps({"adapter":source["kind"]}))) - db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id) VALUES(?,?)",(org,cur.lastrowid)); total+=1 - except sqlite3.IntegrityError: pass + cur=db.execute("INSERT INTO source_records(organization_id,source_id,discovery_query_id,discovery_run_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json,query_context_json,response_metadata_json,processing_status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)",(org,source["id"],query["id"] if query else None,run["id"] if run else None,digest,raw,norm,nkey,str(config.get("source_url","")),json.dumps({"adapter":source["kind"],"metadata":page.metadata},sort_keys=True),json.dumps(payload.get("criteria",{}),sort_keys=True),json.dumps(page.metadata,sort_keys=True),"raw")) + record_id=cur.lastrowid; total+=1 + except sqlite3.IntegrityError: + existing=db.execute("SELECT id,processing_status FROM source_records WHERE organization_id=? AND source_id=? AND content_hash=?",(org,source["id"],digest)).fetchone() + if existing and existing["processing_status"] in ("processed","matched"): continue + record_id=existing["id"] if existing else None + handler.add_job_event(db,job["id"],org,"source.normalized",f"Normalized {normalized.get('name','')}",35) + suppressed=is_suppressed(normalized,[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1",(org,))]) + if suppressed: + if record_id: db.execute("UPDATE source_records SET processing_status='skipped' WHERE id=? AND organization_id=?",(record_id,org)) + handler.add_job_event(db,job["id"],org,"review.skipped",f"Suppressed source record {record_id}",40,"SUPPRESSED"); continue + existing=db.execute("SELECT * FROM businesses WHERE organization_id=? AND ((website_domain<>'' AND website_domain=?) OR (email<>'' AND email=?) OR (phone<>'' AND phone=?) OR (name=? AND city=?)) ORDER BY id LIMIT 1",(org,normalized["website_domain"],normalized["email"],normalized["phone"],normalized["name"],normalized["city"])).fetchone() + if existing: bid=existing["id"]; handler.add_job_event(db,job["id"],org,"business.matched",f"Matched business {bid}",50) + else: + scored=score_business(normalized); cur=db.execute("INSERT INTO businesses(organization_id,name,website,website_domain,email,phone,description,province,city,suburb,score,score_version,score_factors,website_class,review_status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(org,normalized["name"],normalized["website"],normalized["website_domain"],normalized["email"],normalized["phone"],normalized.get("description",""),normalized["province"],normalized["city"],normalized["suburb"],scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"],"pending")); bid=cur.lastrowid + db.execute("INSERT INTO score_history(organization_id,business_id,score,eligible,priority_band,score_version,explanations_json,signals_json) VALUES(?,?,?,?,?,?,?,?)",(org,bid,scored["score"],1,"high" if scored["score"]>=70 else "medium" if scored["score"]>=40 else "low",scored["score_version"],json.dumps(scored["factors"]),json.dumps({"source":source["kind"]}))) + handler.add_job_event(db,job["id"],org,"business.created",f"Created business {bid}",60) + if normalized["website_domain"] and not db.execute("SELECT 1 FROM domains WHERE organization_id=? AND business_id=? AND domain=?",(org,bid,normalized["website_domain"])).fetchone(): db.execute("INSERT INTO domains(business_id,organization_id,domain,kind) VALUES(?,?,?,?)",(bid,org,normalized["website_domain"],"website")) + if normalized["website"] and not db.execute("SELECT 1 FROM websites WHERE organization_id=? AND business_id=? AND url=?",(org,bid,normalized["website"])).fetchone(): db.execute("INSERT INTO websites(business_id,organization_id,url,website_class) VALUES(?,?,?,?)",(bid,org,normalized["website"],"business_site")) + if normalized["website"] or normalized["website_domain"]: db.execute("INSERT OR IGNORE INTO evidence(business_id,organization_id,kind,url,claim) VALUES(?,?,?,?,?)",(bid,org,"source_record",normalized["website"],"Discovered by "+source["kind"])) + if normalized["email"] or normalized["phone"]: + if not db.execute("SELECT 1 FROM contacts WHERE organization_id=? AND business_id=? AND email=? AND phone=?",(org,bid,normalized["email"],normalized["phone"])).fetchone(): db.execute("INSERT INTO contacts(business_id,organization_id,email,phone) VALUES(?,?,?,?)",(bid,org,normalized["email"],normalized["phone"])) + if record_id: db.execute("UPDATE source_records SET processing_status='processed',normalized_json=?,normalized_key=? WHERE id=? AND organization_id=?",(norm,nkey,record_id,org)); db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id,status) VALUES(?,?,?)",(org,record_id,"completed")); db.execute("UPDATE enrichment_queue SET status='completed',updated_at=CURRENT_TIMESTAMP WHERE source_record_id=? AND organization_id=?",(record_id,org)) + handler.add_job_event(db,job["id"],org,"enrichment.queued",f"Enrichment complete for business {bid}",75); handler.add_job_event(db,job["id"],org,"review.queued",f"Business {bid} queued for review",90) if run: db.execute("UPDATE discovery_runs SET result_count=?,lifecycle='succeeded',updated_at=CURRENT_TIMESTAMP WHERE id=?",(total,run["id"])) handler.add_job_event(db,job["id"],org,"discovery.completed",f"Persisted {total} source records",100) @@ -1525,6 +1580,11 @@ def _job_worker(server): continue if job["type"] == "source_discovery": try: + run_state=db.execute("SELECT lifecycle FROM discovery_runs WHERE organization_id=? AND job_id=?",(org,jid)).fetchone() + if run_state and run_state["lifecycle"] == "cancelled": + db.execute("UPDATE jobs SET status='cancelled',completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,)); server_handler.add_job_event(db,jid,org,"cancelled","Discovery cancelled",job["progress"]); db.commit(); continue + if run_state and run_state["lifecycle"] == "paused": + db.execute("UPDATE jobs SET status='queued',updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,)); server_handler.add_job_event(db,jid,org,"paused","Discovery paused",job["progress"]); db.commit(); continue _run_source_discovery(db, job, server_handler) db.execute("UPDATE jobs SET status='succeeded',progress=100,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (jid,)); db.commit() except Exception as exc: diff --git a/apps/api/app/sources.py b/apps/api/app/sources.py index 0acd32d..23ef93c 100644 --- a/apps/api/app/sources.py +++ b/apps/api/app/sources.py @@ -85,10 +85,16 @@ def backoff_delay(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter return exponential_backoff(attempt, base, maximum, jitter) def circuit_is_open(consecutive_failures: int, threshold: int = 3) -> bool: - return int(consecutive_failures) >= threshold + return int(consecutive_failures) >= max(1, int(threshold)) + +def quota_remaining(used: int, limit: int | None) -> int | None: + """Return remaining quota, clamped so malformed values fail closed.""" + if limit is None: return None + return max(0, int(limit) - max(0, int(used))) def quota_allowed(used: int, limit: int | None) -> bool: - return limit is None or (limit >= 0 and used < limit) + remaining = quota_remaining(used, limit) + return remaining is None or remaining > 0 def rate_limit_delay(last_request: float | None, min_interval: float) -> float: if last_request is None: return 0.0 diff --git a/apps/api/schema.sql b/apps/api/schema.sql index 5f3ce6a..e2e241c 100644 --- a/apps/api/schema.sql +++ b/apps/api/schema.sql @@ -158,11 +158,12 @@ CREATE INDEX IF NOT EXISTS idx_discovery_queries_org ON discovery_queries(organi CREATE TABLE IF NOT EXISTS source_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE, discovery_query_id INTEGER REFERENCES discovery_queries(id) ON DELETE SET NULL, discovery_run_id INTEGER REFERENCES discovery_runs(id) ON DELETE SET NULL, content_hash TEXT NOT NULL, raw_json TEXT NOT NULL, - normalized_json TEXT NOT NULL, normalized_key TEXT NOT NULL DEFAULT '', source_url TEXT NOT NULL DEFAULT '', provenance_json TEXT NOT NULL DEFAULT '{}', query_context_json TEXT NOT NULL DEFAULT '{}', processing_status TEXT NOT NULL DEFAULT 'raw', + normalized_json TEXT NOT NULL, normalized_key TEXT NOT NULL DEFAULT '', source_url TEXT NOT NULL DEFAULT '', provenance_json TEXT NOT NULL DEFAULT '{}', query_context_json TEXT NOT NULL DEFAULT '{}', response_metadata_json TEXT NOT NULL DEFAULT '{}', processing_status TEXT NOT NULL DEFAULT 'raw' CHECK(processing_status IN ('raw','processed','matched','failed','skipped')), cursor_json TEXT NOT NULL DEFAULT '{}', rate_policy_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,source_id,content_hash) ); CREATE INDEX IF NOT EXISTS idx_source_records_org ON source_records(organization_id,id DESC); +CREATE INDEX IF NOT EXISTS idx_source_records_processing ON source_records(organization_id,processing_status,created_at DESC); CREATE TABLE IF NOT EXISTS enrichment_queue ( id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, source_record_id INTEGER NOT NULL REFERENCES source_records(id) ON DELETE CASCADE, status TEXT NOT NULL DEFAULT 'queued', attempts INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,source_record_id) ); diff --git a/apps/api/tests/test_sources_phase5.py b/apps/api/tests/test_sources_phase5.py index a9ecd9c..604ac22 100644 --- a/apps/api/tests/test_sources_phase5.py +++ b/apps/api/tests/test_sources_phase5.py @@ -47,8 +47,80 @@ class SourceApiTests(unittest.TestCase): _,q=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'q','query':{'csv':'name\nA'}}) 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_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_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') + 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()