import importlib.util import json from pathlib import Path import sqlite3 import tempfile import unittest from unittest.mock import patch HERE = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('inventory_sync', HERE/'scripts/inventory-sync.py') bridge = importlib.util.module_from_spec(spec) spec.loader.exec_module(bridge) class Tests(unittest.TestCase): def setUp(self): self.report = { 'system_uuid':'12345678-1234-1234-1234-123456789012','asset_id':'PC-0001','cpu':'AMD Ryzen 5 PRO 3500U','model':'HP EliteBook 745 G6', 'ram_gib':13.54,'installed_ram_gib':16.0, 'disks':[{'name':'sda','removable':False,'rotational':False,'size_gib':119}, {'name':'sdb','removable':True,'rotational':True,'size_gib':28}] } def test_maps_physical_ram_and_excludes_usb(self): self.assertEqual(bridge.map_fields(self.report), {'cpu':'AMD Ryzen 5 PRO 3500U','model':'HP EliteBook 745 G6','ram_gb':16,'ssd_gb':128}) def test_hardware_identity_stability_and_fallback(self): a = {**self.report, 'asset_id': '45'} b = {**a, 'asset_id': '69', 'installed_ram_gib': 64, 'disks': []} self.assertEqual(bridge.hardware_id(a), bridge.hardware_id(b)) self.assertEqual(bridge.hardware_id(a), 'dmi-uuid:12345678-1234-1234-1234-123456789012') serial = {'manufacturer': ' HP ', 'serial_number': ' abc123 '} self.assertEqual(bridge.hardware_id(serial), bridge.hardware_id({'manufacturer':'hp','serial_number':'ABC123'})) for data in [{}, {'system_uuid':'00000000-0000-0000-0000-000000000000'}, {'serial_number':'To Be Filled By O.E.M.','manufacturer':'HP'}]: with self.assertRaises(ValueError): bridge.hardware_id(data) def test_missing_hwid_never_uses_manual_number(self): with tempfile.TemporaryDirectory() as tmp, sqlite3.connect(':memory:') as db: bridge.init_db(db) path=Path(tmp)/'report.json' path.write_text(json.dumps({'asset_id':'45','model':'Laptop'})) with patch.object(bridge,'api_request') as api: with self.assertRaisesRegex(ValueError,'Hardware-Kennung'): bridge.sync_one(path,{},db) api.assert_not_called() def test_sync_revision_and_idempotence(self): with tempfile.TemporaryDirectory() as tmp: folder=Path(tmp); (folder/'reports').mkdir(); path=folder/'reports'/'report.json' path.write_text(json.dumps(self.report)) calls=[] def fake(base,key,method,suffix,payload=None): calls.append((method,suffix,payload)) if method=='POST': return {'number':501,'revision':3,'model':'HP EliteBook 745 G6','cpu':'old','ram_gb':8,'ssd_gb':128} return {'revision':4} with sqlite3.connect(':memory:') as db: bridge.init_db(db) with patch.object(bridge,'api_request',side_effect=fake), patch.dict('os.environ',{'CSR_API_KEY':'fake'}): res=bridge.sync_one(path,{'url':'https://example.org'},db) self.assertEqual(res,('SYNCED',501)) self.assertEqual(calls[0][1], 'devices/resolve/') self.assertEqual(calls[1][1], 'devices/501/') self.assertEqual(calls[1][0],'PATCH') self.assertEqual(calls[1][2]['expected_revision'],3) self.assertNotIn('ssd_gb',calls[1][2]) self.assertEqual(bridge.sync_one(path,{'url':'https://example.org'},db),('ALREADY_SYNCED',501)) self.assertEqual(len(calls),2) def test_mismatch_not_patched(self): with tempfile.TemporaryDirectory() as tmp: path=Path(tmp)/'report.json';path.write_text(json.dumps(self.report)) with sqlite3.connect(':memory:') as db: bridge.init_db(db) with patch.object(bridge,'api_request',return_value={'number':501,'revision':1,'model':'Different machine'}),patch.dict('os.environ',{'CSR_API_KEY':'fake'}): with self.assertRaisesRegex(ValueError,'MODEL_MISMATCH'): bridge.sync_one(path,{'url':'https://example.org'},db) class QueueTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.path = self.root / 'reports/PC-0042/100-1/report.json' self.path.parent.mkdir(parents=True) self.report = {'system_uuid':'12345678-1234-1234-1234-123456789012', 'asset_id': 'PC-0042', 'model': 'Test Laptop', 'cpu': 'CPU', 'timestamp_unix': 100} self.path.write_text(json.dumps(self.report)) self.db = sqlite3.connect(self.root / 'queue.sqlite3') self.addCleanup(self.db.close) bridge.init_db(self.db) def test_discovery_is_persistent_without_credentials(self): bridge.discover_queue(self.root, self.db) bridge.discover_queue(self.root, self.db) self.assertEqual(self.db.execute('SELECT count(*) FROM inventory_queue').fetchone()[0], 1) with patch.object(bridge, 'api_request') as api: self.assertEqual(bridge.process_queue(self.root, self.db, {}), 1) api.assert_not_called() state = self.db.execute('SELECT status,attempts,error,next_attempt FROM inventory_queue').fetchone() self.assertEqual(state[:2], ('ERROR', 1)) self.assertIn('not configured', state[2]) self.assertGreater(state[3], 0) # Backoff skips repeated attempts until due; manual retry bypasses it. self.assertEqual(bridge.process_queue(self.root, self.db, {}), 0) self.assertEqual(self.db.execute('SELECT attempts FROM inventory_queue').fetchone()[0], 1) bridge.process_queue(self.root, self.db, {}, retry=True) self.assertEqual(self.db.execute('SELECT attempts FROM inventory_queue').fetchone()[0], 2) def test_retry_after_ambiguous_response_does_not_patch_twice(self): remote = {'number': 501, 'revision': 1, 'model': 'Test Laptop', 'cpu': 'old'} patches = [] def api(base, key, method, suffix, payload=None): if method == 'POST': return dict(remote) patches.append(payload) remote.update({k:v for k,v in payload.items() if k != 'expected_revision'}) remote['revision'] += 1 raise OSError('connection lost after server committed') with patch.object(bridge, 'api_request', side_effect=api), patch.dict('os.environ', {'CSR_API_KEY':'secret'}): self.assertEqual(bridge.process_queue(self.root, self.db, {'url':'https://example.org'}), 1) self.assertEqual(bridge.process_queue(self.root, self.db, {'url':'https://example.org'}, retry=True), 0) self.assertEqual(len(patches), 1) self.assertEqual(self.db.execute('SELECT status FROM inventory_queue').fetchone()[0], 'SYNCED') bridge.process_queue(self.root, self.db, {'url':'https://example.org'}, retry=True) self.assertEqual(len(patches), 1) def test_changed_report_and_http_error_remain_unsynced(self): bridge.discover_queue(self.root, self.db) self.path.write_text(json.dumps({**self.report, 'cpu':'changed'})) with patch.object(bridge, 'api_request', side_effect=bridge.urllib.error.HTTPError('https://example.org', 409, 'conflict', {}, None)), patch.dict('os.environ', {'CSR_API_KEY':'TOPSECRET'}): bridge.process_queue(self.root, self.db, {'url':'https://example.org'}, retry=True) rows = self.db.execute('SELECT status,error FROM inventory_queue').fetchall() self.assertTrue(all(s == 'ERROR' for s, _ in rows)) self.assertTrue(any('changed' in e for _, e in rows)) self.assertTrue(any('409' in e for _, e in rows)) self.assertFalse(any('TOPSECRET' in e for _, e in rows)) def test_interrupted_upload_rechecks_server(self): bridge.discover_queue(self.root, self.db) with self.db: self.db.execute("UPDATE inventory_queue SET status='UPLOADING'") with patch.object(bridge, 'api_request', return_value={'number':501,'revision':2,'model':'Test Laptop','cpu':'CPU'}) as api, patch.dict('os.environ', {'CSR_API_KEY':'secret'}): self.assertEqual(bridge.process_queue(self.root, self.db, {'url':'https://example.org'}), 0) self.assertEqual(api.call_count, 1) self.assertEqual(api.call_args.args[2], 'POST') if __name__=='__main__': unittest.main() class ConnectionTests(unittest.TestCase): def test_api_check_uses_certificate_and_never_writes(self): with patch.object(bridge, 'api_request', return_value={}) as api, patch.dict('os.environ', {'CSR_API_KEY': 'test-key'}): message = bridge.check_connection({'url': 'https://inventory.test', 'ca_file': '/usb/config/inventory-server.crt'}) api.assert_called_once_with('https://inventory.test', 'test-key', 'GET', 'choices/', ca_file='/usb/config/inventory-server.crt') self.assertIn('No records uploaded', message) def test_ca_file_keeps_certificate_verification_enabled(self): context = bridge.ssl.create_default_context() with patch.object(bridge.ssl, 'create_default_context', return_value=context), patch.object(context, 'load_verify_locations') as load, patch.object(bridge.urllib.request, 'urlopen') as request: request.return_value.__enter__.return_value.read.return_value = b'{}' bridge.api_request('https://inventory.test', 'test-key', 'GET', 'choices/', ca_file='/usb/cert.pem') load.assert_called_once_with(cafile='/usb/cert.pem') self.assertEqual(context.verify_mode, bridge.ssl.CERT_REQUIRED) self.assertTrue(context.check_hostname) self.assertIs(request.call_args.kwargs['context'], context) def test_network_errors_are_actionable_without_raw_secrets(self): cases = [ (bridge.ssl.SSLCertVerificationError(1, 'secret'), 'certificate'), (bridge.socket.gaierror(-2, 'secret'), 'DNS'), (TimeoutError('secret'), 'timed out'), (ConnectionRefusedError('secret'), 'refused'), ('https://secret@example.test', 'unreachable'), ] for reason, expected in cases: message = bridge.connection_error(bridge.urllib.error.URLError(reason)) self.assertIn(expected, message) self.assertNotIn('secret', message) error = bridge.urllib.error.HTTPError('https://secret@example.test', 401, 'secret', {}, None) self.assertIn('API key', bridge.connection_error(error)) self.assertNotIn('secret', bridge.connection_error(error)) error.close()