75 lines
4.0 KiB
Python
75 lines
4.0 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fgai.config import ConfigStore
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def test_public_config_hides_token_and_preserves_blank_update(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
store.update({"log_source": "graylog_mcp", "graylog_mcp_token": "secret"})
|
|
public = store.update({"graylog_mcp_token": ""})
|
|
self.assertEqual(public["log_source"], "graylog_mcp")
|
|
self.assertTrue(public["graylog_mcp_token_configured"])
|
|
self.assertNotIn("graylog_mcp_token", public)
|
|
|
|
def test_graylog_range_seconds_is_numeric_and_bounded(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"graylog_range_seconds": "30"})
|
|
self.assertEqual(public["graylog_range_seconds"], 60)
|
|
public = store.update({"graylog_range_seconds": "7200"})
|
|
self.assertEqual(public["graylog_range_seconds"], 7200)
|
|
|
|
def test_graylog_max_events_per_stream_is_numeric_and_bounded(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"graylog_max_events_per_stream": "0"})
|
|
self.assertEqual(public["graylog_max_events_per_stream"], 1)
|
|
public = store.update({"graylog_max_events_per_stream": "25000"})
|
|
self.assertEqual(public["graylog_max_events_per_stream"], 25000)
|
|
|
|
def test_graylog_fetch_mode_and_raw_sample_events_are_saved(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"graylog_fetch_mode": "aggregate", "graylog_raw_sample_events": "2500"})
|
|
self.assertEqual(public["graylog_fetch_mode"], "aggregate")
|
|
self.assertEqual(public["graylog_raw_sample_events"], 2500)
|
|
|
|
def test_graylog_mcp_timeout_settings_are_numeric_and_bounded(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"graylog_mcp_call_timeout_seconds": "0", "graylog_mcp_poll_timeout_seconds": "30"})
|
|
self.assertEqual(public["graylog_mcp_call_timeout_seconds"], 1)
|
|
self.assertEqual(public["graylog_mcp_poll_timeout_seconds"], 60)
|
|
public = store.update({"graylog_mcp_call_timeout_seconds": "6", "graylog_mcp_poll_timeout_seconds": "180"})
|
|
self.assertEqual(public["graylog_mcp_call_timeout_seconds"], 6)
|
|
self.assertEqual(public["graylog_mcp_poll_timeout_seconds"], 180)
|
|
|
|
def test_llm_timeout_setting_is_numeric_and_bounded(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"llm_timeout": "0"})
|
|
self.assertEqual(public["llm_timeout"], 1)
|
|
public = store.update({"llm_timeout": "300"})
|
|
self.assertEqual(public["llm_timeout"], 300)
|
|
|
|
def test_graylog_tls_verify_can_be_disabled(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"graylog_tls_verify": False})
|
|
self.assertFalse(public["graylog_tls_verify"])
|
|
|
|
def test_empty_stream_update_does_not_clear_existing_streams(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
store.update({"graylog_streams": [{"id": "stream-1", "title": "Firewall", "enabled": True}]})
|
|
public = store.update({"graylog_streams": []})
|
|
self.assertEqual(public["graylog_streams"], [{"id": "stream-1", "title": "Firewall", "enabled": True}])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|