"""Tests for application inventory delta-based updates. Tests the _replace_applications() method with fingerprint-based delta logic, including safety features: fingerprint validation, delete count verification, memory limits, and concurrent update protection. """ import hashlib import pytest from plugins.endpoints.models import EndpointApplication from plugins.endpoints.plugin import EndpointsPlugin def compute_test_fingerprint(name, version, publisher): """Compute fingerprint matching agent algorithm.""" name = name or '' version = version or '' publisher = publisher or '' data = f"{name}|{version}|{publisher}" return hashlib.sha256(data.encode('utf-8')).hexdigest() def _make_asset(db, assetnumber): """A bare computer-type Asset; _replace_applications needs only the id.""" from shopdb.core.models import Asset, AssetType assettype = AssetType.query.filter_by(assettype='computer').first() if assettype is None: assettype = AssetType(assettype='computer') db.session.add(assettype) db.session.flush() asset = Asset(assetnumber=assetnumber, name=assetnumber, assettypeid=assettype.assettypeid, isactive=True) db.session.add(asset) db.session.commit() return asset.assetid class TestApplicationDelta: """Test delta-based application inventory updates.""" @pytest.fixture def plugin(self): return EndpointsPlugin() @pytest.fixture def asset_id(self, db): """Create a test asset.""" return _make_asset(db, 'TEST-001') def test_fingerprint_validation_valid(self, plugin): """Valid fingerprints (64 lowercase hex chars) pass validation.""" valid_fps = [ 'a' * 64, # All lowercase a '0' * 64, # All zeros 'abc123' * 10 + 'abcd', # Mix of hex chars (64 total) ] for fp in valid_fps: assert plugin._is_valid_fingerprint(fp), f"Should accept: {fp}" def test_fingerprint_validation_invalid(self, plugin): """Invalid fingerprints are rejected.""" invalid_fps = [ '', # Empty 'abc', # Too short 'A' * 64, # Uppercase 'g' * 64, # Non-hex character 'abc-def' * 9, # Contains hyphens '0' * 63, # One char too short '0' * 65, # One char too long ] for fp in invalid_fps: assert not plugin._is_valid_fingerprint(fp), f"Should reject: {fp}" def test_delta_no_changes(self, plugin, asset_id, db): """No apps added/removed when inventory unchanged.""" # Initial insert: 3 apps apps_v1 = [ { 'displayname': 'Chrome', 'displayversion': '117.0', 'publisher': 'Google', 'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google') }, { 'displayname': 'Firefox', 'displayversion': '119.0', 'publisher': 'Mozilla', 'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla') }, { 'displayname': 'Office', 'displayversion': '16.0', 'publisher': 'Microsoft', 'fingerprint': compute_test_fingerprint('Office', '16.0', 'Microsoft') }, ] warnings = [] result = plugin._replace_applications(asset_id, apps_v1, warnings) db.session.commit() assert result == 3 assert len(warnings) == 0 assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 3 # Second report: same apps (no changes) warnings = [] result = plugin._replace_applications(asset_id, apps_v1, warnings) db.session.commit() assert result == 0 # 0 added assert len(warnings) == 0 assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 3 # Still 3 def test_delta_add_remove(self, plugin, asset_id, db): """Detect adds and removes correctly.""" # Initial: 2 apps apps_v1 = [ { 'displayname': 'Chrome', 'displayversion': '117.0', 'publisher': 'Google', 'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google') }, { 'displayname': 'Firefox', 'displayversion': '119.0', 'publisher': 'Mozilla', 'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla') }, ] warnings = [] plugin._replace_applications(asset_id, apps_v1, warnings) db.session.commit() assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 2 # Update: Remove Firefox, Add Office apps_v2 = [ { 'displayname': 'Chrome', 'displayversion': '117.0', 'publisher': 'Google', 'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google') }, { 'displayname': 'Office', 'displayversion': '16.0', 'publisher': 'Microsoft', 'fingerprint': compute_test_fingerprint('Office', '16.0', 'Microsoft') }, ] warnings = [] result = plugin._replace_applications(asset_id, apps_v2, warnings) db.session.commit() assert result == 1 # 1 added (Office) assert len(warnings) == 0 apps = EndpointApplication.query.filter_by(assetid=asset_id).all() assert len(apps) == 2 app_names = {app.displayname for app in apps} assert app_names == {'Chrome', 'Office'} def test_delta_version_change(self, plugin, asset_id, db): """Version change detected (different fingerprint).""" # Initial: Chrome 117.0 apps_v1 = [ { 'displayname': 'Chrome', 'displayversion': '117.0', 'publisher': 'Google', 'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google') }, ] warnings = [] plugin._replace_applications(asset_id, apps_v1, warnings) db.session.commit() # Update: Chrome 118.0 (different fingerprint) apps_v2 = [ { 'displayname': 'Chrome', 'displayversion': '118.0', 'publisher': 'Google', 'fingerprint': compute_test_fingerprint('Chrome', '118.0', 'Google') }, ] warnings = [] result = plugin._replace_applications(asset_id, apps_v2, warnings) db.session.commit() assert result == 1 # 1 added (new version) apps = EndpointApplication.query.filter_by(assetid=asset_id).all() assert len(apps) == 1 assert apps[0].displayversion == '118.0' def test_invalid_fingerprint_skipped(self, plugin, asset_id, db): """Apps with invalid fingerprints are skipped with warning.""" apps = [ { 'displayname': 'Chrome', 'displayversion': '117.0', 'publisher': 'Google', 'fingerprint': 'INVALID' # Too short }, { 'displayname': 'Firefox', 'displayversion': '119.0', 'publisher': 'Mozilla', 'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla') }, ] warnings = [] result = plugin._replace_applications(asset_id, apps, warnings) db.session.commit() assert result == 1 # Only Firefox added assert len(warnings) == 1 assert 'invalid fingerprint' in warnings[0].lower() assert 'Chrome' in warnings[0] apps = EndpointApplication.query.filter_by(assetid=asset_id).all() assert len(apps) == 1 assert apps[0].displayname == 'Firefox' def test_legacy_fallback_no_fingerprints(self, plugin, asset_id, db): """Legacy mode works when agent doesn't send fingerprints.""" apps = [ { 'displayname': 'Chrome', 'displayversion': '117.0', 'publisher': 'Google', # No fingerprint key }, { 'displayname': 'Firefox', 'displayversion': '119.0', 'publisher': 'Mozilla', # No fingerprint key }, ] warnings = [] result = plugin._replace_applications(asset_id, apps, warnings) db.session.commit() assert result == 2 assert len(warnings) == 0 apps = EndpointApplication.query.filter_by(assetid=asset_id).all() assert len(apps) == 2 assert all(app.appfingerprint is None for app in apps) def test_empty_inventory(self, plugin, asset_id, db): """Empty inventory removes all apps.""" # Initial: 2 apps apps_v1 = [ { 'displayname': 'Chrome', 'displayversion': '117.0', 'publisher': 'Google', 'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google') }, { 'displayname': 'Firefox', 'displayversion': '119.0', 'publisher': 'Mozilla', 'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla') }, ] warnings = [] plugin._replace_applications(asset_id, apps_v1, warnings) db.session.commit() assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 2 # Update: Empty list warnings = [] result = plugin._replace_applications(asset_id, [], warnings) db.session.commit() assert result == 0 assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 0 def test_missing_displayname_skipped(self, plugin, asset_id, db): """Apps without displayname are skipped.""" apps = [ { # No displayname 'displayversion': '117.0', 'publisher': 'Google', 'fingerprint': compute_test_fingerprint('', '117.0', 'Google') }, { 'displayname': 'Firefox', 'displayversion': '119.0', 'publisher': 'Mozilla', 'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla') }, ] warnings = [] result = plugin._replace_applications(asset_id, apps, warnings) db.session.commit() assert result == 1 # Only Firefox added assert len(warnings) == 1 assert 'no displayname' in warnings[0].lower() def test_string_truncation(self, plugin, asset_id, db): """Long strings are truncated to column limits.""" apps = [ { 'displayname': 'A' * 300, # Exceeds 255 char limit 'displayversion': 'B' * 200, # Exceeds 128 char limit 'publisher': 'C' * 300, # Exceeds 255 char limit 'fingerprint': compute_test_fingerprint('A' * 300, 'B' * 200, 'C' * 300) }, ] warnings = [] result = plugin._replace_applications(asset_id, apps, warnings) db.session.commit() assert result == 1 app = EndpointApplication.query.filter_by(assetid=asset_id).first() assert len(app.displayname) == 255 assert len(app.displayversion) == 128 assert len(app.publisher) == 255 class TestApplicationDeltaSafety: """Test safety features: locking, delete verification, memory limits.""" @pytest.fixture def plugin(self): return EndpointsPlugin() @pytest.fixture def asset_id(self, db): """Create a test asset.""" return _make_asset(db, 'TEST-SAFETY-001') def test_compute_fingerprint_consistency(self): """Fingerprint algorithm is deterministic.""" fp1 = compute_test_fingerprint('Chrome', '117.0', 'Google') fp2 = compute_test_fingerprint('Chrome', '117.0', 'Google') assert fp1 == fp2 # Different inputs produce different fingerprints fp3 = compute_test_fingerprint('Chrome', '118.0', 'Google') assert fp1 != fp3 def test_fingerprint_null_handling(self): """Null values handled correctly (converted to empty string).""" fp1 = compute_test_fingerprint('App', None, None) fp2 = compute_test_fingerprint('App', '', '') assert fp1 == fp2 # None and '' should produce same fingerprint def test_fingerprint_format(self): """Fingerprints are 64 lowercase hex chars.""" fp = compute_test_fingerprint('Test', '1.0', 'Vendor') assert len(fp) == 64 assert fp.islower() assert all(c in '0123456789abcdef' for c in fp)