397 lines
16 KiB
Python
397 lines
16 KiB
Python
"""Regression tests for the fixes listed in docs/AUDIT.md."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from javis.core.config import Config, PathsConfig, load_config
|
|
|
|
|
|
class SchemaBridgeTests(unittest.TestCase):
|
|
"""MCP-01 / MCP-03."""
|
|
|
|
def test_input_schema_read_from_snake_and_camel_case(self) -> None:
|
|
from javis.mcp_client.schema_bridge import mcp_tool_to_ollama
|
|
|
|
schema = {"type": "object", "properties": {"q": {"type": "string"}}}
|
|
v2 = SimpleNamespace(name="search", description="d", input_schema=schema)
|
|
v1 = SimpleNamespace(name="search", description="d", inputSchema=schema)
|
|
for tool in (v2, v1):
|
|
entry = mcp_tool_to_ollama("ddg", tool)
|
|
self.assertEqual(entry["function"]["parameters"], schema)
|
|
self.assertEqual(entry["function"]["name"], "ddg__search")
|
|
|
|
def test_real_mcp_tool_schema_is_forwarded(self) -> None:
|
|
import mcp.types as t
|
|
from javis.mcp_client.schema_bridge import mcp_tool_to_ollama
|
|
|
|
schema = {"type": "object", "properties": {"a": {"type": "string"}}}
|
|
fields = t.Tool.model_fields
|
|
key = "input_schema" if "input_schema" in fields else "inputSchema"
|
|
tool = t.Tool(name="x", **{key: schema})
|
|
self.assertEqual(mcp_tool_to_ollama("s", tool)["function"]["parameters"], schema)
|
|
|
|
def test_bare_server_name_is_not_mapped_to_a_tool(self) -> None:
|
|
from javis.mcp_client.schema_bridge import validate_tool_name
|
|
|
|
self.assertEqual(validate_tool_name("homey", {"homey": ["delete_all"]}), ("", "homey"))
|
|
self.assertEqual(validate_tool_name("homey__find_devices"), ("homey", "find_devices"))
|
|
|
|
|
|
class ResultToTextTests(unittest.TestCase):
|
|
def test_is_error_detected_for_both_sdk_generations(self) -> None:
|
|
from javis.mcp_client.host import _result_to_text
|
|
|
|
content = [SimpleNamespace(text="boom")]
|
|
self.assertEqual(_result_to_text(SimpleNamespace(content=content, is_error=True)), "ERROR: boom")
|
|
self.assertEqual(_result_to_text(SimpleNamespace(content=content, isError=True)), "ERROR: boom")
|
|
self.assertEqual(_result_to_text(SimpleNamespace(content=content, is_error=False)), "boom")
|
|
|
|
|
|
def _host_with_fake_server(approval: str = "ask"):
|
|
from javis.mcp_client.host import ConnectedServer, MCPHost, ServerSpec
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
class FakeSession:
|
|
async def call_tool(self, tool, arguments):
|
|
calls.append((tool, arguments))
|
|
return SimpleNamespace(content=[SimpleNamespace(text="ok")], is_error=False)
|
|
|
|
host = MCPHost(Config())
|
|
spec = ServerSpec(name="fs", command="x", approval=approval)
|
|
host.servers["fs"] = ConnectedServer(spec=spec, session=FakeSession(),
|
|
tools=[SimpleNamespace(name="write")])
|
|
return host, calls
|
|
|
|
|
|
class ApprovalTests(unittest.IsolatedAsyncioTestCase):
|
|
"""MCP-02: "Allow once" must not become "Always allow"."""
|
|
|
|
async def test_allow_once_lasts_only_for_the_turn(self) -> None:
|
|
host, calls = _host_with_fake_server()
|
|
asked: list[str] = []
|
|
|
|
async def approve(server, tool, args):
|
|
asked.append(tool)
|
|
return True
|
|
|
|
host.approval_cb = approve
|
|
self.assertEqual(await host.call("fs__write", {"p": 1}), "ok")
|
|
self.assertEqual(await host.call("fs__write", {"p": 2}), "ok")
|
|
self.assertEqual(len(asked), 1) # same turn: not asked again
|
|
host.clear_denials() # next turn
|
|
await host.call("fs__write", {"p": 3})
|
|
self.assertEqual(len(asked), 2)
|
|
self.assertEqual(len(calls), 3)
|
|
|
|
async def test_always_allow_persists(self) -> None:
|
|
host, _ = _host_with_fake_server()
|
|
asked: list[str] = []
|
|
|
|
async def approve(server, tool, args):
|
|
asked.append(tool)
|
|
return True
|
|
|
|
host.approval_cb = approve
|
|
host.allow_tool("fs", "write")
|
|
host.clear_denials()
|
|
await host.call("fs__write", {})
|
|
self.assertEqual(asked, [])
|
|
|
|
async def test_bare_server_name_does_not_execute_anything(self) -> None:
|
|
host, calls = _host_with_fake_server(approval="allow")
|
|
result = await host.call("fs", {"p": 1})
|
|
self.assertTrue(result.startswith("ERROR:"))
|
|
self.assertIn("fs__write", result)
|
|
self.assertEqual(calls, [])
|
|
|
|
async def test_protocol_error_is_not_retried(self) -> None:
|
|
from javis.mcp_client import host as host_mod
|
|
|
|
host, _ = _host_with_fake_server(approval="allow")
|
|
attempts = []
|
|
|
|
class FailingSession:
|
|
async def call_tool(self, tool, arguments):
|
|
attempts.append(tool)
|
|
raise host_mod._MCPError(-32602, "invalid params")
|
|
|
|
host.servers["fs"].session = FailingSession()
|
|
|
|
async def no_reconnect(*a, **kw): # pragma: no cover - must not run
|
|
raise AssertionError("protocol errors must not trigger a reconnect")
|
|
|
|
with patch.object(host_mod.MCPHost, "_reconnect", no_reconnect):
|
|
result = await host.call("fs__write", {})
|
|
self.assertTrue(result.startswith("ERROR:"))
|
|
self.assertEqual(attempts, ["write"])
|
|
|
|
async def test_destructive_tool_asks_even_on_allow_server(self) -> None:
|
|
host, calls = _host_with_fake_server(approval="allow")
|
|
host.servers["fs"].tools = [
|
|
SimpleNamespace(name="write", annotations=SimpleNamespace(destructive_hint=True)),
|
|
SimpleNamespace(name="read", annotations=SimpleNamespace(
|
|
read_only_hint=True, destructive_hint=True)),
|
|
]
|
|
asked: list[str] = []
|
|
|
|
async def deny(server, tool, args):
|
|
asked.append(tool)
|
|
return False
|
|
|
|
host.approval_cb = deny
|
|
self.assertTrue((await host.call("fs__write", {})).startswith("ERROR:"))
|
|
self.assertEqual(await host.call("fs__read", {}), "ok") # read-only wins
|
|
self.assertEqual(asked, ["write"])
|
|
self.assertEqual(calls, [("read", {})])
|
|
|
|
host.config.mcp.confirm_destructive = False
|
|
host.clear_denials()
|
|
self.assertEqual(await host.call("fs__write", {}), "ok")
|
|
self.assertEqual(asked, ["write"])
|
|
|
|
|
|
class ConversationTrimTests(unittest.TestCase):
|
|
"""LLM-02: the in-progress turn is never trimmed away."""
|
|
|
|
def test_current_turn_survives_oversized_tool_results(self) -> None:
|
|
from javis.llm.conversation import Conversation
|
|
|
|
conv = Conversation(system_prompt="s" * 100, max_turns=20, max_chars=500)
|
|
conv.add_user("old question")
|
|
conv.add_assistant("old answer")
|
|
conv.add_user("current question")
|
|
conv.add_assistant("", tool_calls=[{"function": {"name": "a__b", "arguments": {}}}])
|
|
conv.add_tool_result("a__b", "x" * 2000)
|
|
roles = [m["role"] for m in conv.history]
|
|
self.assertEqual(roles, ["user", "assistant", "tool"])
|
|
self.assertEqual(conv.history[0]["content"], "current question")
|
|
|
|
|
|
class QuickEngineValidationTests(unittest.TestCase):
|
|
def test_invalid_regex_rejected_and_never_breaks_matching(self) -> None:
|
|
from javis.agent.quick_engine import QuickAction, QuickCommand, QuickEngine
|
|
|
|
with tempfile.TemporaryDirectory() as d:
|
|
q = QuickEngine(Path(d) / "q.json")
|
|
with self.assertRaises(ValueError):
|
|
q.add("([unclosed", "regex", [{"tool": "a__b", "args": {}}])
|
|
with self.assertRaises(ValueError):
|
|
q.add("x", "fuzzy", [{"tool": "a__b", "args": {}}])
|
|
# A bad pattern already on disk must be skipped, not raise.
|
|
q._commands["bad"] = QuickCommand("bad", "([", "regex", [QuickAction("a__b")])
|
|
q.add("licht an", "contains", [{"tool": "homey__on", "args": {}}])
|
|
self.assertIsNotNone(q.match("licht an"))
|
|
|
|
|
|
class QuickPrivilegeTests(unittest.IsolatedAsyncioTestCase):
|
|
"""AGT-01: quick commands honour speaker privileges."""
|
|
|
|
async def test_denied_action_skips_quick_path(self) -> None:
|
|
from javis.agent.orchestrator import Orchestrator
|
|
from javis.agent.quick_engine import QuickEngine
|
|
from javis.stt.voice_profiler import SpeakerPrivileges
|
|
|
|
with tempfile.TemporaryDirectory() as d:
|
|
orch = object.__new__(Orchestrator)
|
|
orch.quick = QuickEngine(Path(d) / "q.json")
|
|
orch.quick.add("licht an", "contains", [{"tool": "homey__on", "args": {}}])
|
|
executed = []
|
|
|
|
class FakeMCP:
|
|
async def call(self, name, args):
|
|
executed.append(name)
|
|
return "ok"
|
|
|
|
orch.mcp = FakeMCP()
|
|
guest = SpeakerPrivileges(denied_tools=["homey__*"])
|
|
ran = await orch._try_quick_response("licht an", privileges=guest)
|
|
self.assertFalse(ran)
|
|
self.assertEqual(executed, [])
|
|
|
|
|
|
class StopIdempotentTests(unittest.IsolatedAsyncioTestCase):
|
|
"""AGT-03: concurrent stop() calls tear down exactly once."""
|
|
|
|
async def test_concurrent_stop_runs_teardown_once(self) -> None:
|
|
from javis.agent.orchestrator import Orchestrator
|
|
|
|
orch = object.__new__(Orchestrator)
|
|
orch._stop = asyncio.Event()
|
|
orch._stop_lock = asyncio.Lock()
|
|
orch._stopped = False
|
|
orch.mic = None
|
|
orch._barge_in_task = None
|
|
counts = {"mcp": 0, "llm": 0}
|
|
|
|
class Timers:
|
|
async def aclose(self):
|
|
await asyncio.sleep(0.01)
|
|
|
|
class MCP:
|
|
async def stop(self):
|
|
counts["mcp"] += 1
|
|
await asyncio.sleep(0.01)
|
|
|
|
class LLM:
|
|
async def aclose(self):
|
|
counts["llm"] += 1
|
|
|
|
orch.timer_service = Timers()
|
|
orch.mcp = MCP()
|
|
orch.llm = orch.memory_llm = LLM()
|
|
await asyncio.gather(orch.stop(), orch.stop(), orch.stop())
|
|
self.assertEqual(counts, {"mcp": 1, "llm": 1})
|
|
|
|
|
|
class PcmWriterTests(unittest.IsolatedAsyncioTestCase):
|
|
"""AGT-06: cancelling playback stops the writer thread before returning."""
|
|
|
|
async def test_cancel_waits_for_writer_thread(self) -> None:
|
|
import threading
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
from javis.audio_io.stream import write_pcm_blocks
|
|
|
|
writes = []
|
|
closed = threading.Event()
|
|
|
|
class FakeStream:
|
|
def write(self, block):
|
|
if closed.is_set():
|
|
raise AssertionError("write after close")
|
|
time.sleep(0.01)
|
|
writes.append(len(block))
|
|
|
|
task = asyncio.ensure_future(
|
|
write_pcm_blocks(FakeStream(), np.zeros((16000, 1), np.float32), 160))
|
|
await asyncio.sleep(0.05)
|
|
task.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await task
|
|
closed.set() # caller would close the stream now
|
|
n = len(writes)
|
|
await asyncio.sleep(0.05)
|
|
self.assertEqual(len(writes), n)
|
|
self.assertLess(n, 100)
|
|
|
|
|
|
class ConsoleLaunchTests(unittest.TestCase):
|
|
"""GUI-04: every argument is quoted for cmd.exe."""
|
|
|
|
def test_metacharacters_are_quoted(self) -> None:
|
|
from javis.gui.main_window import _launch_in_console
|
|
|
|
with patch("subprocess.Popen") as popen:
|
|
_launch_in_console(["C:\\Py thon\\python.exe", "--voice", "a&calc"], cwd=".")
|
|
line = popen.call_args.args[0]
|
|
self.assertEqual(line, 'cmd.exe /k ""C:\\Py thon\\python.exe" "--voice" "a&calc""')
|
|
with self.assertRaises(ValueError):
|
|
_launch_in_console(['x"y'], cwd=".")
|
|
|
|
|
|
class VoiceImportTests(unittest.TestCase):
|
|
"""VOI-01: pickle checkpoints are not imported by default."""
|
|
|
|
def test_checkpoints_skipped(self) -> None:
|
|
from javis.voices.io import import_profile
|
|
|
|
with tempfile.TemporaryDirectory() as d:
|
|
arc = Path(d) / "v.javisvoice"
|
|
with zipfile.ZipFile(arc, "w") as zf:
|
|
zf.writestr("MANIFEST.json", json.dumps({"kind": "javis.voice_profile", "name": "v"}))
|
|
zf.writestr("profile.json", "{}")
|
|
zf.writestr("refs/a.wav", b"RIFF")
|
|
zf.writestr("training/piper_lightning/last.ckpt", b"\x80\x04evil")
|
|
zf.writestr("training/base_checkpoint.txt", "C:/x.ckpt")
|
|
res = import_profile(arc, Path(d) / "voices")
|
|
self.assertTrue((res.path / "refs" / "a.wav").exists())
|
|
self.assertFalse((res.path / "training" / "piper_lightning" / "last.ckpt").exists())
|
|
self.assertFalse((res.path / "training" / "base_checkpoint.txt").exists())
|
|
|
|
|
|
class VoiceExportConfigTests(unittest.TestCase):
|
|
"""An exported model needs its .onnx.json even after piper_config.json was deleted."""
|
|
|
|
def test_voice_config_rebuilt_from_training_run(self) -> None:
|
|
try:
|
|
from piper.config import PiperConfig
|
|
except ImportError:
|
|
self.skipTest("piper-tts not installed")
|
|
from javis.voices.piper_train import _rebuild_voice_config
|
|
|
|
with tempfile.TemporaryDirectory() as d:
|
|
run = Path(d) / "lightning_logs" / "version_0"
|
|
run.mkdir(parents=True)
|
|
(run / "config.yaml").write_text(
|
|
"data:\n espeak_voice: de\n sample_rate: 22050\n num_symbols: 256\n"
|
|
" phoneme_type: null\n phonemes_path: null\n vowel_clusters: null\n",
|
|
encoding="utf-8")
|
|
dest = Path(d) / "piper_model.onnx.json"
|
|
self.assertTrue(_rebuild_voice_config(Path(d), dest))
|
|
cfg = PiperConfig.from_dict(json.loads(dest.read_text(encoding="utf-8")))
|
|
self.assertEqual((cfg.espeak_voice, cfg.sample_rate, cfg.num_symbols), ("de", 22050, 256))
|
|
self.assertIn("a", cfg.phoneme_id_map)
|
|
|
|
(run / "config.yaml").write_text(
|
|
"data:\n espeak_voice: de\n phonemes_path: custom.json\n", encoding="utf-8")
|
|
self.assertFalse(_rebuild_voice_config(Path(d), dest))
|
|
|
|
|
|
class ConfigValidationTests(unittest.TestCase):
|
|
def test_sample_rate_and_threshold_are_validated(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = Path(d) / "config.yaml"
|
|
p.write_text(
|
|
f"paths: {{base_dir: '{Path(d).as_posix()}'}}\n"
|
|
"audio: {sample_rate: 48000}\nspeaker_id: {threshold: 3}\n",
|
|
encoding="utf-8",
|
|
)
|
|
cfg = load_config(p)
|
|
self.assertEqual(cfg.audio.sample_rate, 16000)
|
|
self.assertEqual(cfg.speaker_id.threshold, 0.65)
|
|
|
|
def test_non_mapping_root_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = Path(d) / "config.yaml"
|
|
p.write_text("- just\n- a list\n", encoding="utf-8")
|
|
with self.assertRaises(ValueError):
|
|
load_config(p)
|
|
|
|
|
|
class ApiBodyLimitTests(unittest.TestCase):
|
|
"""API-01: oversized / unsized bodies are rejected before parsing."""
|
|
|
|
def test_body_size_guard(self) -> None:
|
|
from fastapi.testclient import TestClient
|
|
|
|
from javis.api.devices import SlaveDeviceStore
|
|
from javis.api.pairing import PairingCodeManager
|
|
from javis.api.server import build_app
|
|
|
|
with tempfile.TemporaryDirectory() as d:
|
|
cfg = Config(paths=PathsConfig(base_dir=Path(d)))
|
|
cfg.api.max_upload_bytes = 1000
|
|
app = build_app(cfg, orchestrator=None, device_store=SlaveDeviceStore(Path(d) / "s.json"),
|
|
pairing_mgr=PairingCodeManager())
|
|
client = TestClient(app)
|
|
big = client.post("/api/v1/turn", content=b"x" * 200_000,
|
|
headers={"content-type": "application/octet-stream"})
|
|
self.assertEqual(big.status_code, 413)
|
|
unauth = client.post("/api/v1/turn", files={"audio": ("a.wav", b"RIFF", "audio/wav")})
|
|
self.assertEqual(unauth.status_code, 401)
|
|
self.assertEqual(client.get("/api/v1/health").status_code, 200)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|