""" LOGAR - Comprehensive Unit Tests for Edge Clients & Forwarders Covers log_collector, client_transport, client_updater, Win_Client, and Linux_Client. """ import os import sys import json import time import tempfile import unittest import socket import ssl from unittest.mock import MagicMock, patch from src.clients.log_collector import ( get_machine_identifier, is_path_writable, get_default_data_dir, resolve_writable_path, get_state_path, load_state, save_state, commit_state, WindowsLogCollector, LinuxLogCollector, BaseLogCollector, get_log_collector, is_routine_service_noise, collect_system_metrics ) from src.clients.client_updater import ClientUpdater from src.clients.client_transport import ( is_cert_expiring_soon, enroll_client_if_needed, load_config, ensure_enrolled, send_encrypted_logs_over_socket, check_and_apply_auto_update ) import src.clients.Win_Client as Win_Client import src.clients.Linux_Client as Linux_Client class TestClientsComprehensive(unittest.TestCase): def setUp(self): self.test_dir = tempfile.mkdtemp() def tearDown(self): import shutil if os.path.exists(self.test_dir): try: shutil.rmtree(self.test_dir) except Exception: pass def test_log_collector_utilities(self): """Tests machine identifier, path resolution, and state persistence.""" mid = get_machine_identifier() self.assertTrue(len(mid) > 0) data_dir = get_default_data_dir("TEST-LOGAR") self.assertTrue(isinstance(data_dir, str)) resolved = resolve_writable_path(os.path.join(self.test_dir, "test.db"), "TEST-LOGAR") self.assertEqual(os.path.abspath(resolved), os.path.abspath(os.path.join(self.test_dir, "test.db"))) state_path = get_state_path("test_state.json", "TEST-LOGAR") self.assertTrue(isinstance(state_path, str)) # State load/save/commit st_file = os.path.join(self.test_dir, "st.json") st_data = load_state(st_file) self.assertIsInstance(st_data, dict) st_data["processed_records"] = 42 st_data["new_last_record_number"] = 100 save_state(st_file, st_data) loaded = load_state(st_file) self.assertEqual(loaded["processed_records"], 42) commit_state(loaded, st_file) loaded2 = load_state(st_file) self.assertEqual(loaded2["last_record_number"], 100) def test_metrics_collection_and_noise_filter(self): """Tests system metrics telemetry and routine noise filtering.""" metrics = collect_system_metrics() self.assertIn("cpu_percent", metrics) self.assertIn("memory_percent", metrics) self.assertIn("disk_percent_free", metrics) # Routine noise filter: signature, message, severity self.assertTrue(is_routine_service_noise("Service Control Manager", "Windows Error Reporting Service entered the stopped state.", "INFO")) self.assertTrue(is_routine_service_noise("Service Control Manager", "Background Intelligent Transfer Service entered the running state.", "INFO")) self.assertFalse(is_routine_service_noise("NTFS", "Fatal disk corruption error detected on C:", "ERROR")) def test_log_collector_classes(self): """Tests WindowsLogCollector and LinuxLogCollector implementations.""" # Linux Log Collector linux_col = LinuxLogCollector() self.assertIsNotNone(linux_col) # Windows Log Collector win_col = WindowsLogCollector() self.assertEqual(win_col.channels, ["Application"]) factory_col = get_log_collector() self.assertIsInstance(factory_col, BaseLogCollector) def test_client_updater_lifecycle(self): """Tests ClientUpdater download, local hash match, and update application routines.""" updater = ClientUpdater(config={"server_host": "127.0.0.1", "hermes_port": 8443}, os_type="linux") self.assertEqual(updater.os_type, "linux") base_url = updater.get_server_base_url() self.assertEqual(base_url, "http://127.0.0.1:8443") # Linux apply update mock dummy_bin = os.path.join(self.test_dir, "dummy_new_bin") dummy_target = os.path.join(self.test_dir, "dummy_target_bin") with open(dummy_bin, "wb") as f: f.write(b"NEW_BINARY_CONTENT") with open(dummy_target, "wb") as f: f.write(b"OLD_BINARY_CONTENT") ok = updater.apply_update_linux(dummy_bin, dummy_target) self.assertTrue(ok) with open(dummy_target, "rb") as f: self.assertEqual(f.read(), b"NEW_BINARY_CONTENT") # Windows updater batch generation updater_win = ClientUpdater(config={"server_host": "127.0.0.1", "hermes_port": 8443}, os_type="windows") with patch("subprocess.Popen") as mock_popen: win_bin = os.path.join(self.test_dir, "win_new.exe") win_target = os.path.join(self.test_dir, "win_target.exe") with open(win_bin, "wb") as f: f.write(b"WIN_EXE") with open(win_target, "wb") as f: f.write(b"WIN_TARGET") res = updater_win.apply_update_windows(win_bin, win_target) self.assertTrue(res) self.assertTrue(mock_popen.called) def test_client_transport_config_and_enrollment(self): """Tests loading client config, cert expiration check, and fallback.""" cfg_file = os.path.join(self.test_dir, "client_config.json") cert_dir = os.path.join(self.test_dir, "certs") os.makedirs(cert_dir, exist_ok=True) cfg_data = { "server_host": "127.0.0.1", "server_port": 9443, "hermes_host": "127.0.0.1", "hermes_port": 8443, "enrollment_secret": "test_sec", "cert_dir": cert_dir, "server_fingerprint": "FP123", "server_public_key": "", "auth_token": "token123" } with open(cfg_file, "w", encoding="utf-8") as f: json.dump(cfg_data, f) loaded_cfg = load_config(cfg_file) self.assertEqual(loaded_cfg["server_host"], "127.0.0.1") # Expiry check self.assertTrue(is_cert_expiring_soon(os.path.join(cert_dir, "nonexistent.crt"))) def test_client_main_helpers(self): """Tests Win_Client and Linux_Client config loading and log retrieval fallbacks.""" logs = Win_Client.get_recent_windows_logs(hours=1, state={}) self.assertIsInstance(logs, list) logs_l = Linux_Client.get_recent_linux_logs(hours=1, state={}) self.assertIsInstance(logs_l, list) if __name__ == "__main__": unittest.main()