me0nline
df9f5d191b
CI Test Suite / Run Component Tests & Pipeline Verification (push) Failing after 1m18s
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
"""
|
|
Comprehensive unit tests for LOGAR Client Transport, Windows & Linux forwarders, and update checks.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from src.clients import client_transport
|
|
from src.clients import Win_Client, Linux_Client
|
|
|
|
|
|
class TestClientsDeep(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.cert_dir = os.path.join(self.temp_dir, "certs")
|
|
os.makedirs(self.cert_dir, exist_ok=True)
|
|
self.config_path = os.path.join(self.temp_dir, "client_config.json")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
|
|
|
def test_client_transport_config_loader_and_helpers(self):
|
|
"""Tests client configuration loading, writable directory fallback, and path resolution."""
|
|
sample_config = {
|
|
"server_host": "127.0.0.1",
|
|
"server_port": 9443,
|
|
"hermes_port": 8443,
|
|
"cert_dir": self.cert_dir,
|
|
"enrollment_secret": "test_enroll_secret_2026",
|
|
"auth_token": "token_123",
|
|
"ca_cert": "-----BEGIN CERTIFICATE-----\nMIIB...\n-----END CERTIFICATE-----\n"
|
|
}
|
|
with open(self.config_path, "w", encoding="utf-8") as f:
|
|
json.dump(sample_config, f)
|
|
|
|
cfg = client_transport.load_config(self.config_path)
|
|
self.assertEqual(cfg["server_host"], "127.0.0.1")
|
|
self.assertEqual(cfg["server_port"], 9443)
|
|
self.assertTrue(os.path.exists(os.path.join(cfg["cert_dir"], "ca.crt")))
|
|
|
|
# Test cert expiry check on non-existent cert
|
|
self.assertTrue(client_transport.is_cert_expiring_soon(os.path.join(self.cert_dir, "nonexistent.crt")))
|
|
|
|
def test_client_transport_ensure_enrolled(self):
|
|
"""Tests proactive client enrollment bootstrap check."""
|
|
cfg = {
|
|
"server_host": "127.0.0.1",
|
|
"server_port": 9443,
|
|
"hermes_host": "127.0.0.1",
|
|
"hermes_port": 8443,
|
|
"cert_dir": self.cert_dir,
|
|
"enrollment_secret": "secret"
|
|
}
|
|
|
|
with patch("src.clients.client_transport.enroll_client_if_needed") as mock_enroll:
|
|
mock_enroll.return_value = True
|
|
ok = client_transport.ensure_enrolled(cfg, "machine_01", os_type="windows")
|
|
self.assertTrue(ok)
|
|
mock_enroll.assert_called_once()
|
|
|
|
def test_client_transport_openpgp_fallback_transmission(self):
|
|
"""Tests fallback to OpenPGP envelope over plain TCP socket when certs are absent."""
|
|
from src.hub.storage_engine import StorageCryptoManager
|
|
priv, pub, fp = StorageCryptoManager.generate_server_keypair("Hub Test Key")
|
|
|
|
cfg = {
|
|
"server_host": "127.0.0.1",
|
|
"server_port": 9443,
|
|
"cert_dir": os.path.join(self.temp_dir, "empty_certs"),
|
|
"server_public_key": pub,
|
|
"server_fingerprint": fp,
|
|
"auth_token": "client_token_abc"
|
|
}
|
|
|
|
mock_socket = MagicMock()
|
|
# Mock server returning 4-byte big-endian length + JSON response
|
|
resp_data = json.dumps({"status": "success", "processed": 5}).encode("utf-8")
|
|
resp_len_header = len(resp_data).to_bytes(4, byteorder="big")
|
|
|
|
mock_socket.recv.side_effect = [resp_len_header, resp_data, b""]
|
|
|
|
with patch("socket.socket") as mock_sock_cls:
|
|
mock_sock_cls.return_value.__enter__.return_value = mock_socket
|
|
logs = [{"timestamp": "2026-01-01T00:00:00Z", "event_id": 4625, "message": "Failed logon"}]
|
|
res = client_transport.send_encrypted_logs_over_socket(cfg, logs, machine_id="test_win_node")
|
|
self.assertEqual(res.get("status"), "success")
|
|
self.assertEqual(res.get("processed"), 5)
|
|
|
|
def test_win_client_and_linux_client_entrypoints(self):
|
|
"""Tests command-line parsing, help options, and diagnostic checks."""
|
|
# Test help flag execution
|
|
with self.assertRaises(SystemExit) as cm:
|
|
with patch("sys.argv", ["Win_Client.py", "--help"]):
|
|
Win_Client.main()
|
|
self.assertEqual(cm.exception.code, 0)
|
|
|
|
with self.assertRaises(SystemExit) as cm:
|
|
with patch("sys.argv", ["Linux_Client.py", "--help"]):
|
|
Linux_Client.main()
|
|
self.assertEqual(cm.exception.code, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|