koko210Serve
366bee2e43
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)
512 lines
20 KiB
Python
512 lines
20 KiB
Python
"""
|
|
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)
|