LOGAR/tests/test_updates.py

"""
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 (Unlicensed should block auto_update)
        chk_blocked = Server.check_client_update(
            client_id="win-client-01",
            os="windows",
            current_version="0.3.5.1"
        )
        self.assertFalse(chk_blocked["update_available"])
        self.assertIn("not authorized", chk_blocked.get("reason", ""))

        # 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_update_memoization_and_cache_invalidation(self):
        """Tests that package info is served from memory cache and invalidates on update."""
        dummy_pkg = os.path.join(self.test_dir, "LOGAR_Client_Linux")
        with open(dummy_pkg, "wb") as f:
            f.write(b"LINUX_CLIENT_PAYLOAD_V1")

        self.update_mgr.register_client_package("linux", dummy_pkg, version="0.3.6.0")
        pkgs_1 = self.update_mgr.get_hosted_packages()
        sha1 = pkgs_1["linux"]["sha256"]

        # Repeated call should hit memory cache with identical hash
        pkgs_2 = self.update_mgr.get_hosted_packages()
        self.assertEqual(pkgs_2["linux"]["sha256"], sha1)

        # Invalidate cache on new registration
        with open(dummy_pkg, "wb") as f:
            f.write(b"LINUX_CLIENT_PAYLOAD_V2_UPDATED")
        self.update_mgr.register_client_package("linux", dummy_pkg, version="0.3.7.0")
        pkgs_3 = self.update_mgr.get_hosted_packages()
        self.assertEqual(pkgs_3["linux"]["version"], "0.3.7.0")
        self.assertNotEqual(pkgs_3["linux"]["sha256"], sha1)


if __name__ == "__main__":
    unittest.main()