Add Windows Server vs Client differentiation, installed server roles introspection, dedicated /client-info directory, and best practice baseline analyzer
CI Test Suite / Run Component Tests & Pipeline Verification (push) Failing after 1m50s
CI Test Suite / Run Component Tests & Pipeline Verification (push) Failing after 1m50s
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
"""
|
||||
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.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
|
||||
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_client_info_html_rendering(self):
|
||||
"""Verifies GET /client-info redirects to /login when unauthenticated, and renders HTML 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 renders the full directory page
|
||||
res1 = self.client.get("/client-info", cookies={SessionManager.SESSION_COOKIE_NAME: self.session_token})
|
||||
self.assertEqual(res1.status_code, 200)
|
||||
self.assertIn("LOGAR Server & Client Role Directory", res1.text)
|
||||
self.assertIn("Node Telemetry & Server Roles Inventory", res1.text)
|
||||
|
||||
res2 = self.client.get("/webui/client-info", cookies={SessionManager.SESSION_COOKIE_NAME: self.session_token})
|
||||
self.assertEqual(res2.status_code, 200)
|
||||
self.assertIn("LOGAR Server & Client Role Directory", res2.text)
|
||||
|
||||
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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user