bf1109be81
Feedback rules can now be suggested for merging when a human has already reviewed the same pattern multiple times (never guesses about new values). LLM defaults sized for constrained VRAM instead of llama3.3/llama3.1. Adds systemd units, a state backup script, and reverse-proxy auth guidance for production deployment. Findings view is collapsible and sorts suggested-review rows to the top for a quicker overview.
105 lines
5.9 KiB
Python
105 lines
5.9 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_profile_advisor_interval_seconds_is_numeric_and_bounded(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"profile_advisor_interval_seconds": "10"})
|
|
self.assertEqual(public["profile_advisor_interval_seconds"], 60)
|
|
public = store.update({"profile_advisor_interval_seconds": "600"})
|
|
self.assertEqual(public["profile_advisor_interval_seconds"], 600)
|
|
|
|
def test_field_confidence_threshold_is_numeric_and_clamped(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"field_confidence_threshold": "-5"})
|
|
self.assertEqual(public["field_confidence_threshold"], 0)
|
|
public = store.update({"field_confidence_threshold": "500"})
|
|
self.assertEqual(public["field_confidence_threshold"], 100)
|
|
public = store.update({"field_confidence_threshold": "70"})
|
|
self.assertEqual(public["field_confidence_threshold"], 70)
|
|
public = store.update({"field_confidence_threshold": "not-a-number"})
|
|
self.assertEqual(public["field_confidence_threshold"], 70)
|
|
|
|
def test_feedback_generalization_min_values_is_numeric_and_clamped(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = ConfigStore(str(Path(directory) / "config.json"))
|
|
public = store.update({"feedback_generalization_min_values": "1"})
|
|
self.assertEqual(public["feedback_generalization_min_values"], 2)
|
|
public = store.update({"feedback_generalization_min_values": "500"})
|
|
self.assertEqual(public["feedback_generalization_min_values"], 20)
|
|
public = store.update({"feedback_generalization_min_values": "5"})
|
|
self.assertEqual(public["feedback_generalization_min_values"], 5)
|
|
|
|
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()
|