Add server update alerting and fleet client auto-update system (synchronized release lockstep)

tests/test_updates.py

@@ -0,0 +1,160 @@
+"""
+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()