36 lines
3.4 KiB
Python
36 lines
3.4 KiB
Python
import json, os, sqlite3, threading, unittest
|
|
from http.client import HTTPConnection
|
|
from tempfile import TemporaryDirectory
|
|
from app.main import create_server
|
|
|
|
class Phase6ApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=TemporaryDirectory(); os.environ['BOOTSTRAP_ADMIN_EMAIL']='owner@phase6.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=3); self.cookie=None
|
|
self.request('POST','/api/v1/auth/login',{'email':'owner@phase6.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); r=self.conn.getresponse(); c=r.getheader('Set-Cookie');
|
|
if c: self.cookie=c.split(';',1)[0]
|
|
return r.status,json.loads(r.read() or b'{}')
|
|
def test_merge_preserves_children_and_reverse_restores_ownership(self):
|
|
_, source=self.request('POST','/api/v1/businesses',{'name':'Acme Consulting','phone':'082 555 1234'})
|
|
_, target=self.request('POST','/api/v1/businesses',{'name':'Acme Consulting HQ','city':'Johannesburg'})
|
|
self.assertEqual(self.request('POST',f"/api/v1/businesses/{source['id']}/notes",{'body':'evidence'})[0],201)
|
|
self.assertEqual(self.request('POST',f"/api/v1/businesses/{source['id']}/evidence",{'kind':'source','claim':'claim'})[0],201)
|
|
status, merged=self.request('POST',f"/api/v1/businesses/{source['id']}/merge",{'target_business_id':target['id']})
|
|
self.assertEqual(status,200); self.assertEqual(merged['status'],'merged')
|
|
status, detail=self.request('GET',f"/api/v1/businesses/{target['id']}"); self.assertEqual(status,200); self.assertEqual(len(detail['notes']),1); self.assertEqual(len(detail['evidence']),1)
|
|
status, source_detail=self.request('GET',f"/api/v1/businesses/{source['id']}"); self.assertEqual(status,200); self.assertEqual(source_detail['merge_status'],'merged')
|
|
status, reversed_=self.request('POST',f"/api/v1/merge-history/{merged['merge_history_id']}/reverse",{}); self.assertEqual(status,200); self.assertEqual(reversed_['status'],'reversed')
|
|
_, restored=self.request('GET',f"/api/v1/businesses/{source['id']}"); self.assertEqual(len(restored['notes']),1); self.assertEqual(len(restored['evidence']),1)
|
|
def test_matches_endpoint_is_thresholded_and_history_is_tenant_scoped(self):
|
|
_, a=self.request('POST','/api/v1/businesses',{'name':'Bright Co','email':'hello@bright.test'})
|
|
_, b=self.request('POST','/api/v1/businesses',{'name':'Bright Company','website':'https://bright.test'})
|
|
status, matches=self.request('GET',f"/api/v1/businesses/{a['id']}/matches?threshold=0.4"); self.assertEqual(status,200); self.assertEqual(matches['items'][0]['id'],b['id']); self.assertIn('similar_name',matches['items'][0]['reasons'])
|
|
self.assertEqual(self.request('GET','/api/v1/merge-history')[0],200)
|
|
|
|
if __name__=='__main__': unittest.main()
|