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 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.8-27b", "qwen/qwen3.8-27b"), ADAPTER.normalize_model("lmstudio/qwen/qwen3.8-27b", "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.8-27b"}, 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()