"""Reviewed hardware intake: source quantities, serials, provenance and retries.""" import asyncio from contextlib import contextmanager from decimal import Decimal import os import unittest from unittest.mock import AsyncMock, Mock, patch from uuid import uuid4 from fastapi import HTTPException, UploadFile from io import BytesIO from app.modules.hardware.backend import intake class IntakeTests(unittest.TestCase): def test_lines_keep_zero_and_do_not_guess_missing_values(self): result = intake._lines([{'id':7,'line_number':1,'description':'PC','quantity':2,'unit_price':0}])[0] self.assertEqual(result['key'], '7') self.assertEqual(result['quantity'], '2') self.assertEqual(result['unit_price'], '0') self.assertIsNone(intake._lines([{'description':'PC'}])[0]['quantity']) def test_file_line_keys_remain_stable_after_extraction_gets_database_ids(self): before = intake._lines([{'line_number':2,'description':'PC'}], use_line_numbers=True)[0] after = intake._lines([{'id':900,'line_number':2,'description':'PC'}], use_line_numbers=True)[0] self.assertEqual(before['key'], after['key']) def test_duplicate_source_line_numbers_are_rejected(self): with self.assertRaises(HTTPException): intake._lines([{'line_number':1},{'line_number':1}], use_line_numbers=True) def test_invoice_requires_preview_and_barcode_requires_lookup(self): request = intake.IntakeInput(request_id=uuid4(),mode='barcode',hardware={'asset_type':'pc'}) with self.assertRaises(HTTPException) as error: asyncio.run(intake.create_from_intake(request)) self.assertEqual(error.exception.status_code, 422) def test_repeated_serials_rejected_before_writes(self): request = intake.IntakeInput(request_id=uuid4(),hardware={'asset_type':'pc'},count=2,serial_numbers=['SN-123','sn123']) with self.assertRaises(HTTPException): asyncio.run(intake.create_from_intake(request)) def test_upload_reuses_supplier_file_upload_and_returns_review(self): from app.billing.backend import supplier_invoices preview = {'preview_id':'test','lines':[{'description':'PC','quantity':'1'}]} with patch.object(supplier_invoices, 'upload_supplier_invoice', AsyncMock(return_value={'file_id':42})), \ patch.object(intake, '_file_preview', AsyncMock(return_value=preview)) as parse: result = asyncio.run(intake.upload_preview(UploadFile(filename='invoice.pdf',file=BytesIO(b'test fixture')))) self.assertEqual(result, preview) parse.assert_awaited_once_with(42) @unittest.skipUnless(os.environ.get('BMC_HARDWARE_INTAKE_DB_TESTS') == '1', 'Opt-in rollback-only DB tests') class IntakeDatabaseTests(unittest.TestCase): def test_invoice_units_retry_partial_creation_and_atomic_serial_failure(self): import psycopg2 from psycopg2.extras import RealDictCursor from app.core.config import settings from app.modules.hardware.backend import router as hardware conn = psycopg2.connect(settings.DATABASE_URL) def query(sql, params=None): with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(sql, params) return list(cur.fetchall()) if cur.description else cur.rowcount @contextmanager def transaction(): query('SAVEPOINT intake_test_request') try: yield conn except BaseException: query('ROLLBACK TO SAVEPOINT intake_test_request') raise finally: query('RELEASE SAVEPOINT intake_test_request') token=uuid4().hex try: with patch.object(intake,'execute_query',query), patch.object(intake,'atomic_transaction',transaction), patch.object(hardware,'execute_query',query): invoice = query("INSERT INTO supplier_invoices(invoice_number,vendor_name,invoice_date,status) VALUES(%s,'Hardware intake test',CURRENT_DATE,'pending') RETURNING id", (token,))[0]['id'] line_id = query("INSERT INTO supplier_invoice_lines(supplier_invoice_id,line_number,description,quantity,unit_price,line_total) VALUES(%s,1,'Test PC',3,100,300) RETURNING id", (invoice,))[0]['id'] preview = intake._invoice_preview(invoice) request = intake.IntakeInput(request_id=uuid4(),mode='invoice',preview_id=preview['preview_id'],line_key=str(line_id), count=2,serial_numbers=[token+'-a',token+'-b'],approved_unit_cost=Decimal('100'), hardware={'asset_type':'pc','model':'Test PC '+token,'current_owner_type':'bmc','bmc_asset_category':'internal'}) result = asyncio.run(intake.create_from_intake(request)) ids=result['hardware_ids'] self.assertEqual(len(ids),2) saved = query('SELECT id,serial_number,hardware_specs FROM hardware_assets WHERE id=ANY(%s)', (ids,)) self.assertEqual(len({row['serial_number'] for row in saved}),2) self.assertEqual(saved[0]['hardware_specs']['acquisition']['approved_unit_cost'],'100') self.assertEqual(query('SELECT status FROM supplier_invoices WHERE id=%s',(invoice,))[0]['status'],'pending') self.assertEqual(len(query("SELECT id FROM supplier_invoice_relations WHERE supplier_invoice_id=%s AND relation_type='asset'", (invoice,))),2) retry=asyncio.run(intake.create_from_intake(request)) self.assertTrue(retry['already_created']) self.assertEqual(retry['hardware_ids'], sorted(ids)) remaining = intake._invoice_preview(invoice) self.assertEqual(len(remaining['created_units'][str(line_id)]),2) request2=request.model_copy(update={'request_id':uuid4(),'count':1,'serial_numbers':[token+'-c']}) self.assertEqual(len(asyncio.run(intake.create_from_intake(request2))['hardware_ids']),1) with self.assertRaises(HTTPException): asyncio.run(intake.create_from_intake(request.model_copy(update={'request_id':uuid4()}))) # A manual batch fails as one transaction if a later serial already exists. before=query('SELECT COUNT(*) AS n FROM hardware_assets')[0]['n'] bad=intake.IntakeInput(request_id=uuid4(),count=2,serial_numbers=[token+'-new',token+'-a'],hardware={'asset_type':'pc','model':'Rollback test'}) with self.assertRaises(HTTPException): asyncio.run(intake.create_from_intake(bad)) self.assertEqual(query('SELECT COUNT(*) AS n FROM hardware_assets')[0]['n'],before) # Changed source quantity cannot be approved from an old preview. query('UPDATE supplier_invoice_lines SET quantity=4 WHERE id=%s',(line_id,)) with self.assertRaises(HTTPException) as stale: asyncio.run(intake.create_from_intake(request.model_copy(update={'request_id':uuid4(),'count':1,'serial_numbers':[]}))) self.assertEqual(stale.exception.status_code,409) # Manual and barcode creation also preserve their source and hardware class. for mode in ('manual','barcode'): simple=intake.IntakeInput(request_id=uuid4(),mode=mode,code='5700000000000' if mode=='barcode' else None, hardware={'asset_type':'pc','model':token+mode,'current_owner_type':'bmc','bmc_asset_category':'rental'}) asset_id=asyncio.run(intake.create_from_intake(simple))['hardware_ids'][0] row=query('SELECT bmc_asset_category,hardware_specs FROM hardware_assets WHERE id=%s',(asset_id,))[0] self.assertEqual(row['bmc_asset_category'],'rental') self.assertEqual(row['hardware_specs']['acquisition']['mode'],mode) finally: conn.rollback() conn.close() def test_real_pdf_upload_review_and_reupload_do_not_duplicate_units(self): import tempfile import psycopg2 from psycopg2.extras import RealDictCursor from reportlab.pdfgen import canvas from app.core.config import settings from app.modules.hardware.backend import router as hardware from app.billing.backend import supplier_invoices from app.services import invoice2data_service from app.services.ollama_service import ollama_service conn = psycopg2.connect(settings.DATABASE_URL) def query(sql, params=None): with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(sql, params) return list(cur.fetchall()) if cur.description else cur.rowcount def one(sql, params=None): rows=query(sql,params) return rows[0] if rows else None @contextmanager def transaction(): query('SAVEPOINT pdf_intake_request') try: yield conn except BaseException: query('ROLLBACK TO SAVEPOINT pdf_intake_request') raise finally: query('RELEASE SAVEPOINT pdf_intake_request') token=uuid4().hex pdf=BytesIO() document=canvas.Canvas(pdf) document.drawString(50,750,'Hardware fixture '+token) document.drawString(50,720,'PC 2 x 100 DKK') document.save() def extract(text): self.assertIn('Hardware fixture',text) return {'invoice_number':token,'vendor_name':'Hardware fixture','currency':'DKK', 'lines':[{'line_number':1,'description':'PC','quantity':2,'unit_price':100}]} parser=Mock(extract=Mock(side_effect=extract)) try: with tempfile.TemporaryDirectory() as folder, patch.object(settings,'UPLOAD_DIR',folder), \ patch.object(intake,'execute_query',query), patch.object(intake,'atomic_transaction',transaction), \ patch.object(hardware,'execute_query',query), patch.object(supplier_invoices,'execute_query_single',one), \ patch.object(invoice2data_service,'get_invoice2data_service',return_value=parser), \ patch.object(ollama_service,'extract_from_text',AsyncMock(side_effect=AssertionError('Template fixture must not invoke AI'))): preview=asyncio.run(intake.upload_preview(UploadFile(filename=token+'.pdf',file=BytesIO(pdf.getvalue())))) self.assertEqual(preview['source_kind'],'file') self.assertEqual(preview['lines'][0]['quantity'],'2') request=intake.IntakeInput(request_id=uuid4(),mode='invoice',preview_id=preview['preview_id'],line_key='1',count=2, serial_numbers=[token+'X',token+'Y'],hardware={'asset_type':'pc','model':'PDF fixture','current_owner_type':'bmc','bmc_asset_category':'rental'}) self.assertEqual(len(asyncio.run(intake.create_from_intake(request))['hardware_ids']),2) repeated=asyncio.run(intake.upload_preview(UploadFile(filename=token+'.pdf',file=BytesIO(pdf.getvalue())))) self.assertEqual(repeated['source_id'],preview['source_id']) self.assertEqual(len(repeated['created_units']['1']),2) invoice=one("INSERT INTO supplier_invoices(invoice_number,vendor_name,invoice_date,file_path,status) SELECT %s,'Hardware fixture',CURRENT_DATE,file_path,'pending' FROM incoming_files WHERE file_id=%s RETURNING id", (token,preview['source_id']))['id'] query("INSERT INTO supplier_invoice_lines(supplier_invoice_id,line_number,description,quantity,unit_price) VALUES(%s,1,'PC',2,100)", (invoice,)) registered=intake._invoice_preview(invoice) self.assertEqual(len(registered['created_units']['1']),2) with self.assertRaises(HTTPException): asyncio.run(intake.create_from_intake(request.model_copy(update={'request_id':uuid4(),'preview_id':registered['preview_id']}))) finally: conn.rollback() conn.close() if __name__ == '__main__': unittest.main()