"""Database and HTTP integration tests. Writes verification.json after the run.""" import concurrent.futures import http.cookiejar import json import tempfile import threading import time import unittest import urllib.error import urllib.request from pathlib import Path from app import Application, Catalog, DomainError, ROOT HEADER='part_number,description,price,currency,unit\n' def source(price='13.25',part='BRG-10'): return HEADER+f'{part},Bearing,{price},USD,EA\nSHIM-2,Shim,2.10,USD,EA\n' class DatabaseTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory();self.catalog=Catalog(Path(self.temp.name)/'catalog.db');self.catalog.seed() def tearDown(self): self.temp.cleanup() def staged(self,text=None): bid=self.catalog.stage('ACME',text or source())['id'];return self.catalog.batch(bid) def approve(self,b,**kw): return self.catalog.approve(b['id'],b['base_version'],b['review_digest'],**kw) def test_quoted_csv_aliases_diff_and_missing_rows(self): b=self.staged((ROOT/'fixtures/price-update.csv').read_text()) self.assertEqual(b['summary'],{'added':1,'changed':1,'unchanged':1,'missing':1,'rejected':0}) self.approve(b) parts={x['part_number']:x for x in self.catalog.items('ACME')['items']} self.assertEqual(parts['BRG-10']['description'],'Bearing, 10 mm') self.assertEqual(parts['BRG-10']['price_cents'],1325) self.assertIn('SEAL-5',parts) def test_duplicate_normalized_keys_reject_both_and_block_batch(self): b=self.staged((ROOT/'fixtures/rejected.csv').read_text()) self.assertEqual(b['status'],'blocked') self.assertEqual({e['row'] for e in b['errors']},{2,3,4,5}) with self.assertRaises(DomainError): self.approve(b) self.assertEqual(self.catalog.summary()['suppliers'][0]['version'],0) def test_explicit_european_decimal_mapping(self): bid=self.catalog.stage('EURO',(ROOT/'fixtures/euro-prices.csv').read_text(),'de-DE')['id'];b=self.catalog.batch(bid) self.assertEqual(b['status'],'staged');self.approve(b) self.assertEqual(self.catalog.items('EURO')['items'][1]['price_cents'],123450) def test_ambiguous_decimal_and_extra_precision_rejected(self): for amount in ['1,23','1.234','-2','NaN','1e5','$2.00']: with self.subTest(amount=amount): rows,errors,_=self.catalog.parse(source(amount),'en-US') self.assertTrue(errors) def test_source_digest_is_idempotent_and_bytes_preserved(self): text=source();first=self.catalog.stage('ACME',text);second=self.catalog.stage('ACME',text) self.assertEqual(first['id'],second['id']);self.assertTrue(second['reused']) self.assertEqual(self.catalog.source(first['id']),text) self.assertEqual(len(self.catalog.summary()['batches']),1) def test_response_loss_retry_returns_committed_result_once(self): b=self.staged();result=self.approve(b);again=self.approve(b) self.assertEqual(result['version'],again['version']);self.assertTrue(again['replayed']) self.assertEqual(sum(x['action']=='applied' for x in self.catalog.batch(b['id'])['audit']),1) def test_mid_batch_fault_rolls_back_items_version_and_audit(self): before=self.catalog.items('ACME');b=self.staged() with self.assertRaises(RuntimeError): self.approve(b,fail_after=1) self.assertEqual(before,self.catalog.items('ACME')) self.assertEqual(self.catalog.summary()['suppliers'][0]['version'],0) self.assertEqual(self.catalog.batch(b['id'])['status'],'staged') self.assertEqual(len(self.catalog.batch(b['id'])['audit']),1) self.approve(b) def test_concurrent_reviews_one_wins_other_is_stale(self): a=self.staged(source('14.00'));b=self.staged(source('15.00'));barrier=threading.Barrier(2) def submit(batch): barrier.wait() try:return self.approve(batch) except DomainError as exc:return {'error':str(exc),'status':exc.status} with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: results=list(pool.map(submit,[a,b])) self.assertEqual(sum('version' in x for x in results),1) self.assertEqual(sum(x.get('status')==409 for x in results),1) self.assertEqual(self.catalog.summary()['suppliers'][0]['version'],1) def test_refresh_changes_review_digest_and_requires_new_review(self): a=self.staged(source('14.00'));old=self.staged(source('15.00'));self.approve(a) refreshed=self.catalog.refresh(old['id']) self.assertNotEqual(old['review_digest'],refreshed['review_digest']) with self.assertRaises(DomainError): self.approve(old) self.approve(refreshed) self.assertEqual(self.catalog.items('ACME')['items'][1]['price_cents'],1500) def test_duplicate_headers_and_unclosed_quote_are_blocked(self): for text in ['part,part,description,price,currency,unit\nA,B,x,1,USD,EA\n',HEADER+'A,"unclosed,1,USD,EA']: with self.subTest(text=text): b=self.staged(text);self.assertEqual(b['status'],'blocked') def test_newlines_in_quoted_description_are_retained(self): b=self.staged(HEADER+'A,"Line one\nLine two",1.00,USD,EA\n') self.assertEqual(b['changes'][0]['after']['description'],'Line one\nLine two') def test_full_report_includes_all_pages(self): b=self.staged();page=self.catalog.batch(b['id'],size=1);report=self.catalog.report(b['id']) self.assertEqual(len(page['changes']),1) self.assertEqual(len(report['changes']),report['total']) def test_invalid_unicode_is_rejected_without_server_error(self): with self.assertRaises(DomainError): self.catalog.parse('\ud800','en-US') def test_currency_and_unit_change_remain_explicit_in_diff(self): b=self.staged(HEADER+'BRG-10,Bearing,13.25,EUR,box\n') change=b['changes'][0] self.assertEqual(change['before']['currency'],'USD');self.assertEqual(change['after']['currency'],'EUR') self.assertEqual(change['after']['unit'],'BOX') class HttpTests(DatabaseTests): # Do not rerun inherited DB cases in this suite. def setUp(self): super().setUp();self.server=Application(('127.0.0.1',0),self.catalog) self.thread=threading.Thread(target=self.server.serve_forever,daemon=True);self.thread.start() self.base=f'http://127.0.0.1:{self.server.server_port}';self.opener=urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()));self.csrf='' def tearDown(self): self.server.shutdown();self.server.server_close();self.thread.join();super().tearDown() def request(self,path,body=None,headers=None): h={'Content-Type':'application/json','X-CSRF-Token':self.csrf};h.update(headers or {}) req=urllib.request.Request(self.base+path,data=json.dumps(body).encode() if body is not None else None,headers=h) try: with self.opener.open(req) as r:return r.status,r.read(),dict(r.headers) except urllib.error.HTTPError as e:return e.code,e.read(),dict(e.headers) def login(self,who='reviewer'): password={'reviewer':'review-demo','importer':'import-demo','viewer':'view-demo'}[who] status,body,_=self.request('/api/login',{'username':who,'password':password});self.assertEqual(status,200);self.csrf=json.loads(body)['csrf'] def test_http_full_review_and_download_workflow(self): self.login();status,body,_=self.request('/api/stage',{'supplier':'ACME','csv':source()});self.assertEqual(status,201);bid=json.loads(body)['id'] _,raw,_=self.request('/api/batches/'+bid);b=json.loads(raw) status,raw,_=self.request('/api/batches/'+bid+'/approve',{'expected_version':b['base_version'],'review_digest':b['review_digest']});self.assertEqual(status,200) status,raw,headers=self.request('/api/batches/'+bid+'/report');self.assertEqual(status,200);self.assertEqual(json.loads(raw)['status'],'applied');self.assertIn('attachment',headers['Content-Disposition']) def test_http_reader_cannot_stage_or_approve(self): b=self.staged();self.login('viewer') self.assertEqual(self.request('/api/stage',{'supplier':'ACME','csv':source()})[0],403) self.assertEqual(self.request('/api/batches/'+b['id']+'/approve',{'expected_version':0,'review_digest':b['review_digest']})[0],403) def test_http_importer_cannot_approve(self): b=self.staged();self.login('importer') self.assertEqual(self.request('/api/batches/'+b['id']+'/approve',{'expected_version':0,'review_digest':b['review_digest']})[0],403) def test_http_authentication_csrf_origin_and_host_checks(self): self.assertEqual(self.request('/api/summary')[0],401);self.login() self.assertEqual(self.request('/api/stage',{'supplier':'ACME','csv':source()}, {'X-CSRF-Token':'wrong'})[0],403) self.assertEqual(self.request('/api/stage',{'supplier':'ACME','csv':source()}, {'Origin':'https://example.com'})[0],403) self.assertEqual(self.request('/api/summary',headers={'Host':'evil.example'})[0],403) def test_http_path_traversal_is_not_a_file_route(self): self.login();self.assertEqual(self.request('/../../app.py')[0],404) def test_http_static_page_and_malformed_input(self): status,raw,headers=self.request('/');self.assertEqual(status,200);self.assertIn(b'Supplier catalog review',raw) self.assertEqual(self.request('/api/login',{'username':[], 'password':'x'})[0],401) self.login();self.assertEqual(self.request('/api/stage',{'supplier':'bad supplier','csv':source()})[0],422) self.assertEqual(self.request('/api/catalog?page=oops')[0],422) if __name__=='__main__': suite=unittest.TestSuite() suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(DatabaseTests)) suite.addTests(HttpTests(name) for name in unittest.defaultTestLoader.getTestCaseNames(HttpTests) if name.startswith('test_http_')) names=[str(test) for test in suite] started=time.perf_counter();result=unittest.TextTestRunner(verbosity=2).run(suite) payload={'suite':'supplier-catalog-imports','tests':result.testsRun,'passed':result.testsRun-len(result.failures)-len(result.errors),'failures':len(result.failures),'errors':len(result.errors),'elapsed_seconds':round(time.perf_counter()-started,3),'cases':names,'command':'python tests.py'} (ROOT/'verification.json').write_text(json.dumps(payload,indent=2)+'\n') raise SystemExit(not result.wasSuccessful())