operator-main
105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
test_variables_engine.py — Unit and integration tests for:
|
|
1. Schema validation (int, float, bool, string, ranges, choices)
|
|
2. Dual-synced atomic persistence between local and /srv/box/
|
|
3. Append-only audit history logging to variables-history.jsonl
|
|
4. Atomic rollback to previous revisions
|
|
5. box-ctl.py allowlisted RPC actions for variables
|
|
"""
|
|
import unittest
|
|
import subprocess
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
|
BIN_DIR = REPO_ROOT / "bin"
|
|
VARS_PY = BIN_DIR / "variables.py"
|
|
BOX_CTL = BIN_DIR / "box-ctl.py"
|
|
|
|
sys.path.insert(0, str(BIN_DIR))
|
|
import variables
|
|
|
|
|
|
class TestVariablesEngine(unittest.TestCase):
|
|
"""Test schema enforcement and in-memory variable operations."""
|
|
|
|
def setUp(self):
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
self.var_file = Path(self.tmp_dir) / "variables.json"
|
|
self.hist_file = Path(self.tmp_dir) / "variables-history.jsonl"
|
|
self.engine = variables.Variables(path=str(self.var_file), history_path=str(self.hist_file))
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp_dir, ignore_errors=True)
|
|
|
|
def test_default_values_load(self):
|
|
val = self.engine.get("loop_health_threshold")
|
|
self.assertEqual(val, 0.5)
|
|
val_sec = self.engine.get("loop_health_interval_s")
|
|
self.assertEqual(val_sec, 900)
|
|
|
|
def test_schema_validation_out_of_range(self):
|
|
# loop_health_threshold range: 0.0 .. 1.0
|
|
with self.assertRaises(ValueError):
|
|
self.engine.set("loop_health_threshold", 1.5)
|
|
with self.assertRaises(ValueError):
|
|
self.engine.set("loop_health_threshold", -0.1)
|
|
|
|
def test_schema_validation_type_coercion(self):
|
|
self.engine.set("loop_health_interval_s", "1200")
|
|
self.assertEqual(self.engine.get("loop_health_interval_s"), 1200)
|
|
|
|
def test_history_logging_and_rollback(self):
|
|
initial = self.engine.get("max_nudge_count")
|
|
self.engine.set("max_nudge_count", 5, by="operator_test")
|
|
self.assertEqual(self.engine.get("max_nudge_count"), 5)
|
|
|
|
# Check history recorded
|
|
hist = self.engine.history(name="max_nudge_count", limit=5)
|
|
self.assertGreaterEqual(len(hist), 1)
|
|
self.assertEqual(hist[0]["action"], "set")
|
|
self.assertEqual(hist[0]["new_value"], 5)
|
|
|
|
# Rollback to initial
|
|
rb_val = self.engine.rollback("max_nudge_count", by="operator_test")
|
|
self.assertEqual(rb_val, initial)
|
|
self.assertEqual(self.engine.get("max_nudge_count"), initial)
|
|
|
|
def test_reset_variable(self):
|
|
self.engine.set("sweeper_interval_s", 120)
|
|
self.assertEqual(self.engine.get("sweeper_interval_s"), 120)
|
|
res_val = self.engine.reset("sweeper_interval_s")
|
|
self.assertEqual(res_val, 60)
|
|
self.assertEqual(self.engine.get("sweeper_interval_s"), 60)
|
|
|
|
|
|
|
|
class TestVariablesCliAndRpc(unittest.TestCase):
|
|
"""Test CLI commands and box-ctl RPC actions for variables."""
|
|
|
|
def test_box_ctl_vars_list(self):
|
|
cmd = [sys.executable, str(BOX_CTL), "vars-list"]
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(res.returncode, 0)
|
|
data = json.loads(res.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
self.assertIn("variables", data)
|
|
self.assertIn("loop_health_threshold", data["variables"])
|
|
|
|
def test_box_ctl_vars_history(self):
|
|
cmd = [sys.executable, str(BOX_CTL), "vars-history", "loop_health_threshold", "5"]
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(res.returncode, 0)
|
|
data = json.loads(res.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
self.assertIsInstance(data.get("history"), list)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|