LOGAR/tests/test_hub_server_deep.py

"""
Comprehensive unit tests for LOGAR Server FastAPI endpoints, state management, and log processing.
"""

import os
import sys
import json
import shutil
import tempfile
import sqlite3
import unittest
from unittest.mock import MagicMock, patch

from src.hub import Server
from src.hub.server_models import (
    ClientEnrollRequest,
    ClientActionRequest,
    HermesTokenCreateRequest,
    HermesTokenRevokeRequest
)
from src.hub.auth_manager import SessionManager


class TestHubServerDeep(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.db_path = os.path.join(self.temp_dir, "test_server.db")
        self.config_path = os.path.join(self.temp_dir, "server_config.json")

        # Initialize test state
        Server.StorageEngine.init_db(self.db_path, enrollment_secret="test_secret", max_seats=50)
        Server.SERVER_STATE["config"] = {
            "db_path": self.db_path,
            "enrollment_secret": "test_secret",
            "max_seats": 50,
            "cert_dir": self.cert_dir,
            "encrypt_at_rest": False
        }
        # Generate and load CA
        ca_cert, ca_key, ca_pem, _ = Server.enrollment.generate_ca_if_needed(cert_dir=self.cert_dir)
        Server.SERVER_STATE["ca_cert"] = ca_cert
        Server.SERVER_STATE["ca_key"] = ca_key
        Server.SERVER_STATE["ca_cert_pem"] = ca_pem

    def tearDown(self):
        shutil.rmtree(self.temp_dir, ignore_errors=True)

    def test_server_status_and_update_routes(self):
        """Tests /api/update/status and health check handler."""
        status_data = Server.get_update_status(user="admin")
        self.assertIn("hosted_packages", status_data)

        health = Server.health_check()
        self.assertEqual(health.get("status"), "healthy")

    def test_client_enrollment_and_action_endpoints(self):
        """Tests client enrollment, listing, revoke, approve, and delete handlers."""
        # Ensure license in DB is marked active with seats
        conn = sqlite3.connect(self.db_path)
        conn.execute("UPDATE license_config SET max_seats = 50, license_status = 'active' WHERE id = 1")
        conn.commit()
        conn.close()

        # Enroll client
        enroll_req = ClientEnrollRequest(
            client_id="test_agent_deep_01",
            hostname="workstation-01",
            os="windows",
            enrollment_secret="test_secret"
        )
        enroll_data = Server.enroll_client(enroll_req)
        self.assertEqual(enroll_data.get("status"), "success")
        self.assertIn("client_cert", enroll_data)
        self.assertIn("client_key", enroll_data)

        # Check client list
        clients_data = Server.list_clients(auth_user="admin")
        self.assertGreaterEqual(clients_data.get("active_seats", 0), 1)

        # Client action: revoke
        action_req = ClientActionRequest(client_id="test_agent_deep_01")
        rev_data = Server.revoke_client(action_req, auth_user="admin")
        self.assertEqual(rev_data.get("status"), "revoked")

        # Reactivate
        app_data = Server.approve_client(action_req, auth_user="admin")
        self.assertEqual(app_data.get("status"), "active")

        # Delete
        del_data = Server.delete_client(action_req, auth_user="admin")
        self.assertEqual(del_data.get("status"), "deleted")

    def test_hermes_tokens_and_auth_routes(self):
        """Tests Hermes API token creation, listing, and revoking handlers."""
        # Create token
        tok_req = HermesTokenCreateRequest(name="Test Integration Agent")
        tok_data = Server.api_create_hermes_token(tok_req, auth_user="admin")
        token_val = tok_data.get("token")
        self.assertIsNotNone(token_val)

        # List tokens
        tokens_list = Server.api_list_hermes_tokens(auth_user="admin")
        self.assertTrue(any(t.get("token") == token_val for t in tokens_list))

        # Revoke token
        rev_req = HermesTokenRevokeRequest(token=token_val)
        rev_data = Server.api_revoke_hermes_token(rev_req, auth_user="admin")
        self.assertEqual(rev_data.get("status"), "success")

    def test_process_ingested_logs_and_temporal_tracking(self):
        """Tests processing ingested log entries, deduplication, and anomaly promotion."""
        payload = {
            "server": "web-prod-01",
            "timestamp": "2026-01-01T12:00:00Z",
            "logs": [
                {
                    "site_name": "Default-Site",
                    "server": "web-prod-01",
                    "signature": "SECURITY_4625_BRUTEFORCE",
                    "severity": "CRITICAL",
                    "message": "Repeated failed logon attempts detected from 10.0.0.99",
                    "os_type": "windows",
                    "timestamp": "2026-01-01T12:00:00Z"
                }
            ]
        }
        res = Server.process_ingested_logs(
            payload=payload,
            db_path=self.db_path,
            window_hours=24,
            min_runs=1
        )
        self.assertEqual(res.get("processed"), 1)

        # Query database directly to confirm issue recorded
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        c.execute("SELECT signature, severity, run_count FROM active_issues")
        row = c.fetchone()
        conn.close()
        self.assertIsNotNone(row)
        self.assertEqual(row[0], "SECURITY_4625_BRUTEFORCE")
        self.assertEqual(row[1], "CRITICAL")
        self.assertEqual(row[2], 1)


if __name__ == "__main__":
    unittest.main()