55 lines
3.9 KiB
Python
55 lines
3.9 KiB
Python
|
import json, os, sqlite3, threading, unittest
|
||
|
|
from http.client import HTTPConnection
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
from app.main import create_server
|
||
|
|
from app.sources import CsvSource, ManualSource
|
||
|
|
|
||
|
|
class SourceAdapterTests(unittest.TestCase):
|
||
|
|
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())
|
||
|
|
|
||
|
|
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})
|
||
|
|
_,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')
|
||
|
|
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)
|
||
|
|
|
||
|
|
if __name__=='__main__': unittest.main()
|