""" LOGAR - Unit and Integration Tests for Auto-Update & Update Alerting Tests version comparison, ServerUpdateManager, ClientUpdater, REST endpoints, and WebUI rendering. """ import os import sys import json import uuid import tempfile import unittest import warnings from typing import Optional warnings.filterwarnings("ignore") # Ensure parent directory and src directory are in path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))) import Server from version import LOGAR_VERSION, parse_version, compare_versions, is_newer_version from update_manager import ServerUpdateManager from client_updater import ClientUpdater class TestUpdateSystem(unittest.TestCase): def setUp(self): self.test_dir = tempfile.mkdtemp() self.test_db = os.path.join(self.test_dir, f"test_db_{uuid.uuid4().hex[:8]}.db") self.test_updates_dir = os.path.join(self.test_dir, "updates") Server.init_db(self.test_db) self.update_mgr = ServerUpdateManager(update_dir=self.test_updates_dir, check_enabled=False) Server.SERVER_STATE["update_manager"] = self.update_mgr Server.SERVER_STATE["config"] = { "db_path": self.test_db, "config_file_path": os.path.join(self.test_dir, "test_server_config.json") } def tearDown(self): import shutil if os.path.exists(self.test_dir): try: shutil.rmtree(self.test_dir) except Exception: pass def test_semantic_version_parsing_and_comparison(self): """Tests robust 4-part semantic version comparison.""" self.assertEqual(parse_version("0.3.5.2"), (0, 3, 5, 2)) self.assertEqual(parse_version("v0.3.5.2"), (0, 3, 5, 2)) self.assertEqual(parse_version("0.3.5"), (0, 3, 5, 0)) self.assertEqual(parse_version("1.0.0.0"), (1, 0, 0, 0)) self.assertEqual(parse_version(""), (0, 0, 0, 0)) # Comparisons self.assertEqual(compare_versions("0.3.5.3", "0.3.5.2"), 1) self.assertEqual(compare_versions("0.3.5.2", "0.3.5.2"), 0) self.assertEqual(compare_versions("0.3.5.1", "0.3.5.2"), -1) self.assertEqual(compare_versions("0.4.0.0", "0.3.9.9"), 1) self.assertEqual(compare_versions("1.0.0.0", "0.9.9.9"), 1) self.assertTrue(is_newer_version("0.3.6.0", "0.3.5.2")) self.assertFalse(is_newer_version("0.3.5.2", "0.3.5.2")) self.assertFalse(is_newer_version("0.3.5.1", "0.3.5.2")) def test_server_update_manager_package_registration(self): """Tests registering client update packages and calculating SHA256 hashes.""" dummy_pkg = os.path.join(self.test_dir, "LOGAR_Client_Setup.exe") with open(dummy_pkg, "wb") as f: f.write(b"LOGAR_WINDOWS_CLIENT_BINARY_PAYLOAD_V0360") manifest = self.update_mgr.register_client_package( os_type="windows", file_path=dummy_pkg, version="0.3.6.0", filename="LOGAR_Client_Setup.exe" ) self.assertEqual(manifest["os"], "windows") self.assertEqual(manifest["version"], "0.3.6.0") self.assertTrue(len(manifest["sha256"]) == 64) # Check hosted packages dictionary packages = self.update_mgr.get_hosted_packages() self.assertTrue(packages["windows"]["exists"]) self.assertEqual(packages["windows"]["version"], "0.3.6.0") self.assertEqual(packages["windows"]["sha256"], manifest["sha256"]) # Check client update evaluation for an older client eval_res = self.update_mgr.check_client_update( client_id="node-test-win", os_type="windows", current_version="0.3.5.2" ) self.assertTrue(eval_res["update_available"]) self.assertEqual(eval_res["latest_version"], "0.3.6.0") self.assertEqual(eval_res["sha256"], manifest["sha256"]) self.assertIn("/api/client/update/download", eval_res["download_url"]) # Client already on current version -> no update needed eval_current = self.update_mgr.check_client_update( client_id="node-test-win", os_type="windows", current_version="0.3.6.0" ) self.assertFalse(eval_current["update_available"]) def test_client_updater_checksum_validation(self): """Tests ClientUpdater SHA-256 verification and file integrity validation.""" updater = ClientUpdater(config={"server_host": "127.0.0.1", "hermes_port": 8443}, os_type="windows") test_file = os.path.join(self.test_dir, "test_bin.exe") content = b"TEST_PAYLOAD_FOR_AUTO_UPDATER" with open(test_file, "wb") as f: f.write(content) import hashlib expected_sha = hashlib.sha256(content).hexdigest() self.assertEqual(updater.calculate_sha256(test_file), expected_sha.lower()) def test_server_update_endpoints_and_webui(self): """Tests FastAPI update status, client check, and WebUI rendering with update components.""" # 1. Status API status = Server.get_update_status() self.assertIn("current_server_version", status) self.assertIn("hosted_packages", status) # 2. Register Windows package dummy_pkg = os.path.join(self.test_dir, "LOGAR_Client_Setup.exe") with open(dummy_pkg, "wb") as f: f.write(b"WIN_CLIENT_V0360") self.update_mgr.register_client_package("windows", dummy_pkg, version="0.3.6.0") # 3. Client update check endpoint (Allowed regardless of license) chk_unlicensed = Server.check_client_update( client_id="win-client-01", os="windows", current_version="0.3.5.1" ) self.assertTrue(chk_unlicensed["update_available"]) self.assertEqual(chk_unlicensed["latest_version"], "0.3.6.0") # 4. Enable auto_update feature flag in license config import sqlite3 conn = sqlite3.connect(self.test_db) conn.execute("UPDATE license_config SET license_status = 'active', features = '[\"auto_update\"]' WHERE id = 1") conn.commit() conn.close() # 5. Client update check endpoint (Licensed) chk_resp = Server.check_client_update( client_id="win-client-01", os="windows", current_version="0.3.5.1" ) self.assertTrue(chk_resp["update_available"]) self.assertEqual(chk_resp["latest_version"], "0.3.6.0") # 4. Download package endpoint from starlette.requests import Request as StarletteRequest req_scope = { "type": "http", "method": "GET", "path": "/api/client/update/download", "headers": [], "query_string": b"os=windows" } init_req = StarletteRequest(req_scope) dl_resp = Server.download_client_update_package(request=init_req, os="windows") self.assertEqual(dl_resp.status_code, 200) self.assertIn("X-Checksum-SHA256", dl_resp.headers) self.assertIn("ETag", dl_resp.headers) etag = dl_resp.headers["ETag"] # 5. Conditional GET with ETag (304 Not Modified) cond_scope = { "type": "http", "method": "GET", "path": "/api/client/update/download", "headers": [(b"if-none-match", etag.encode("utf-8"))], "query_string": b"os=windows" } cached_req = StarletteRequest(cond_scope) cached_resp = Server.download_client_update_package(request=cached_req, os="windows") self.assertEqual(cached_resp.status_code, 304) # 6. WebUI rendering verification # Simulate server update available self.update_mgr.server_update_available = True self.update_mgr.latest_upstream_version = "0.3.6.0" html = Server.WebUIRenderer.render_dashboard_html( server_state=Server.SERVER_STATE, ingest_queue=Server.INGEST_QUEUE ) self.assertIn("LOGAR Server Update Available: v0.3.6.0", html) self.assertIn("tabBtn-updates", html) self.assertIn("tabPane-updates", html) self.assertIn("Client Package Distribution Repository", html) def test_auto_update_switches_and_modes(self): """Tests On/Off toggle, Auto vs Manual modes, and config API endpoint.""" import sqlite3 conn = sqlite3.connect(self.test_db) conn.execute("UPDATE license_config SET license_status = 'active', features = '[\"auto_update\"]' WHERE id = 1") conn.commit() conn.close() dummy_pkg = os.path.join(self.test_dir, "LOGAR_Client_Setup.exe") with open(dummy_pkg, "wb") as f: f.write(b"WIN_CLIENT_V0360") self.update_mgr.register_client_package("windows", dummy_pkg, version="0.3.6.0") # 1. Active & auto mode -> update offered chk1 = Server.check_client_update(client_id="win-node", os="windows", current_version="0.3.5.2") self.assertTrue(chk1["update_available"]) # 2. Toggle Auto-Update OFF via API from server_models import ServerUpdateConfigRequest cfg_req = ServerUpdateConfigRequest(auto_update_enabled=False, update_mode="manual") res = Server.update_server_update_config(req=cfg_req, auth_user="admin") self.assertEqual(res["status"], "success") self.assertFalse(self.update_mgr.auto_update_enabled) self.assertEqual(self.update_mgr.update_mode, "manual") # Client update check while OFF -> blocked chk2 = Server.check_client_update(client_id="win-node", os="windows", current_version="0.3.5.2") self.assertFalse(chk2["update_available"]) self.assertIn("disabled", chk2.get("reason", "")) # 3. Toggle back ON with 'auto' mode cfg_req2 = ServerUpdateConfigRequest(auto_update_enabled=True, update_mode="auto") res2 = Server.update_server_update_config(req=cfg_req2, auth_user="admin") self.assertEqual(res2["status"], "success") self.assertTrue(self.update_mgr.auto_update_enabled) self.assertEqual(self.update_mgr.update_mode, "auto") # 4. Verify WebUI renders without download buttons in packages table html = Server.WebUIRenderer.render_dashboard_html( server_state=Server.SERVER_STATE, ingest_queue=Server.INGEST_QUEUE ) self.assertNotIn("/api/client/update/download?os=windows", html) self.assertNotIn("/api/client/update/download?os=linux", html) self.assertIn("Fleet Auto-Update Service", html) self.assertIn("Image Acquisition & Hosting Mode", html) self.assertIn("autoUpdateEnabledRadio", html) self.assertIn("updateModeRadio", html) self.assertIn("Ready", html) if __name__ == "__main__": unittest.main()