""" 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} 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 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 dl_resp = Server.download_client_update_package(os="windows") self.assertEqual(dl_resp.status_code, 200) self.assertIn("X-Checksum-SHA256", dl_resp.headers) # 5. 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) if __name__ == "__main__": unittest.main()