import importlib.util import json import tempfile import unittest from pathlib import Path ADAPTER_PATH = Path(__file__).with_name("opencode-adapter.py") SPEC = importlib.util.spec_from_file_location("opencode_adapter", ADAPTER_PATH) ADAPTER = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(ADAPTER) class OpenCodeAdapterTests(unittest.TestCase): def test_model_reference_keeps_upstream_slashes(self): self.assertEqual( ( "tensorx/qwen/qwen3.8-flash-next", "qwen/qwen3.8-flash-next", ), ADAPTER.normalize_model("qwen/qwen3.8-flash-next"), ) self.assertEqual( ( "tensorx/qwen/qwen3.8-flash-next", "qwen/qwen3.8-flash-next", ), ADAPTER.normalize_model("tensorx/qwen/qwen3.8-flash-next"), ) def test_custom_mode_translates_agents_and_restricts_delegation(self): base = { "provider": { "tensorx": { "models": {"qwen/qwen3.8-flash-next": {"name": "Qwen"}} } } } with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) root = temp / "root" output = root / "run" / "Ergebnisse" root.mkdir() output.mkdir(parents=True) agents = temp / "agents.json" agents.write_text( json.dumps( { "reviewer": { "description": "Prueft Fakten", "prompt": "Pruefe nur die zugewiesenen Fakten.", } } ), encoding="utf-8", ) config = ADAPTER.build_run_config( base, "tensorx/qwen/qwen3.8-flash-next", "qwen/qwen3.8-flash-next", "custom", root, output, agents, ) self.assertEqual("allow", config["permission"]["task"]["reviewer"]) self.assertEqual("deny", config["permission"]["task"]["*"]) self.assertEqual("subagent", config["agent"]["reviewer"]["mode"]) self.assertEqual( "tensorx/qwen/qwen3.8-flash-next", config["agent"]["reviewer"]["model"], ) self.assertEqual("deny", config["agent"]["reviewer"]["permission"]["task"]) self.assertIn( ADAPTER.normalized_path(output) + "/**", config["permission"]["edit"], ) self.assertIn( "run/Ergebnisse/**", config["permission"]["edit"], ) def test_output_permissions_include_parent_relative_worktree_path(self): with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) root = temp / "QuellCode" / "Product" output = temp / "Versuche" / "Lauf" / "Ergebnisse" root.mkdir(parents=True) output.mkdir(parents=True) patterns = ADAPTER.output_permission_patterns(root, output, temp) self.assertIn("Versuche/Lauf/Ergebnisse/**", patterns) self.assertIn("../../Versuche/Lauf/Ergebnisse", patterns) self.assertIn("../../Versuche/Lauf/Ergebnisse/**", patterns) self.assertIn(ADAPTER.normalized_path(output) + "/**", patterns) def test_normalize_result_uses_exported_metrics_and_tool_calls(self): session = { "info": { "id": "ses_test", "version": "1.18.25", "model": { "id": "qwen/qwen3.8-flash-next", "providerID": "tensorx", }, "tokens": { "input": 100, "output": 10, "reasoning": 5, "cache": {"read": 20, "write": 0}, }, "cost": 0.1, }, "messages": [ { "info": {"role": "assistant", "finish": "stop"}, "parts": [ { "type": "tool", "tool": "task", "state": { "status": "completed", "input": { "subagent_type": "reviewer", "description": "Pruefen", }, }, }, {"type": "text", "text": "Fertig"}, ], } ], } with tempfile.TemporaryDirectory() as temp_dir: output = Path(temp_dir) (output / "StRS.md").write_text("Inhalt", encoding="utf-8") result = ADAPTER.normalize_result( session=session, events=[], model_ref="tensorx/qwen/qwen3.8-flash-next", mode="custom", effort="low", exit_code=0, timed_out=False, interrupted=False, duration_s=1.5, output_dir=output, errors=[], ) self.assertFalse(result["is_error"]) self.assertEqual(135, result["usage"]["total_tokens"]) self.assertEqual(1, result["tool_call_count"]) self.assertEqual(1, result["subagent_stats"]["spawned"]) self.assertEqual({"reviewer": 1}, result["subagent_stats"]["by_type"]) self.assertEqual("Fertig", result["result"]) self.assertEqual(1, len(result["written_files"])) class ReadonlyShellTests(unittest.TestCase): """Denylist: alles erlaubt ausser schreibenden und bauenden Kommandos.""" def test_catch_all_allow_comes_first(self): # OpenCode wertet der Reihe nach aus, die letzte passende Regel gewinnt. # Stuende das Catch-all hinten, waere jede Sperre wirkungslos. perms = ADAPTER.readonly_shell_permissions() self.assertEqual("allow", perms["*"]) self.assertEqual("*", next(iter(perms))) def test_writing_and_building_commands_are_denied(self): perms = ADAPTER.readonly_shell_permissions() for muster in ("rm *", "mv *", "sed -i*", "git commit*", "git push*", "dotnet *", "msbuild *", "npm install*", "Remove-Item *", "Set-Content *", "Out-File *"): self.assertEqual("deny", perms[muster], muster) def test_reading_commands_need_no_rule(self): # Der Kern der Umstellung: Lesekommandos - auch als Pipeline - sind # erlaubt, ohne einzeln aufgefuehrt zu sein. Genau daran scheiterte # 'Get-ChildItem ... | Format-Table ...' unter der Allowlist. perms = ADAPTER.readonly_shell_permissions() for kommando in ("ls -la", "cat foo.cs", "rg muster", "Get-ChildItem -Recurse | Format-Table Name", "find . -name *.cs"): self.assertNotIn(kommando, perms) self.assertEqual("allow", perms["*"]) def test_no_reading_command_is_denied(self): verboten_praefixe = ("ls", "cat", "rg", "grep", "find", "head", "tail", "Get-ChildItem", "Get-Content", "Select-String", "git status", "git log", "dir", "type") for muster, aktion in ADAPTER.readonly_shell_permissions().items(): if aktion != "deny": continue for praefix in verboten_praefixe: self.assertFalse( muster.lower().startswith(praefix.lower()), f"Lesekommando gesperrt: {muster}", ) class SpiegelTests(unittest.TestCase): """Der Spiegel loest relative Ausgabepfade, ohne die Quelle zu beruehren.""" def test_relativer_schreibpfad_landet_im_laufverzeichnis(self): with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) quelle = temp / "codebasis" (quelle / "src").mkdir(parents=True) (quelle / "src" / "a.cs").write_text("Quellcode", encoding="utf-8") (quelle / "README.md").write_text("Liesmich", encoding="utf-8") ausgabe = temp / "lauf" / "Ergebnisse" ausgabe.mkdir(parents=True) spiegel = temp / "spiegel" ADAPTER.erstelle_spiegel(quelle, spiegel, ausgabe, lambda _: None) try: self.assertTrue((spiegel / "src" / "a.cs").is_file()) self.assertTrue((spiegel / "README.md").is_file()) # `Ergebnisse` ist im Spiegel ein echtes Verzeichnis - ein # relativer Schreibpfad landet dort und wird danach uebernommen. (spiegel / "Ergebnisse" / "StRS.md").write_text("X", encoding="utf-8") self.assertFalse((ausgabe / "StRS.md").is_file()) anzahl = ADAPTER.uebernimm_ergebnisse(spiegel, ausgabe, lambda _: None) self.assertEqual(1, anzahl) self.assertTrue((ausgabe / "StRS.md").is_file()) self.assertEqual("X", (ausgabe / "StRS.md").read_text(encoding="utf-8")) finally: ADAPTER.entferne_spiegel(spiegel, lambda _: None) # Der Abbau darf die Quelle nicht durch die Junction hindurch loeschen. self.assertFalse(spiegel.exists()) self.assertTrue((quelle / "src" / "a.cs").is_file()) self.assertEqual("Quellcode", (quelle / "src" / "a.cs").read_text(encoding="utf-8")) self.assertTrue((ausgabe / "StRS.md").is_file()) def test_freigabe_enthaelt_arbeitsverzeichnis_relative_form(self): """resolve() folgt der Junction - die relative Form darf nicht verloren gehen.""" with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) quelle = temp / "codebasis" (quelle / "src").mkdir(parents=True) ausgabe = temp / "lauf" / "Ergebnisse" ausgabe.mkdir(parents=True) spiegel = temp / "spiegel" ADAPTER.erstelle_spiegel(quelle, spiegel, ausgabe, lambda _: None) try: config = ADAPTER.build_run_config( {}, "lmstudio/x", "x", "solo", spiegel, ausgabe, None, provider="lmstudio", zusatz_ausgaben=[spiegel / ausgabe.name], ) finally: ADAPTER.entferne_spiegel(spiegel, lambda _: None) erlaubt = [k for k, v in config["permission"]["edit"].items() if v == "allow"] self.assertIn("Ergebnisse", erlaubt) self.assertIn("Ergebnisse/**", erlaubt) def test_quelle_bleibt_ohne_neue_eintraege(self): with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) quelle = temp / "codebasis" (quelle / "src").mkdir(parents=True) vorher = sorted(p.name for p in quelle.iterdir()) ausgabe = temp / "lauf" / "Ergebnisse" ausgabe.mkdir(parents=True) spiegel = temp / "spiegel" ADAPTER.erstelle_spiegel(quelle, spiegel, ausgabe, lambda _: None) (spiegel / "Ergebnisse" / "datei.md").write_text("X", encoding="utf-8") ADAPTER.uebernimm_ergebnisse(spiegel, ausgabe, lambda _: None) ADAPTER.entferne_spiegel(spiegel, lambda _: None) self.assertEqual(vorher, sorted(p.name for p in quelle.iterdir())) class UsageCapturedTests(unittest.TestCase): """Eine Null aus einem laufenden Subagenten ist kein Messwert.""" def _result(self, session, output): return ADAPTER.normalize_result( session=session, events=[], model_ref="lmstudio/google/gemma-4-e4b", mode="builtin", effort="high", exit_code=1, timed_out=True, interrupted=False, duration_s=3600.0, output_dir=output, errors=[], provider="lmstudio", effort_applied=False, ) def test_running_subagent_with_zero_tokens_is_not_captured(self): session = { "info": {"model": {"id": "google/gemma-4-e4b"}, "tokens": {"input": 0, "output": 0, "reasoning": 0, "cache": {"read": 0, "write": 0}}}, "messages": [{ "info": {"role": "assistant"}, "parts": [{"type": "tool", "tool": "task", "state": {"status": "running", "input": {"subagent_type": "explore"}}}], }], } with tempfile.TemporaryDirectory() as temp_dir: result = self._result(session, Path(temp_dir)) self.assertFalse(result["usage_captured"]) self.assertIn("nicht erfasst", result["usage_note"]) self.assertIn("1 Subagent(en)", result["usage_note"]) def test_real_token_counts_are_marked_captured(self): session = { "info": {"model": {"id": "google/gemma-4-e4b"}, "tokens": {"input": 100, "output": 10, "reasoning": 0, "cache": {"read": 0, "write": 0}}}, "messages": [{"info": {"role": "assistant"}, "parts": [{"type": "text", "text": "ok"}]}], } with tempfile.TemporaryDirectory() as temp_dir: result = self._result(session, Path(temp_dir)) self.assertTrue(result["usage_captured"]) self.assertNotIn("usage_note", result) class OpenCodeLmStudioTests(unittest.TestCase): def test_model_reference_uses_lmstudio_provider(self): self.assertEqual( ("lmstudio/google/gemma-4-e4b", "google/gemma-4-e4b"), ADAPTER.normalize_model("google/gemma-4-e4b", "lmstudio"), ) self.assertEqual( ("lmstudio/qwen/qwen3.5-9b", "qwen/qwen3.5-9b"), ADAPTER.normalize_model("lmstudio/qwen/qwen3.5-9b", "lmstudio"), ) def test_template_declares_both_local_models_without_key(self): template = json.loads( (ADAPTER_PATH.parent / "opencode-lmstudio.json").read_text( encoding="utf-8-sig" ) ) provider = template["provider"]["lmstudio"] self.assertEqual( "http://localhost:1234/v1", provider["options"]["baseURL"] ) self.assertEqual( {"google/gemma-4-e4b", "qwen/qwen3.5-9b"}, set(provider["models"]), ) # Lokale Server pruefen den Key nicht; er darf nur ein Platzhalter sein. self.assertEqual("lm-studio", provider["options"]["apiKey"]) def test_build_run_config_pins_loaded_context_window(self): base = json.loads( (ADAPTER_PATH.parent / "opencode-lmstudio.json").read_text( encoding="utf-8-sig" ) ) with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "root" output = root / "run" / "Ergebnisse" output.mkdir(parents=True) config = ADAPTER.build_run_config( base, "lmstudio/google/gemma-4-e4b", "google/gemma-4-e4b", "solo", root, output, None, provider="lmstudio", context_limit=32768, ) model = config["provider"]["lmstudio"]["models"]["google/gemma-4-e4b"] self.assertEqual(32768, model["limit"]["context"]) self.assertEqual("deny", config["permission"]["task"]) self.assertEqual("deny", config["permission"]["webfetch"]) def test_normalize_result_reports_local_runtime_and_zero_cost(self): runtime = { "provider": "lmstudio", "quantization": "Q4_K_M", "arch": "gemma4", "compatibility_type": "gguf", "loaded_context_length": 32768, "max_context_length": 131072, } session = { "info": { "id": "ses_local", "model": {"id": "google/gemma-4-e4b"}, "tokens": {"input": 10, "output": 4, "reasoning": 2, "cache": {"read": 0, "write": 0}}, "cost": 0, }, "messages": [ { "info": {"role": "assistant", "finish": "stop"}, "parts": [{"type": "text", "text": "Fertig"}], } ], } with tempfile.TemporaryDirectory() as temp_dir: output = Path(temp_dir) (output / "StRS.md").write_text("Inhalt", encoding="utf-8") result = ADAPTER.normalize_result( session=session, events=[], model_ref="lmstudio/google/gemma-4-e4b", mode="solo", effort="high", exit_code=0, timed_out=False, interrupted=False, duration_s=2.0, output_dir=output, errors=[], provider="lmstudio", effort_applied=False, local_runtime=runtime, ) self.assertEqual("lmstudio", result["provider"]) self.assertEqual("opencode-lmstudio", result["adapter"]) self.assertFalse(result["effort_applied"]) self.assertEqual(32768, result["context_window"]) self.assertEqual("Q4_K_M", result["local_runtime"]["quantization"]) self.assertEqual(0, result["cost"]) self.assertIn("lokaler Betrieb", result["cost_source"]) self.assertEqual(16, result["usage"]["total_tokens"]) def test_tensorx_result_keeps_remote_shape(self): session = {"info": {"model": {"id": "qwen/qwen3.8-flash-next"}, "tokens": {"input": 1, "output": 1}}, "messages": []} result = ADAPTER.normalize_result( session=session, events=[], model_ref="tensorx/qwen/qwen3.8-flash-next", mode="solo", effort="low", exit_code=0, timed_out=False, interrupted=False, duration_s=1.0, output_dir=Path("."), errors=[], ) self.assertEqual("tensorx", result["provider"]) self.assertEqual("opencode-tensorx", result["adapter"]) self.assertTrue(result["effort_applied"]) self.assertNotIn("local_runtime", result) self.assertNotIn("context_window", result) if __name__ == "__main__": unittest.main()