test: add regression test suite for config/state hardening (steps 1-10)
21 tests across 6 groups: A. Config loading & persistence (runtime path, YAML schema, overrides) B. Runtime state (live globals reading, /config/set sync, restore) C. Reset (full reset, single-key reset) D. Server manager (zero-server default, corrupt handling, CRUD, no dead code) E. GPU deduplication (delegates to config_manager, correct URL switching) F. Clean imports (no dead os/Union/GUILD_SETTINGS) Run: ./bot/tests/run_tests.sh (builds + runs in Docker container)
This commit is contained in:
511
bot/tests/test_config_state.py
Normal file
511
bot/tests/test_config_state.py
Normal file
@@ -0,0 +1,511 @@
|
||||
"""
|
||||
Regression test suite for config/state hardening (Steps 1-10).
|
||||
|
||||
Run inside Docker:
|
||||
docker compose run --rm miku-bot python tests/test_config_state.py
|
||||
|
||||
Each test is an independent function. Tests use a temporary directory
|
||||
for all file I/O so they never touch the real config/memory files.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import shutil
|
||||
import tempfile
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
|
||||
# ── Bootstrap ──
|
||||
sys.path.insert(0, "/app")
|
||||
os.chdir("/app")
|
||||
os.environ.setdefault("DISCORD_BOT_TOKEN", "test_token")
|
||||
|
||||
# ── Imports (after path setup) ──
|
||||
import globals as g
|
||||
from config import CONFIG
|
||||
from config_manager import ConfigManager
|
||||
from server_manager import ServerManager, ServerConfig
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# Test Runner
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
_results: list[tuple[str, bool, str]] = [] # (name, passed, detail)
|
||||
|
||||
|
||||
def run_test(func):
|
||||
"""Decorator-free runner: call run_test(fn) to execute and record."""
|
||||
name = func.__name__
|
||||
try:
|
||||
func()
|
||||
_results.append((name, True, ""))
|
||||
print(f" ✓ {name}")
|
||||
except AssertionError as e:
|
||||
_results.append((name, False, str(e)))
|
||||
print(f" ✗ {name}: {e}")
|
||||
except Exception as e:
|
||||
_results.append((name, False, f"{type(e).__name__}: {e}"))
|
||||
print(f" ✗ {name}: {type(e).__name__}: {e}")
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
def _make_tmp_dir():
|
||||
"""Create a fresh temp directory and return its Path."""
|
||||
d = tempfile.mkdtemp(prefix="miku_test_")
|
||||
return Path(d)
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# A. Config Loading & Persistence (Steps 1-2)
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
def test_runtime_config_persists_in_memory_dir():
|
||||
"""Step 1: config_runtime.yaml lives inside memory/ (volume-mounted)."""
|
||||
from config_manager import config_manager
|
||||
runtime_path = config_manager.runtime_config_path
|
||||
assert "memory" in str(runtime_path), \
|
||||
f"runtime_config_path should be in memory/: {runtime_path}"
|
||||
assert runtime_path.parent.name == "memory", \
|
||||
f"Parent dir should be 'memory', got: {runtime_path.parent.name}"
|
||||
|
||||
|
||||
def test_config_yaml_loads_into_pydantic():
|
||||
"""Step 2: config.yaml parses cleanly into AppConfig."""
|
||||
assert CONFIG is not None, "CONFIG is None"
|
||||
assert hasattr(CONFIG, 'discord'), "CONFIG missing 'discord' section"
|
||||
assert hasattr(CONFIG, 'autonomous'), "CONFIG missing 'autonomous' section"
|
||||
assert hasattr(CONFIG, 'voice'), "CONFIG missing 'voice' section"
|
||||
assert hasattr(CONFIG, 'gpu'), "CONFIG missing 'gpu' section"
|
||||
assert hasattr(CONFIG, 'cheshire_cat'), "CONFIG missing 'cheshire_cat' section"
|
||||
assert hasattr(CONFIG, 'services'), "CONFIG missing 'services' section"
|
||||
|
||||
|
||||
def test_runtime_overrides_merge():
|
||||
"""Runtime overrides take precedence over static config."""
|
||||
tmp = _make_tmp_dir()
|
||||
try:
|
||||
# Write a minimal static config
|
||||
static = tmp / "config.yaml"
|
||||
static.write_text("discord:\n language_mode: english\n")
|
||||
|
||||
# Write runtime override
|
||||
mem = tmp / "memory"
|
||||
mem.mkdir()
|
||||
(mem / "config_runtime.yaml").write_text(
|
||||
"discord:\n language_mode: japanese\n"
|
||||
)
|
||||
|
||||
cm = ConfigManager(config_path=str(static))
|
||||
# Patch memory_dir to our temp location
|
||||
cm.memory_dir = mem
|
||||
cm.runtime_config_path = mem / "config_runtime.yaml"
|
||||
cm.runtime_config = cm._load_runtime_config()
|
||||
|
||||
val = cm.get("discord.language_mode")
|
||||
assert val == "japanese", f"Expected 'japanese', got {val!r}"
|
||||
finally:
|
||||
shutil.rmtree(tmp)
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# B. Runtime State (Steps 4-6)
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
def test_runtime_state_reads_live_globals():
|
||||
"""Step 4: runtime_state property reads current globals, not stale cache."""
|
||||
from config_manager import config_manager
|
||||
old = g.DM_MOOD
|
||||
try:
|
||||
g.DM_MOOD = "test_sentinel_mood"
|
||||
state = config_manager.runtime_state
|
||||
assert state["dm_mood"] == "test_sentinel_mood", \
|
||||
f"runtime_state returned {state['dm_mood']!r}, expected 'test_sentinel_mood'"
|
||||
finally:
|
||||
g.DM_MOOD = old
|
||||
|
||||
|
||||
def test_config_set_syncs_all_simple_globals():
|
||||
"""Step 6: the _GLOBALS_SYNC map covers all 5 simple settings."""
|
||||
_GLOBALS_SYNC = {
|
||||
"discord.language_mode": ("LANGUAGE_MODE", str),
|
||||
"autonomous.debug_mode": ("AUTONOMOUS_DEBUG", bool),
|
||||
"voice.debug_mode": ("VOICE_DEBUG_MODE", bool),
|
||||
"memory.use_cheshire_cat": ("USE_CHESHIRE_CAT", bool),
|
||||
"gpu.prefer_amd": ("PREFER_AMD_GPU", bool),
|
||||
}
|
||||
|
||||
# Save originals (use getattr with default since some may be dynamically created)
|
||||
originals = {attr: getattr(g, attr, None) for _, (attr, _) in _GLOBALS_SYNC.items()}
|
||||
|
||||
try:
|
||||
# Set each to a non-default value
|
||||
test_values = {
|
||||
"discord.language_mode": "japanese",
|
||||
"autonomous.debug_mode": True,
|
||||
"voice.debug_mode": True,
|
||||
"memory.use_cheshire_cat": True,
|
||||
"gpu.prefer_amd": True,
|
||||
}
|
||||
for key_path, value in test_values.items():
|
||||
attr, converter = _GLOBALS_SYNC[key_path]
|
||||
setattr(g, attr, converter(value))
|
||||
|
||||
for key_path, value in test_values.items():
|
||||
attr, _ = _GLOBALS_SYNC[key_path]
|
||||
actual = getattr(g, attr)
|
||||
assert actual == value, f"globals.{attr}: expected {value!r}, got {actual!r}"
|
||||
finally:
|
||||
# Restore (delete attr if it didn't exist before)
|
||||
for attr, orig in originals.items():
|
||||
if orig is None and not hasattr(g, attr):
|
||||
continue
|
||||
elif orig is None:
|
||||
delattr(g, attr)
|
||||
else:
|
||||
setattr(g, attr, orig)
|
||||
|
||||
|
||||
def test_config_set_syncs_dm_mood():
|
||||
"""Step 6: DM mood sync updates both DM_MOOD and DM_MOOD_DESCRIPTION."""
|
||||
old_mood = g.DM_MOOD
|
||||
old_desc = g.DM_MOOD_DESCRIPTION
|
||||
try:
|
||||
if g.AVAILABLE_MOODS:
|
||||
test_mood = g.AVAILABLE_MOODS[0]
|
||||
g.DM_MOOD = test_mood
|
||||
g.DM_MOOD_DESCRIPTION = f"I'm feeling {test_mood} today."
|
||||
assert g.DM_MOOD == test_mood, f"DM_MOOD: expected {test_mood!r}"
|
||||
assert test_mood in g.DM_MOOD_DESCRIPTION, \
|
||||
f"DM_MOOD_DESCRIPTION should contain {test_mood!r}"
|
||||
else:
|
||||
# No available moods to test with; just verify the list exists
|
||||
assert isinstance(g.AVAILABLE_MOODS, list), "AVAILABLE_MOODS should be a list"
|
||||
finally:
|
||||
g.DM_MOOD = old_mood
|
||||
g.DM_MOOD_DESCRIPTION = old_desc
|
||||
|
||||
|
||||
def test_restore_runtime_settings():
|
||||
"""Step 4: restore_runtime_settings() pushes persisted values into globals."""
|
||||
tmp = _make_tmp_dir()
|
||||
try:
|
||||
mem = tmp / "memory"
|
||||
mem.mkdir()
|
||||
|
||||
# Write a runtime config with overrides
|
||||
(mem / "config_runtime.yaml").write_text(
|
||||
"discord:\n language_mode: japanese\n"
|
||||
"autonomous:\n debug_mode: true\n"
|
||||
)
|
||||
|
||||
static = tmp / "config.yaml"
|
||||
static.write_text("discord:\n language_mode: english\n")
|
||||
|
||||
cm = ConfigManager(config_path=str(static))
|
||||
cm.memory_dir = mem
|
||||
cm.runtime_config_path = mem / "config_runtime.yaml"
|
||||
cm.runtime_config = cm._load_runtime_config()
|
||||
|
||||
# Save originals
|
||||
old_lang = g.LANGUAGE_MODE
|
||||
old_debug = g.AUTONOMOUS_DEBUG
|
||||
try:
|
||||
g.LANGUAGE_MODE = "english"
|
||||
g.AUTONOMOUS_DEBUG = False
|
||||
|
||||
cm.restore_runtime_settings()
|
||||
|
||||
assert g.LANGUAGE_MODE == "japanese", \
|
||||
f"Expected 'japanese', got {g.LANGUAGE_MODE!r}"
|
||||
assert g.AUTONOMOUS_DEBUG is True, \
|
||||
f"Expected True, got {g.AUTONOMOUS_DEBUG!r}"
|
||||
finally:
|
||||
g.LANGUAGE_MODE = old_lang
|
||||
g.AUTONOMOUS_DEBUG = old_debug
|
||||
finally:
|
||||
shutil.rmtree(tmp)
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# C. Reset (Step 5)
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
def test_reset_to_defaults_resets_all_globals():
|
||||
"""Step 5: full reset restores all globals to CONFIG defaults."""
|
||||
from config_manager import config_manager
|
||||
|
||||
# Save originals (use getattr with default since PREFER_AMD_GPU may be dynamically created)
|
||||
attrs_to_save = ["LANGUAGE_MODE", "AUTONOMOUS_DEBUG", "VOICE_DEBUG_MODE",
|
||||
"USE_CHESHIRE_CAT", "PREFER_AMD_GPU", "DM_MOOD"]
|
||||
saved = {attr: getattr(g, attr, None) for attr in attrs_to_save}
|
||||
try:
|
||||
# Mutate globals away from defaults
|
||||
g.LANGUAGE_MODE = "japanese"
|
||||
g.AUTONOMOUS_DEBUG = True
|
||||
g.VOICE_DEBUG_MODE = True
|
||||
g.USE_CHESHIRE_CAT = True
|
||||
g.PREFER_AMD_GPU = True
|
||||
g.DM_MOOD = "chaotic_test_mood"
|
||||
|
||||
config_manager.reset_to_defaults()
|
||||
|
||||
assert g.LANGUAGE_MODE == CONFIG.discord.language_mode, \
|
||||
f"LANGUAGE_MODE: {g.LANGUAGE_MODE!r}"
|
||||
assert g.AUTONOMOUS_DEBUG == CONFIG.autonomous.debug_mode, \
|
||||
f"AUTONOMOUS_DEBUG: {g.AUTONOMOUS_DEBUG!r}"
|
||||
assert g.VOICE_DEBUG_MODE == CONFIG.voice.debug_mode, \
|
||||
f"VOICE_DEBUG_MODE: {g.VOICE_DEBUG_MODE!r}"
|
||||
assert g.USE_CHESHIRE_CAT == CONFIG.cheshire_cat.enabled, \
|
||||
f"USE_CHESHIRE_CAT: {g.USE_CHESHIRE_CAT!r}"
|
||||
assert getattr(g, 'PREFER_AMD_GPU', CONFIG.gpu.prefer_amd) == CONFIG.gpu.prefer_amd, \
|
||||
f"PREFER_AMD_GPU: {getattr(g, 'PREFER_AMD_GPU', None)!r}"
|
||||
assert g.DM_MOOD == "neutral", f"DM_MOOD: {g.DM_MOOD!r}"
|
||||
finally:
|
||||
for attr, val in saved.items():
|
||||
if val is None and hasattr(g, attr):
|
||||
delattr(g, attr)
|
||||
elif val is not None:
|
||||
setattr(g, attr, val)
|
||||
|
||||
|
||||
def test_reset_single_key():
|
||||
"""Step 5: single-key reset only affects that one global."""
|
||||
from config_manager import config_manager
|
||||
|
||||
old_lang = g.LANGUAGE_MODE
|
||||
old_debug = g.AUTONOMOUS_DEBUG
|
||||
try:
|
||||
g.LANGUAGE_MODE = "japanese"
|
||||
g.AUTONOMOUS_DEBUG = True
|
||||
|
||||
config_manager.reset_to_defaults("discord.language_mode")
|
||||
|
||||
assert g.LANGUAGE_MODE == CONFIG.discord.language_mode, \
|
||||
f"LANGUAGE_MODE should be default, got {g.LANGUAGE_MODE!r}"
|
||||
# Other globals should NOT have been reset
|
||||
assert g.AUTONOMOUS_DEBUG is True, \
|
||||
"AUTONOMOUS_DEBUG should still be True (not reset)"
|
||||
finally:
|
||||
g.LANGUAGE_MODE = old_lang
|
||||
g.AUTONOMOUS_DEBUG = old_debug
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# D. Server Manager (Steps 7-9)
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
def test_missing_config_gives_zero_servers():
|
||||
"""Step 9: no servers_config.json → empty servers dict."""
|
||||
sm = ServerManager(config_file="/tmp/_nonexistent_miku_test_.json")
|
||||
assert len(sm.servers) == 0, f"Expected 0 servers, got {len(sm.servers)}"
|
||||
|
||||
|
||||
def test_corrupt_config_gives_zero_servers():
|
||||
"""Step 9: corrupt JSON → zero servers (no hardcoded default)."""
|
||||
path = "/tmp/_corrupt_miku_test_.json"
|
||||
try:
|
||||
with open(path, "w") as f:
|
||||
f.write("{{{invalid json!")
|
||||
sm = ServerManager(config_file=path)
|
||||
assert len(sm.servers) == 0, f"Expected 0 servers, got {len(sm.servers)}"
|
||||
finally:
|
||||
os.remove(path)
|
||||
|
||||
|
||||
def test_valid_config_loads():
|
||||
"""Step 9: valid JSON loads servers correctly."""
|
||||
path = "/tmp/_valid_miku_test_.json"
|
||||
try:
|
||||
data = {
|
||||
"12345": {
|
||||
"guild_id": 12345,
|
||||
"guild_name": "Test Guild",
|
||||
"autonomous_channel_id": 67890,
|
||||
"autonomous_channel_name": "test-chat",
|
||||
"bedtime_channel_ids": [67890],
|
||||
"enabled_features": ["autonomous"],
|
||||
}
|
||||
}
|
||||
with open(path, "w") as f:
|
||||
json.dump(data, f)
|
||||
|
||||
sm = ServerManager(config_file=path)
|
||||
assert len(sm.servers) == 1, f"Expected 1 server, got {len(sm.servers)}"
|
||||
assert 12345 in sm.servers
|
||||
assert sm.servers[12345].guild_name == "Test Guild"
|
||||
finally:
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
|
||||
def test_add_remove_server_roundtrip():
|
||||
"""Steps 8-9: add/remove on empty state works cleanly."""
|
||||
sm = ServerManager(config_file="/tmp/_roundtrip_miku_test_.json")
|
||||
try:
|
||||
assert len(sm.servers) == 0
|
||||
|
||||
ok = sm.add_server(99999, "Roundtrip Guild", 11111, "rt-chat")
|
||||
assert ok is True, "add_server should return True"
|
||||
assert 99999 in sm.servers
|
||||
assert sm.servers[99999].guild_name == "Roundtrip Guild"
|
||||
|
||||
ok = sm.remove_server(99999)
|
||||
assert ok is True, "remove_server should return True"
|
||||
assert 99999 not in sm.servers
|
||||
finally:
|
||||
if os.path.exists("/tmp/_roundtrip_miku_test_.json"):
|
||||
os.remove("/tmp/_roundtrip_miku_test_.json")
|
||||
|
||||
|
||||
def test_no_server_memories_attribute():
|
||||
"""Step 8: server_memories dict and methods were removed."""
|
||||
assert not hasattr(ServerManager, 'get_server_memory'), \
|
||||
"get_server_memory method still exists"
|
||||
assert not hasattr(ServerManager, 'set_server_memory'), \
|
||||
"set_server_memory method still exists"
|
||||
|
||||
sm = ServerManager(config_file="/tmp/_nomem_miku_test_.json")
|
||||
assert not hasattr(sm, 'server_memories'), \
|
||||
"server_memories attribute still exists on instance"
|
||||
|
||||
|
||||
def test_no_create_default_config():
|
||||
"""Step 9: _create_default_config was removed."""
|
||||
assert not hasattr(ServerManager, '_create_default_config'), \
|
||||
"_create_default_config method still exists"
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# E. GPU Deduplication (Step 10)
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
def test_gpu_url_helper_delegates():
|
||||
"""Step 10: get_current_gpu_url() uses config_manager, not direct file read."""
|
||||
import inspect
|
||||
import api
|
||||
src = inspect.getsource(api.get_current_gpu_url)
|
||||
assert "gpu_state.json" not in src, \
|
||||
"get_current_gpu_url still reads gpu_state.json directly"
|
||||
assert "config_manager" in src, \
|
||||
"get_current_gpu_url should delegate to config_manager"
|
||||
|
||||
|
||||
def test_gpu_status_endpoint_delegates():
|
||||
"""Step 10: /gpu-status endpoint uses config_manager, not direct file read."""
|
||||
import inspect
|
||||
import api
|
||||
src = inspect.getsource(api.get_gpu_status)
|
||||
assert "gpu_state.json" not in src, \
|
||||
"get_gpu_status still reads gpu_state.json directly"
|
||||
assert "config_manager" in src, \
|
||||
"get_gpu_status should delegate to config_manager"
|
||||
|
||||
|
||||
def test_gpu_url_returns_correct_url():
|
||||
"""Step 10: URL switches correctly between nvidia/amd."""
|
||||
from config_manager import config_manager
|
||||
import api
|
||||
|
||||
old_gpu = config_manager.get_gpu()
|
||||
try:
|
||||
config_manager.set_gpu("nvidia")
|
||||
assert api.get_current_gpu_url() == g.LLAMA_URL
|
||||
|
||||
config_manager.set_gpu("amd")
|
||||
assert api.get_current_gpu_url() == g.LLAMA_AMD_URL
|
||||
finally:
|
||||
config_manager.set_gpu(old_gpu)
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# F. Clean Imports (Steps 3, 10)
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
def test_config_py_no_os_import():
|
||||
"""Step 10: config.py does not import os (unused)."""
|
||||
src = Path("/app/config.py").read_text()
|
||||
# Check for standalone 'import os' line (not in comments)
|
||||
for line in src.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped == "import os" or stripped.startswith("import os "):
|
||||
assert False, f"config.py still has: {stripped}"
|
||||
|
||||
|
||||
def test_config_manager_no_dead_imports():
|
||||
"""Step 10: config_manager.py has no unused 'os' or 'Union' imports."""
|
||||
src = Path("/app/config_manager.py").read_text()
|
||||
for line in src.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped == "import os" or stripped.startswith("import os "):
|
||||
assert False, f"config_manager.py still has: {stripped}"
|
||||
assert "Union" not in src, "config_manager.py still imports Union"
|
||||
|
||||
|
||||
def test_globals_no_guild_settings():
|
||||
"""Step 10: globals.py has no GUILD_SETTINGS."""
|
||||
assert not hasattr(g, 'GUILD_SETTINGS'), \
|
||||
"GUILD_SETTINGS still exists in globals"
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════
|
||||
# Main
|
||||
# ═══════════════════════════════════════════════════
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("\n══════════════════════════════════════════")
|
||||
print(" Config / State Regression Tests")
|
||||
print("══════════════════════════════════════════\n")
|
||||
|
||||
tests = [
|
||||
# A. Config Loading & Persistence
|
||||
test_runtime_config_persists_in_memory_dir,
|
||||
test_config_yaml_loads_into_pydantic,
|
||||
test_runtime_overrides_merge,
|
||||
# B. Runtime State
|
||||
test_runtime_state_reads_live_globals,
|
||||
test_config_set_syncs_all_simple_globals,
|
||||
test_config_set_syncs_dm_mood,
|
||||
test_restore_runtime_settings,
|
||||
# C. Reset
|
||||
test_reset_to_defaults_resets_all_globals,
|
||||
test_reset_single_key,
|
||||
# D. Server Manager
|
||||
test_missing_config_gives_zero_servers,
|
||||
test_corrupt_config_gives_zero_servers,
|
||||
test_valid_config_loads,
|
||||
test_add_remove_server_roundtrip,
|
||||
test_no_server_memories_attribute,
|
||||
test_no_create_default_config,
|
||||
# E. GPU Deduplication
|
||||
test_gpu_url_helper_delegates,
|
||||
test_gpu_status_endpoint_delegates,
|
||||
test_gpu_url_returns_correct_url,
|
||||
# F. Clean Imports
|
||||
test_config_py_no_os_import,
|
||||
test_config_manager_no_dead_imports,
|
||||
test_globals_no_guild_settings,
|
||||
]
|
||||
|
||||
for t in tests:
|
||||
run_test(t)
|
||||
|
||||
# Summary
|
||||
passed = sum(1 for _, ok, _ in _results if ok)
|
||||
failed = sum(1 for _, ok, _ in _results if not ok)
|
||||
total = len(_results)
|
||||
|
||||
print(f"\n──────────────────────────────────────────")
|
||||
print(f" {passed}/{total} passed, {failed} failed")
|
||||
print(f"──────────────────────────────────────────\n")
|
||||
|
||||
if failed:
|
||||
print("FAILED tests:")
|
||||
for name, ok, detail in _results:
|
||||
if not ok:
|
||||
print(f" ✗ {name}: {detail}")
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("ALL PASSED ✓")
|
||||
sys.exit(0)
|
||||
Reference in New Issue
Block a user