LOGAR/tests/test_server.py

import os
import sys
import json
import socket
import struct
import sqlite3
import unittest
import urllib.request
import warnings
from datetime import datetime, timezone, timedelta

warnings.filterwarnings("ignore")

# Ensure parent directory and src directory are in path to import Server
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")))

import Server
import pgpy


class TestServerComponent(unittest.TestCase):
    def setUp(self):
        self.test_db = "test_server_state.db"
        self.test_config = "test_server_config.json"
        if os.path.exists(self.test_db):
            os.remove(self.test_db)
        if os.path.exists(self.test_config):
            os.remove(self.test_config)

    def tearDown(self):
        if os.path.exists(self.test_db):
            try:
                os.remove(self.test_db)
            except Exception:
                pass
        if os.path.exists(self.test_config):
            try:
                os.remove(self.test_config)
            except Exception:
                pass

    def test_first_run_config_and_keypair_generation(self):
        config = Server.load_or_init_config(self.test_config)
        self.assertTrue(os.path.exists(self.test_config))
        self.assertIn("server_fingerprint", config)
        self.assertIn("public_key", config)
        self.assertIn("private_key", config)
        self.assertIn("auth_token", config)
        self.assertNotIn("site_name", config)

        # Verify keypair
        priv_key, _ = pgpy.PGPKey.from_blob(config["private_key"])
        pub_key, _ = pgpy.PGPKey.from_blob(config["public_key"])
        self.assertEqual(str(pub_key.fingerprint), config["server_fingerprint"])

    def test_create_client_config(self):
        Server.load_or_init_config(self.test_config)
        client_out = "test_client_out.json"
        try:
            client_conf = Server.create_client_config(
                server_host="10.0.0.1",
                server_port=9443,
                output_path=client_out,
                config_path=self.test_config
            )
            self.assertTrue(os.path.exists(client_out))
            self.assertEqual(client_conf["server_host"], "10.0.0.1")
            self.assertEqual(client_conf["server_port"], 9443)
            # Verify no machine name or site_name is included
            self.assertNotIn("server_name", client_conf)
            self.assertNotIn("name", client_conf)
            self.assertNotIn("site_name", client_conf)
        finally:
            if os.path.exists(client_out):
                os.remove(client_out)

    def test_4_run_rule_and_12h_window(self):
        Server.init_db(self.test_db)
        log_entry = {
            "server": "app-worker-01.corp.local",
            "signature": "PostgresConnTimeout",
            "severity": "ERROR",
            "message": "Connection to database pool timed out after 30s",
            "os_type": "linux"
        }
        payload = {
            "server": "app-worker-01.corp.local",
            "logs": [log_entry]
        }

        # Runs 1 to 3: should remain TRANSIENT
        for run_idx in range(1, 4):
            res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4)
            self.assertEqual(res["status"], "success")
            self.assertEqual(res["promoted_verified"], 0)

        conn = sqlite3.connect(self.test_db)
        c = conn.cursor()
        c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnTimeout",))
        row = c.fetchone()
        conn.close()
        self.assertEqual(row[0], 3)
        self.assertEqual(row[1], "TRANSIENT")

        # Run 4: promotes to VERIFIED!
        res4 = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4)
        self.assertEqual(res4["promoted_verified"], 1)

        conn = sqlite3.connect(self.test_db)
        c = conn.cursor()
        c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnTimeout",))
        row = c.fetchone()
        conn.close()
        self.assertEqual(row[0], 4)
        self.assertEqual(row[1], "VERIFIED")

    def test_server_severity_filtering(self):
        Server.init_db(self.test_db)
        payload = {
            "server": "app-worker-01.corp.local",
            "logs": [
                {"server": "app-worker-01", "signature": "SigInfo", "severity": "INFO", "message": "Info msg", "os_type": "linux"},
                {"server": "app-worker-01", "signature": "SigWarn", "severity": "WARNING", "message": "Warn msg", "os_type": "linux"},
                {"server": "app-worker-01", "signature": "SigErr", "severity": "ERROR", "message": "Err msg", "os_type": "linux"},
                {"server": "app-worker-01", "signature": "SigDebug", "severity": "DEBUG", "message": "Debug msg", "os_type": "linux"},
                {"server": "app-worker-01", "signature": "SigTrace", "severity": "TRACE", "message": "Trace msg", "os_type": "linux"}
            ]
        }
        res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4)
        self.assertEqual(res["status"], "success")

        conn = sqlite3.connect(self.test_db)
        c = conn.cursor()
        c.execute("SELECT signature FROM active_issues ORDER BY signature")
        sigs = [r[0] for r in c.fetchall()]
        conn.close()

        self.assertIn("SigInfo", sigs)
        self.assertIn("SigWarn", sigs)
        self.assertIn("SigErr", sigs)
        self.assertNotIn("SigDebug", sigs)
        self.assertNotIn("SigTrace", sigs)


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