""" Unit and integration tests for LOGAR Server Roles detection, Windows Server vs Windows Client differentiation, /client-info subpage, and Best Practice baseline engine. """ import os import sys import json import sqlite3 import unittest from unittest.mock import patch, MagicMock from fastapi.testclient import TestClient from src.common.system_info import ( get_os_details, get_ip_addresses, get_full_system_inventory, get_installed_windows_server_roles ) from src.hub.best_practices import BestPracticeEngine, BestPracticeRule, BEST_PRACTICES_ENGINE from src.hub.best_practice_rules import BASELINE_RULES, get_baseline_rules from src.hub.web_ui import WebUIRenderer from src.hub.Server import app, SERVER_STATE, DEFAULT_DB_FILE from src.hub.auth_manager import SessionManager class TestServerRolesAndBestPractices(unittest.TestCase): def setUp(self): self.client = TestClient(app) self.db_path = SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE if not os.path.exists(self.db_path): self.db_path = DEFAULT_DB_FILE SERVER_STATE.setdefault("config", {})["db_path"] = self.db_path from src.hub.Server import init_db init_db(self.db_path) self.session_token = SessionManager.generate_session_token("admin", db_path=self.db_path) self.auth_headers = {"Authorization": f"Bearer {self.session_token}"} def test_os_details_classification(self): """Verifies OS classification returns valid structure with is_server flag.""" details = get_os_details() self.assertIn("os_type", details) self.assertIn("os_name", details) self.assertIn("is_server", details) self.assertIn("hostname", details) self.assertIn("ip_addresses", details) self.assertIsInstance(details["is_server"], bool) self.assertIn(details["os_type"], ["windows", "windows_server", "linux", "darwin", "unknown"]) def test_full_system_inventory(self): """Verifies full system inventory includes CPU, RAM, disk, and server roles.""" inv = get_full_system_inventory() self.assertIn("hostname", inv) self.assertIn("os_type", inv) self.assertIn("cpu_count", inv) self.assertIn("memory_total_mb", inv) self.assertIn("disk_percent_free", inv) self.assertIn("server_roles", inv) self.assertIsInstance(inv["server_roles"], list) def test_best_practices_engine_evaluation(self): """Tests that baseline best practice checks evaluate properly.""" engine = BestPracticeEngine() rules = engine.get_rules() self.assertGreaterEqual(len(rules), 4) # Test node with healthy specs healthy_node = { "client_id": "test-dc-01", "hostname": "DC01.corp.local", "os_type": "windows_server", "os_name": "Windows Server 2022 Datacenter", "is_server": True, "server_roles": ["Active Directory Domain Services", "DNS Server"], "disk_percent_free": 45.0, "memory_percent": 55.0, "client_version": "1.2.0", "update_status": "up_to_date" } res = engine.evaluate_node(healthy_node) self.assertTrue(all(r["status"] == "PASS" for r in res)) # Test DC with bad practice (co-located IIS Web Server) unhealthy_dc = { "client_id": "test-dc-bad", "hostname": "DC-BAD.corp.local", "os_type": "windows_server", "os_name": "Windows Server 2022 Datacenter", "is_server": True, "server_roles": ["Active Directory Domain Services", "Web Server (IIS)"], "disk_percent_free": 5.0, # Critical disk space (< 10%) "memory_percent": 96.0, # Exhausted memory (>= 95%) "client_version": "1.0.0", "update_status": "outdated" } res_bad = engine.evaluate_node(unhealthy_dc) statuses = {r["rule_id"]: r["status"] for r in res_bad} self.assertEqual(statuses.get("BP-AD-001"), "FAIL") # DC co-location conflict self.assertEqual(statuses.get("BP-SYS-001"), "FAIL") # Critical disk self.assertEqual(statuses.get("BP-SYS-002"), "FAIL") # Exhausted RAM self.assertEqual(statuses.get("BP-SEC-003"), "WARN") # Outdated client def test_best_practices_custom_rule_extensibility(self): """Tests user extensibility: custom rules can be defined and registered dynamically.""" engine = BestPracticeEngine() custom_rule = BestPracticeRule( rule_id="BP-CUSTOM-001", category="Custom Compliance", title="Custom Domain Suffix Enforcement", description="All fleet nodes must belong to the internal domain.", recommendation="Join host to corp.internal.", severity="WARNING", eval_fn=lambda node: {"status": "PASS" if "corp.internal" in node.get("hostname", "") else "FAIL", "message": "Domain check."} ) engine.register_rule(custom_rule) rules = engine.get_rules() self.assertTrue(any(r["rule_id"] == "BP-CUSTOM-001" for r in rules)) eval_pass = engine.evaluate_node({"hostname": "server1.corp.internal"}) self.assertTrue(any(r["rule_id"] == "BP-CUSTOM-001" and r["status"] == "PASS" for r in eval_pass)) eval_fail = engine.evaluate_node({"hostname": "server1.public.com"}) self.assertTrue(any(r["rule_id"] == "BP-CUSTOM-001" and r["status"] == "FAIL" for r in eval_fail)) def test_rules_defined_in_separate_file(self): """Verifies Best Practice rules are defined in separate best_practice_rules module.""" rules = get_baseline_rules() self.assertGreaterEqual(len(rules), 5) for r in rules: self.assertIsInstance(r, BestPracticeRule) self.assertTrue(r.rule_id.startswith("BP-")) self.assertTrue(r.title) self.assertTrue(r.description) self.assertTrue(r.recommendation) self.assertIn(r.severity, ["CRITICAL", "WARNING", "RECOMMENDED"]) self.assertIsNotNone(r.eval_fn) # Engine automatically registers all rules from best_practice_rules module engine = BestPracticeEngine() loaded_ids = [r["rule_id"] for r in engine.get_rules()] self.assertIn("BP-SYS-001", loaded_ids) self.assertIn("BP-SYS-002", loaded_ids) self.assertIn("BP-AD-001", loaded_ids) self.assertIn("BP-WIN-002", loaded_ids) self.assertIn("BP-SEC-003", loaded_ids) def test_client_info_html_rendering(self): """Verifies GET /client-info redirects to /login when unauthenticated, and redirects to /webui#server-roles when authenticated.""" # Unauthenticated redirects to /login res_unauth = self.client.get("/client-info", follow_redirects=False) self.assertEqual(res_unauth.status_code, 303) self.assertEqual(res_unauth.headers.get("location"), "/login") # Authenticated redirects to /webui#server-roles auth_client = TestClient(app) auth_client.cookies.set(SessionManager.SESSION_COOKIE_NAME, self.session_token) res1 = auth_client.get("/client-info", follow_redirects=False) self.assertEqual(res1.status_code, 303) self.assertEqual(res1.headers.get("location"), "/webui#server-roles") res2 = auth_client.get("/webui/client-info", follow_redirects=False) self.assertEqual(res2.status_code, 303) self.assertEqual(res2.headers.get("location"), "/webui#server-roles") # Following redirect leads to main dashboard with Server Roles tab pane res_follow = auth_client.get("/client-info", follow_redirects=True) self.assertEqual(res_follow.status_code, 200) self.assertIn("id=\"tabPane-server-roles\"", res_follow.text) self.assertIn("Node Telemetry & Server Roles Inventory", res_follow.text) def test_dashboard_tabs_structure(self): """Verifies primary tabs (Fleet Forwarders, Server Roles & Client Directory, Events, Administration) and secondary admin subtabs (Hermes, Sub-CA, DB, Updates).""" res = self.client.get("/webui", cookies={SessionManager.SESSION_COOKIE_NAME: self.session_token}) self.assertEqual(res.status_code, 200) html = res.text # Primary navigation tabs (only 4 tabs on top row) self.assertIn("id=\"mainTabsNav\"", html) self.assertIn("id=\"tabBtn-fleet\"", html) self.assertIn(">💻 Fleet Forwarders", html) self.assertIn("id=\"tabBtn-server-roles\"", html) self.assertIn(">🖥️ Server Roles & Client Directory", html) self.assertIn("id=\"tabBtn-events\"", html) self.assertIn(">📊 Events", html) self.assertIn("id=\"tabBtn-admin\"", html) self.assertIn(">⚙️ Administration", html) # Administration sub-navigation tabs (row below) self.assertIn("id=\"adminSubtabs\"", html) self.assertIn("id=\"tabBtn-hermes\"", html) self.assertIn(">🤖 Hermes AI & Tokens", html) self.assertIn("id=\"tabBtn-pki\"", html) self.assertIn(">📜 Sub-CA & Licensing", html) self.assertIn("id=\"tabBtn-db\"", html) self.assertIn(">💾 DB Engine", html) self.assertIn("id=\"tabBtn-updates\"", html) self.assertIn(">📦 Updates & Distribution", html) # Tab panes self.assertIn("id=\"tabPane-fleet\"", html) self.assertIn("id=\"tabPane-server-roles\"", html) self.assertIn("id=\"tabPane-events\"", html) self.assertIn("id=\"tabPane-pki\"", html) self.assertIn("id=\"tabPane-db\"", html) self.assertIn("id=\"tabPane-updates\"", html) self.assertIn("id=\"tabPane-hermes\"", html) # Telemetry JSON inspector modal self.assertIn("id=\"telemetryModal\"", html) self.assertIn("id=\"modalJsonContent\"", html) def test_api_client_info_endpoint(self): """Verifies GET /api/client-info provides structured JSON inventory.""" res = self.client.get("/api/client-info", headers=self.auth_headers) self.assertEqual(res.status_code, 200) data = res.json() self.assertEqual(data.get("status"), "success") self.assertIn("total_nodes", data) self.assertIn("windows_servers", data) self.assertIn("windows_workstations", data) self.assertIn("linux_nodes", data) self.assertIn("clients", data) def test_api_best_practices_status_unlicensed(self): """Verifies /api/best-practices/status returns locked when unlicensed.""" with patch("src.hub.Server.is_feature_licensed", return_value=False): res = self.client.get("/api/best-practices/status", headers=self.auth_headers) self.assertEqual(res.status_code, 200) data = res.json() self.assertEqual(data.get("status"), "locked") self.assertFalse(data.get("licensed")) self.assertIn("enterprise capability", data.get("message", "").lower()) def test_api_best_practices_status_licensed(self): """Verifies /api/best-practices/status returns active evaluation when feature is licensed.""" with patch("src.hub.Server.is_feature_licensed", return_value=True): res = self.client.get("/api/best-practices/status", headers=self.auth_headers) self.assertEqual(res.status_code, 200) data = res.json() self.assertEqual(data.get("status"), "active") self.assertTrue(data.get("licensed")) self.assertIn("evaluation", data) eval_data = data["evaluation"] self.assertIn("score", eval_data) self.assertIn("pass_count", eval_data) self.assertIn("warn_count", eval_data) self.assertIn("fail_count", eval_data) def test_admin_version_info_modal(self): """Verifies admin dropdown menu contains Version & System Info option and renders telemetry modal in /webui.""" from src.version import LOGAR_VERSION auth_client = TestClient(app) auth_client.cookies.set(SessionManager.SESSION_COOKIE_NAME, self.session_token) # Check /webui res_webui = auth_client.get("/webui") self.assertEqual(res_webui.status_code, 200) self.assertIn("id=\"userDropdown\"", res_webui.text) self.assertIn("Version & System Info", res_webui.text) self.assertIn("openVersionInfoModal()", res_webui.text) self.assertIn("id=\"versionInfoModal\"", res_webui.text) self.assertIn(f"v{LOGAR_VERSION}", res_webui.text) self.assertIn("Server Hub Release", res_webui.text) self.assertIn("Hosted Client Binaries", res_webui.text) self.assertIn("Repository Update Mode:", res_webui.text) self.assertIn("Distribution Source:", res_webui.text) self.assertIn("Python Environment:", res_webui.text) self.assertIn("Host Platform & OS:", res_webui.text) self.assertIn("TLS 1.3 mTLS Encrypted", res_webui.text) self.assertIn("Storage Backend Engine:", res_webui.text) self.assertIn("License Status & Seats:", res_webui.text) # Check /client-info redirects to /webui#server-roles and shows the version info modal res_client_info = auth_client.get("/client-info", follow_redirects=True) self.assertEqual(res_client_info.status_code, 200) self.assertIn("id=\"userDropdown\"", res_client_info.text) self.assertIn("Version & System Info", res_client_info.text) self.assertIn("id=\"versionInfoModal\"", res_client_info.text) self.assertIn(f"v{LOGAR_VERSION}", res_client_info.text) if __name__ == "__main__": unittest.main()