203 lines
12 KiB
Python
203 lines
12 KiB
Python
import importlib.util
|
|
import contextlib
|
|
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, contextlib.closing(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 contextlib.closing(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 contextlib.closing(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_resolve_exposes_new_or_existing_registration(self):
|
|
for status, expected in ((201, 'NEW'), (200, 'EXISTING')):
|
|
with self.subTest(status=status), patch.object(bridge.urllib.request, 'urlopen') as request:
|
|
response = request.return_value.__enter__.return_value
|
|
response.status = status
|
|
response.read.return_value = b'{"number": 42}'
|
|
result = bridge.api_request(
|
|
'https://inventory.test', 'test-key', 'POST', 'devices/resolve/', {}
|
|
)
|
|
self.assertEqual(result['_registration'], expected)
|
|
|
|
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()
|
|
|
|
def test_inventory_url_must_be_an_https_origin(self):
|
|
with patch.dict('os.environ', {'CSR_API_KEY': 'test-key'}):
|
|
self.assertEqual(
|
|
bridge.credentials({'url': 'https://inventory.test/'}),
|
|
('https://inventory.test', 'test-key'),
|
|
)
|
|
for url in (
|
|
'http://inventory.test',
|
|
'https://user:secret@inventory.test',
|
|
'https://inventory.test/api',
|
|
'https://inventory.test?token=secret',
|
|
'https://inventory.test#fragment',
|
|
):
|
|
with self.subTest(url=url), self.assertRaisesRegex(ValueError, 'HTTPS origin'):
|
|
bridge.credentials({'url': url})
|