""" Comprehensive unit tests for LOGAR UpdateManager, AuthManager, and IngestQueue. """ import os import sys import time import shutil import tempfile import unittest from unittest.mock import MagicMock, patch from src.hub.update_manager import ServerUpdateManager from src.hub.auth_manager import PasswordManager, TOTPManager, SessionManager, RateLimiter from src.hub.ingest_queue import IngestQueueManager class TestHubUpdateAuthDeep(unittest.TestCase): def setUp(self): self.temp_dir = tempfile.mkdtemp() self.update_dir = os.path.join(self.temp_dir, "updates") self.db_path = os.path.join(self.temp_dir, "auth_test.db") def tearDown(self): shutil.rmtree(self.temp_dir, ignore_errors=True) def test_update_manager_caching_and_manifest(self): """Tests package registration, SHA256 integrity, local caching, and download serving.""" manager = ServerUpdateManager(update_dir=self.update_dir, check_enabled=False) # Create dummy binaries win_bin = os.path.join(self.temp_dir, "Win_Client.exe") linux_bin = os.path.join(self.temp_dir, "Linux_Client") with open(win_bin, "wb") as f: f.write(b"MZ_TEST_WIN_CLIENT_BINARY_2026") with open(linux_bin, "wb") as f: f.write(b"\x7fELF_TEST_LINUX_CLIENT_BINARY_2026") # Register packages res_win = manager.register_client_package( os_type="windows", file_path=win_bin, version="2.5.0", filename="Win_Client.exe" ) self.assertEqual(res_win.get("version"), "2.5.0") self.assertEqual(res_win.get("os"), "windows") self.assertIsNotNone(res_win.get("sha256")) res_linux = manager.register_client_package( os_type="linux", file_path=linux_bin, version="2.5.0", filename="Linux_Client" ) self.assertEqual(res_linux.get("os"), "linux") # Get hosted packages packages = manager.get_hosted_packages() self.assertIn("windows", packages) self.assertIn("linux", packages) self.assertEqual(packages["windows"]["version"], "2.5.0") # Test download path resolution pkg_res = manager.get_client_package_path(os_type="windows") self.assertIsNotNone(pkg_res) win_path, win_filename, win_sha = pkg_res self.assertTrue(os.path.exists(win_path)) self.assertEqual(win_filename, "Win_Client.exe") # Test update decision logic decision = manager.check_client_update(client_id="node-win-01", current_version="1.0.0", os_type="windows") self.assertTrue(decision.get("update_available")) self.assertEqual(decision.get("latest_version"), "2.5.0") # Test download and caching from simulated upstream assets mock_assets = [ {"name": "LOGAR_Client_Setup.exe", "download_url": "http://mock.upstream/LOGAR_Client_Setup.exe"}, {"name": "Linux_Client.bin", "download_url": "http://mock.upstream/Linux_Client.bin"} ] class MockHttpResp: status = 200 def __init__(self): self._read_done = False def __enter__(self): return self def __exit__(self, *args): pass def read(self, size=65536): if not self._read_done: self._read_done = True return b"MOCK_BINARY_DATA_PAYLOAD_2026" return b"" with patch("urllib.request.urlopen", side_effect=lambda req, timeout=60: MockHttpResp()): cache_res = manager.download_and_cache_client_packages(assets=mock_assets, target_version="2.6.0") self.assertTrue(cache_res["windows"]) self.assertTrue(cache_res["linux"]) self.assertEqual(len(cache_res["downloaded"]), 2) self.assertEqual(manager.hosted_client_version, "2.6.0") def test_auth_manager_components(self): """Tests PasswordManager, TOTPManager, SessionManager, and RateLimiter.""" # 1. PasswordManager pwd = "Enterprise_Admin_Password_!2026" pwd_hash, salt = PasswordManager.hash_password(pwd) self.assertTrue(PasswordManager.verify_password(pwd, salt, pwd_hash)) self.assertFalse(PasswordManager.verify_password("WrongPassword", salt, pwd_hash)) # 2. TOTPManager secret = TOTPManager.generate_totp_secret() self.assertTrue(len(secret) >= 16) uri = TOTPManager.get_totp_uri("admin", secret, "LOGAR Fleet Hub") self.assertIn("otpauth://totp/", uri) token = TOTPManager.get_totp_token(secret) self.assertEqual(len(token), 6) self.assertTrue(TOTPManager.verify_totp_code(secret, token)) self.assertFalse(TOTPManager.verify_totp_code(secret, "000000" if token != "000000" else "111111")) # 3. SessionManager session_token = SessionManager.generate_session_token("admin", db_path=self.db_path) self.assertIsNotNone(session_token) username = SessionManager.verify_session_token(session_token, db_path=self.db_path) self.assertEqual(username, "admin") self.assertIsNone(SessionManager.verify_session_token("invalid:token:format", db_path=self.db_path)) # 4. RateLimiter limiter = RateLimiter() ip = "192.168.1.100" for _ in range(5): self.assertTrue(limiter.check_ip_rate_limit(ip, max_attempts=5, window_seconds=60, db_path=self.db_path)) limiter.record_failed_ip_attempt(ip, db_path=self.db_path) # 6th attempt exceeded self.assertFalse(limiter.check_ip_rate_limit(ip, max_attempts=5, window_seconds=60, db_path=self.db_path)) # Reset limiter.reset_failed_ip_attempts(ip, db_path=self.db_path) self.assertTrue(limiter.check_ip_rate_limit(ip, max_attempts=5, window_seconds=60, db_path=self.db_path)) def test_ingest_queue_lifecycle_and_batching(self): """Tests IngestQueueManager push, stats, and flush callback.""" processed_items = [] def mock_processor(payload, db_path, window_hours, min_runs): processed_items.extend(payload.get("logs", [])) return {"processed": len(payload.get("logs", [])), "promoted_verified": 0} queue = IngestQueueManager(flush_interval=0.05, batch_size=10, max_queue=100) # Enqueue items for i in range(15): pushed = queue.push({"server": "node1", "logs": [{"id": i, "data": f"log_{i}"}]}) self.assertTrue(pushed) stats = queue.get_stats() self.assertEqual(stats["queue_size"], 15) self.assertEqual(stats["total_ingested"], 15) # Flush flush_res = queue.flush(db_path=self.db_path, process_callback=mock_processor) self.assertEqual(flush_res["status"], "success") self.assertEqual(flush_res["flushed_runs"], 15) self.assertEqual(flush_res["processed_logs"], 15) self.assertEqual(len(processed_items), 15) stats_after = queue.get_stats() self.assertEqual(stats_after["queue_size"], 0) self.assertEqual(stats_after["total_flushed"], 15) if __name__ == "__main__": unittest.main()