165 lines
5.7 KiB
Python
165 lines
5.7 KiB
Python
import importlib.util
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ADAPTER_PATH = Path(__file__).with_name("opencode-tensorx-adapter.py")
|
|
SPEC = importlib.util.spec_from_file_location("opencode_tensorx_adapter", ADAPTER_PATH)
|
|
ADAPTER = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(ADAPTER)
|
|
|
|
|
|
class OpenCodeTensorXAdapterTests(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"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|