Sanad/tests/test_smoke.py

532 lines
21 KiB
Python

"""Smoke tests — verify imports, paths, basic instantiation, and isolation.
Run with:
PYTHONPATH=/path/to/yslootahtech python3 -m unittest \
Project.Sanad.tests.test_smoke -v
These tests do NOT require any third-party dependency. They prove that
the project loads cleanly and that subsystem failures stay isolated.
"""
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
class TestConfig(unittest.TestCase):
"""Path resolution and config loading."""
def test_base_dir_auto_detect(self):
from Project.Sanad.config import BASE_DIR
self.assertTrue(BASE_DIR.exists(), f"BASE_DIR missing: {BASE_DIR}")
self.assertEqual(BASE_DIR.name, "Sanad")
def test_data_dirs_exist(self):
from Project.Sanad.config import (
DATA_DIR, MOTIONS_DIR, AUDIO_RECORDINGS_DIR,
MOTION_RECORDINGS_DIR, LOGS_DIR, SCRIPTS_DIR,
)
for d in (DATA_DIR, MOTIONS_DIR, AUDIO_RECORDINGS_DIR,
MOTION_RECORDINGS_DIR, LOGS_DIR, SCRIPTS_DIR):
self.assertTrue(d.exists(), f"Missing: {d}")
def test_skills_file_resolves(self):
from Project.Sanad.config import SKILLS_FILE, MOTIONS_DIR
self.assertEqual(SKILLS_FILE.parent, MOTIONS_DIR)
def test_dds_interface_default(self):
from Project.Sanad.config import DDS_NETWORK_INTERFACE
self.assertIsInstance(DDS_NETWORK_INTERFACE, str)
def test_env_override_dds_interface(self):
os.environ["SANAD_DDS_INTERFACE"] = "test_iface"
try:
import importlib
import Project.Sanad.config as cfg
importlib.reload(cfg)
self.assertEqual(cfg.DDS_NETWORK_INTERFACE, "test_iface")
finally:
del os.environ["SANAD_DDS_INTERFACE"]
import importlib
import Project.Sanad.config as cfg
importlib.reload(cfg)
def test_load_config_handles_missing(self):
from Project.Sanad.config import load_config
result = load_config()
self.assertIsInstance(result, dict)
def test_local_tts_paths(self):
from Project.Sanad.config import (
LOCAL_TTS_MODEL_PATH, LOCAL_TTS_HIFIGAN_PATH, LOCAL_TTS_XVECTOR_PATH,
)
self.assertIn("speecht5_tts_clartts_ar", LOCAL_TTS_MODEL_PATH)
self.assertIn("speecht5_hifigan", LOCAL_TTS_HIFIGAN_PATH)
class TestSkillRegistry(unittest.TestCase):
"""SkillRegistry CRUD + atomic writes + validation."""
def setUp(self):
from Project.Sanad.core.skill_registry import SkillRegistry
self.sr = SkillRegistry()
def test_load(self):
skills = self.sr.list_skills()
self.assertIsInstance(skills, list)
def test_invalid_sync_mode_rejected(self):
from Project.Sanad.core.skill_registry import Skill
bad = Skill(id="test_invalid", sync_mode="garbage")
with self.assertRaises(ValueError):
self.sr.add(bad)
def test_update_missing_returns_none(self):
result = self.sr.update("nonexistent_id_12345", {"description": "x"})
self.assertIsNone(result)
def test_delete_missing_returns_none(self):
result = self.sr.delete("nonexistent_id_12345")
self.assertIsNone(result)
class TestEventBus(unittest.TestCase):
"""EventBus emit_sync handles missing event loop and async handlers."""
def test_emit_sync_no_handlers(self):
from Project.Sanad.core.event_bus import EventBus
bus = EventBus()
# Should not raise
bus.emit_sync("nonexistent.event", value=1)
def test_emit_sync_sync_handler(self):
from Project.Sanad.core.event_bus import EventBus
bus = EventBus()
captured = []
bus.on("test.event", lambda **kw: captured.append(kw))
bus.emit_sync("test.event", value=42)
self.assertEqual(captured, [{"value": 42}])
def test_emit_sync_async_handler_no_loop(self):
from Project.Sanad.core.event_bus import EventBus
bus = EventBus()
async def handler(**kw):
pass
bus.on("test.async", handler)
# No running loop — should warn but not crash
bus.emit_sync("test.async", value=1)
def test_handler_exception_isolated(self):
from Project.Sanad.core.event_bus import EventBus
bus = EventBus()
results = []
def good(**kw):
results.append("ok")
def bad(**kw):
raise RuntimeError("intentional")
bus.on("test.iso", bad)
bus.on("test.iso", good)
bus.emit_sync("test.iso", x=1)
# Good handler still ran
self.assertEqual(results, ["ok"])
class TestBrainCallbackWhitelist(unittest.TestCase):
"""Brain._resolve_callback rejects non-whitelisted modules (RCE block)."""
def setUp(self):
from Project.Sanad.core.brain import Brain
self.brain = Brain()
def test_rce_blocked_os(self):
cb = self.brain._resolve_callback("os:system")
self.assertIsNone(cb, "os:system must be rejected")
def test_rce_blocked_subprocess(self):
cb = self.brain._resolve_callback("subprocess:run")
self.assertIsNone(cb, "subprocess:run must be rejected")
def test_rce_blocked_eval(self):
cb = self.brain._resolve_callback("builtins:eval")
self.assertIsNone(cb)
def test_empty_callback_returns_none(self):
self.assertIsNone(self.brain._resolve_callback(""))
self.assertIsNone(self.brain._resolve_callback(None))
def test_invalid_format_returns_none(self):
self.assertIsNone(self.brain._resolve_callback("no_colon"))
def test_whitelisted_prefix_attempted(self):
# Module doesn't exist but the prefix is allowed — must NOT be rejected
# by the whitelist (it'll fail at import_module instead)
cb = self.brain._resolve_callback("Project.Sanad.motion.nonexistent_module:fn")
self.assertIsNone(cb) # fails at import, not at whitelist
class TestWakePhraseMatching(unittest.TestCase):
"""Token-based phrase matching — no false positives on substrings."""
def test_no_false_positive_substring(self):
from Project.Sanad.voice.text_utils import match_phrase
# 'this' contains 'hi' as substring — must NOT match
result = match_phrase("this is a test", {"hi_action": {"hi"}})
self.assertIsNone(result)
def test_exact_word_match(self):
from Project.Sanad.voice.text_utils import match_phrase
result = match_phrase("hi there friend", {"greet": {"hi"}})
self.assertEqual(result, "greet")
def test_multi_word_phrase_all_required(self):
from Project.Sanad.voice.text_utils import match_phrase
# All words must appear
result = match_phrase("please shake hands", {"act": {"shake hands"}})
self.assertEqual(result, "act")
result = match_phrase("just shake", {"act": {"shake hands"}})
self.assertIsNone(result)
def test_longest_phrase_wins(self):
from Project.Sanad.voice.text_utils import match_phrase
sets = {
"short": {"hi"},
"long": {"hi five"},
}
# When both match, prefer the more-specific (longer) phrase
result = match_phrase("hi five there", sets)
self.assertEqual(result, "long")
def test_normalize_arabic(self):
from Project.Sanad.voice.text_utils import normalize_arabic
out = normalize_arabic("مَرْحَبًا")
self.assertNotIn("\u064b", out) # tashkeel removed
class TestArmController(unittest.TestCase):
"""Arm controller in simulation mode."""
def test_constructs(self):
from Project.Sanad.motion.arm_controller import ArmController
arm = ArmController()
self.assertFalse(arm.is_busy)
def test_action_registry(self):
from Project.Sanad.motion.arm_controller import ACTIONS, ACTION_BY_NAME
self.assertGreater(len(ACTIONS), 0)
# SDK actions should be in the registry
sdk_names = ["shake_hand", "high_five", "clap"]
for name in sdk_names:
self.assertIn(name, ACTION_BY_NAME)
def test_motion_files_cache(self):
from Project.Sanad.motion.arm_controller import ArmController
arm = ArmController()
first = arm.list_motion_files()
second = arm.list_motion_files()
# Same call twice — should hit cache
self.assertEqual(len(first), len(second))
def test_unknown_action_raises(self):
from Project.Sanad.motion.arm_controller import ArmController
arm = ArmController()
with self.assertRaises(KeyError):
arm.trigger_by_name("definitely_not_a_real_action_xyz")
def test_arm_adapter_works(self):
"""The MacroPlayer arm adapter should work over a real ArmController."""
from Project.Sanad.motion.arm_controller import ArmController
from Project.Sanad.motion.macro_player import _ArmAdapter
arm = ArmController()
adapter = _ArmAdapter(arm)
# All adapter methods should be callable without raising
q = adapter.get_current_q()
self.assertEqual(len(q), 29)
self.assertIsInstance(adapter.state_age(), float)
# wait_for_state with no real DDS subscriber should just return True
# (since the adapter falls back when wait_for_state is missing) OR
# block until timeout (if the method exists and waits). Either way
# it should NOT raise.
result = adapter.wait_for_state(timeout=0.05)
self.assertIsInstance(result, bool)
class TestSafeIO(unittest.TestCase):
"""Path traversal protection + atomic writes."""
def test_safe_filename_strips_traversal(self):
from Project.Sanad.dashboard.routes._safe_io import safe_filename
with self.assertRaises(Exception):
safe_filename("..")
with self.assertRaises(Exception):
safe_filename("")
with self.assertRaises(Exception):
safe_filename(None)
# Embedded path components should be stripped to basename
self.assertEqual(safe_filename("../../etc/passwd"), "passwd")
self.assertEqual(safe_filename("foo.wav"), "foo.wav")
def test_safe_path_under_blocks_escape(self):
from Project.Sanad.dashboard.routes._safe_io import safe_path_under
from Project.Sanad.config import MOTIONS_DIR
# Normal name → ok
p = safe_path_under(MOTIONS_DIR, "foo.jsonl")
self.assertTrue(str(p).startswith(str(MOTIONS_DIR.resolve())))
# Traversal attempt → rejected (basename strip means it's just "passwd"
# under MOTIONS_DIR, which is safe)
p2 = safe_path_under(MOTIONS_DIR, "../../etc/passwd")
self.assertTrue(str(p2).startswith(str(MOTIONS_DIR.resolve())))
def test_atomic_write_text(self):
import tempfile
from Project.Sanad.dashboard.routes._safe_io import atomic_write_text
with tempfile.TemporaryDirectory() as td:
target = Path(td) / "test.txt"
atomic_write_text(target, "hello\nworld")
self.assertEqual(target.read_text(), "hello\nworld")
def test_atomic_write_json(self):
import tempfile
import json
from Project.Sanad.dashboard.routes._safe_io import atomic_write_json
with tempfile.TemporaryDirectory() as td:
target = Path(td) / "test.json"
atomic_write_json(target, {"a": 1, "b": [1, 2, 3]})
self.assertEqual(json.loads(target.read_text()), {"a": 1, "b": [1, 2, 3]})
class TestGeminiClientStructure(unittest.TestCase):
"""GeminiVoiceClient structural tests — no actual websocket."""
def setUp(self):
try:
from Project.Sanad.voice.gemini_client import GeminiVoiceClient
self.client = GeminiVoiceClient()
except ImportError:
self.skipTest("websockets not installed")
def test_initial_state(self):
self.assertFalse(self.client.connected)
self.assertIsNone(self.client.session_owner)
def test_send_audio_chunk_when_disconnected(self):
"""Should return False, not raise — the no-op-forever bug is fixed."""
import asyncio
result = asyncio.run(self.client.send_audio_chunk("dGVzdA=="))
self.assertFalse(result)
def test_acquire_session_returns_guard(self):
"""acquire_session should be sync and return a context manager."""
guard = self.client.acquire_session("test")
# Has __aenter__/__aexit__
self.assertTrue(hasattr(guard, "__aenter__"))
self.assertTrue(hasattr(guard, "__aexit__"))
def test_session_lock_exclusive(self):
"""Two consecutive acquires should serialize, not deadlock."""
import asyncio
events = []
async def consumer(name):
async with self.client.acquire_session(name):
events.append(f"{name}:enter")
await asyncio.sleep(0.05)
events.append(f"{name}:exit")
async def runner():
await asyncio.gather(consumer("A"), consumer("B"))
asyncio.run(runner())
# Either A fully runs then B, or B fully runs then A — never interleaved
self.assertIn("A:enter", events)
self.assertIn("B:enter", events)
# The exit of one comes before the enter of the other
a_exit = events.index("A:exit")
b_enter = events.index("B:enter")
a_enter = events.index("A:enter")
b_exit = events.index("B:exit")
ok1 = (a_enter < a_exit < b_enter < b_exit)
ok2 = (b_enter < b_exit < a_enter < a_exit)
self.assertTrue(ok1 or ok2, f"Lock not exclusive: {events}")
class TestAudioDevices(unittest.TestCase):
"""audio_devices module — pure helpers tested without pactl."""
def test_profiles_defined(self):
from Project.Sanad.voice import audio_devices as ad
self.assertGreater(len(ad.PROFILES), 0)
ids = {p.id for p in ad.PROFILES}
self.assertIn("anker_powerconf", ids)
self.assertIn("hollyland_builtin", ids)
self.assertIn("builtin", ids)
def test_pactl_available_no_crash(self):
from Project.Sanad.voice import audio_devices as ad
# Should not raise, just return False if pactl missing
result = ad.pactl_available()
self.assertIsInstance(result, bool)
def test_find_first_match(self):
from Project.Sanad.voice import audio_devices as ad
items = [
{"name": "alsa_output.usb-Anker_PowerConf_A3321-DEV-SN1-01.analog-stereo",
"description": "", "index": "0"},
{"name": "alsa_output.platform-sound.analog-stereo",
"description": "", "index": "1"},
{"name": "alsa_input.usb-Shenzhen_Hollyland_Technology_Co._Ltd_Wireless_microphone_C63X223T6MX-01.analog-stereo",
"description": "", "index": "5"},
]
# Anker pattern matches (multi-pattern: "powerconf,anker")
m = ad.find_first_match(items, "powerconf,anker")
self.assertIsNotNone(m)
self.assertIn("Anker_PowerConf", m["name"])
# Built-in pattern matches
m2 = ad.find_first_match(items, "platform-sound")
self.assertIsNotNone(m2)
# Hollyland — matches by "hollyland" OR "wireless_microphone"
m3 = ad.find_first_match(items, "hollyland,wireless_microphone")
self.assertIsNotNone(m3)
self.assertIn("Hollyland", m3["name"])
# Case-insensitive
m4 = ad.find_first_match(items, "HOLLYLAND")
self.assertIsNotNone(m4)
# No match returns None
m5 = ad.find_first_match(items, "nonexistent")
self.assertIsNone(m5)
# Empty pattern returns None
m6 = ad.find_first_match(items, "")
self.assertIsNone(m6)
# Different USB port — Anker on SN2-03 instead of SN1-01
items_port2 = [
{"name": "alsa_output.usb-Anker_PowerConf_A3321-DEV-SN2-03.analog-stereo",
"description": "", "index": "0"},
]
m7 = ad.find_first_match(items_port2, "powerconf,anker")
self.assertIsNotNone(m7, "Must match Anker regardless of USB port suffix")
def test_profile_detection_with_jetson_devices(self):
"""Simulate the exact PulseAudio names from the G1 Jetson and verify
all three profiles match correctly."""
from Project.Sanad.voice import audio_devices as ad
fake_sinks = [
{"name": "alsa_output.platform-sound.analog-stereo", "description": "", "index": "0"},
{"name": "alsa_output.usb-Anker_PowerConf_A3321-DEV-SN1-01.analog-stereo", "description": "", "index": "1"},
]
fake_sources = [
{"name": "alsa_output.platform-sound.analog-stereo.monitor", "description": "", "index": "0"},
{"name": "alsa_input.platform-sound.analog-stereo", "description": "", "index": "1"},
{"name": "alsa_output.usb-Anker_PowerConf_A3321-DEV-SN1-01.analog-stereo.monitor", "description": "", "index": "3"},
{"name": "alsa_input.usb-Anker_PowerConf_A3321-DEV-SN1-01.mono-fallback", "description": "", "index": "4"},
{"name": "alsa_input.usb-Shenzhen_Hollyland_Technology_Co._Ltd_Wireless_microphone_C63X223T6MX-01.analog-stereo", "description": "", "index": "5"},
]
# Patch list_sinks/list_sources
orig_sinks = ad.list_sinks
orig_sources = ad.list_sources
ad.list_sinks = lambda: fake_sinks
ad.list_sources = lambda: fake_sources
try:
detected = ad.detect_plugged_profiles()
detected_ids = [d["profile"]["id"] for d in detected]
self.assertIn("hollyland_builtin", detected_ids,
"Hollyland + built-in must be detected")
self.assertIn("anker_powerconf", detected_ids,
"Anker PowerConf must be detected")
self.assertIn("builtin", detected_ids,
"Built-in must be detected")
# Verify Hollyland gets the correct sink and source
holly = next(d for d in detected if d["profile"]["id"] == "hollyland_builtin")
self.assertIn("platform-sound", holly["sink"]["name"])
self.assertIn("Hollyland", holly["source"]["name"])
# Verify Anker gets Anker sink AND Anker source (not built-in)
anker = next(d for d in detected if d["profile"]["id"] == "anker_powerconf")
self.assertIn("PowerConf", anker["sink"]["name"])
self.assertIn("PowerConf", anker["source"]["name"])
finally:
ad.list_sinks = orig_sinks
ad.list_sources = orig_sources
def test_profile_detection_different_usb_port(self):
"""Verify that Anker is detected even when plugged into a different port
(serial suffix changes from SN1-01 to SN2-03)."""
from Project.Sanad.voice import audio_devices as ad
fake_sinks = [
{"name": "alsa_output.usb-Anker_PowerConf_A3321-DEV-SN2-03.analog-stereo", "description": "", "index": "1"},
]
fake_sources = [
{"name": "alsa_input.usb-Anker_PowerConf_A3321-DEV-SN2-03.mono-fallback", "description": "", "index": "4"},
]
orig_sinks = ad.list_sinks
orig_sources = ad.list_sources
ad.list_sinks = lambda: fake_sinks
ad.list_sources = lambda: fake_sources
try:
detected = ad.detect_plugged_profiles()
detected_ids = [d["profile"]["id"] for d in detected]
self.assertIn("anker_powerconf", detected_ids,
"Anker must be detected regardless of USB port")
finally:
ad.list_sinks = orig_sinks
ad.list_sources = orig_sources
def test_status_no_crash_without_pactl(self):
from Project.Sanad.voice import audio_devices as ad
# Should return a dict even without pactl
s = ad.status()
self.assertIsInstance(s, dict)
self.assertIn("pactl_available", s)
self.assertIn("current", s)
self.assertIn("profiles", s)
# current always has these keys
cur = s["current"]
self.assertIn("sink", cur)
self.assertIn("source", cur)
self.assertIn("source_kind", cur)
def test_load_save_state_atomic(self):
from Project.Sanad.voice import audio_devices as ad
# Round-trip
original = ad.load_state()
try:
ad.save_state({"profile_id": "_test_unit", "sink": "x", "source": "y"})
self.assertEqual(ad.load_state()["profile_id"], "_test_unit")
finally:
ad.save_state(original)
def test_select_unknown_profile_rejected(self):
from Project.Sanad.voice import audio_devices as ad
result = ad.select_profile("definitely_not_a_real_profile")
self.assertFalse(result["ok"])
class TestIsolation(unittest.TestCase):
"""Failure isolation: one missing dep doesn't take down others."""
def test_main_module_imports_with_missing_deps(self):
"""main.py must import even when third-party deps are missing."""
import importlib
if "Project.Sanad.main" in sys.modules:
del sys.modules["Project.Sanad.main"]
m = importlib.import_module("Project.Sanad.main")
# Critical subsystems must be present even with missing deps
self.assertIsNotNone(m.brain, "brain must always be available")
def test_subsystem_status_reported(self):
import Project.Sanad.main as m
self.assertTrue(hasattr(m, "SUBSYSTEMS"))
self.assertIn("brain", m.SUBSYSTEMS)
self.assertIn("arm", m.SUBSYSTEMS)
if __name__ == "__main__":
unittest.main(verbosity=2)