"""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 pytest import hashlib from datetime import datetime from shopdb.extensions import db 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() class TestApplicationDelta: """Test delta-based application inventory updates.""" @pytest.fixture def plugin(self): return EndpointsPlugin() @pytest.fixture def asset_id(self, db_session): """Create a test asset.""" from shopdb.core.models import Asset asset = Asset(assettag='TEST-001') db_session.add(asset) db_session.commit() return asset.assetid 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_session): """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_session): """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_session): """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_session): """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_session): """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.app_fingerprint is None for app in apps) def test_empty_inventory(self, plugin, asset_id, db_session): """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_session): """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_session): """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_session): """Create a test asset.""" from shopdb.core.models import Asset asset = Asset(assettag='TEST-SAFETY-001') db_session.add(asset) db_session.commit() return asset.assetid 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)