109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""
|
|
Unit tests for LOGAR log_collector and client_transport modules.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import unittest
|
|
import warnings
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
warnings.filterwarnings("ignore")
|
|
warnings.simplefilter("ignore")
|
|
|
|
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")))
|
|
|
|
from 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,
|
|
BaseLogCollector,
|
|
WindowsLogCollector,
|
|
LinuxLogCollector,
|
|
get_log_collector
|
|
)
|
|
from client_transport import (
|
|
is_cert_expiring_soon,
|
|
load_config
|
|
)
|
|
|
|
|
|
class TestLogCollectorModule(unittest.TestCase):
|
|
def test_machine_identifier(self):
|
|
mid = get_machine_identifier()
|
|
self.assertIsInstance(mid, str)
|
|
self.assertTrue(len(mid) > 0)
|
|
self.assertNotEqual(mid, "localhost")
|
|
|
|
def test_path_writable_and_data_dir(self):
|
|
data_dir = get_default_data_dir("LOGAR_TEST")
|
|
self.assertIsInstance(data_dir, str)
|
|
self.assertTrue(is_path_writable(data_dir))
|
|
|
|
def test_state_lifecycle_windows_and_linux(self):
|
|
state_file = "test_shared_state.json"
|
|
try:
|
|
state = load_state(state_file)
|
|
self.assertEqual(state, {})
|
|
|
|
# Windows staged state
|
|
state["new_last_record_number"] = 100
|
|
state["new_sent_record_ids"] = ["100:2026-09-27T00:00:00"]
|
|
commit_state(state, state_file)
|
|
self.assertNotIn("new_last_record_number", state)
|
|
self.assertEqual(state["last_record_number"], 100)
|
|
|
|
# Linux staged state
|
|
state["new_last_cursor"] = "c_xyz"
|
|
state["new_last_timestamp_us"] = 123456789.0
|
|
state["new_sent_cursors"] = ["c_xyz"]
|
|
commit_state(state, state_file)
|
|
self.assertNotIn("new_last_cursor", state)
|
|
self.assertEqual(state["last_cursor"], "c_xyz")
|
|
self.assertEqual(state["last_timestamp_us"], 123456789.0)
|
|
|
|
# Reload and verify
|
|
reloaded = load_state(state_file)
|
|
self.assertEqual(reloaded["last_record_number"], 100)
|
|
self.assertEqual(reloaded["last_cursor"], "c_xyz")
|
|
finally:
|
|
if os.path.exists(state_file):
|
|
try:
|
|
os.remove(state_file)
|
|
except Exception:
|
|
pass
|
|
|
|
def test_windows_log_collector_severity_mapping(self):
|
|
state = {}
|
|
collector = WindowsLogCollector(state=state, machine_id="test-win-node", channels=["Application"])
|
|
self.assertEqual(collector.machine_id, "test-win-node")
|
|
self.assertEqual(collector.channels, ["Application"])
|
|
|
|
def test_linux_log_collector_priority_normalization(self):
|
|
state = {}
|
|
collector = LinuxLogCollector(state=state, machine_id="test-linux-node")
|
|
self.assertEqual(collector.machine_id, "test-linux-node")
|
|
|
|
def test_get_log_collector_factory(self):
|
|
win_col = get_log_collector(os_type="windows", machine_id="win-1")
|
|
self.assertIsInstance(win_col, WindowsLogCollector)
|
|
self.assertEqual(win_col.machine_id, "win-1")
|
|
|
|
lin_col = get_log_collector(os_type="linux", machine_id="lin-1")
|
|
self.assertIsInstance(lin_col, LinuxLogCollector)
|
|
self.assertEqual(lin_col.machine_id, "lin-1")
|
|
|
|
auto_col = get_log_collector()
|
|
self.assertIsInstance(auto_col, BaseLogCollector)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|