me0nline
f0f6eba53b
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 2m21s
253 lines
11 KiB
Python
253 lines
11 KiB
Python
"""
|
|
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 (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_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("✅ Hosted & Ready for Fleet", html)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|