121 lines
4.9 KiB
Python
121 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
|
|
SCRIPT = os.path.join(ROOT, "packages", "casan-harness", "scripts", "bash", "goal-orchestrator.py")
|
|
SPEC = importlib.util.spec_from_file_location("goal_orchestrator", SCRIPT)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
def job_file(directory):
|
|
path = os.path.join(directory, "job.json")
|
|
with open(path, "w", encoding="utf-8") as handle:
|
|
json.dump({
|
|
"id": "11111111-1111-1111-1111-111111111111",
|
|
"updated_at": MODULE.now(),
|
|
"stages": [{"id": "cloud-reviewer", "status": "queued"}],
|
|
}, handle)
|
|
return path
|
|
|
|
|
|
class ReviewerFallbackTests(unittest.TestCase):
|
|
def test_account_then_omniroute_then_local_and_persists_ledger(self):
|
|
calls = []
|
|
|
|
def account(provider, prompt, timeout):
|
|
calls.append(("account", provider))
|
|
return False, "", {}, "account_bridge_failed"
|
|
|
|
def model(model, prompt, cloud, timeout):
|
|
calls.append(("model", model, cloud))
|
|
if model.startswith("openai-compatible:"):
|
|
return False, "", {}, "model_timeout"
|
|
return True, "reviewed result", {"output_tokens": 7}, "ok"
|
|
|
|
environment = {
|
|
"CASAN_GOAL_OMNIROUTE_MODELS": "route-a,openai-compatible:route-b",
|
|
"CASAN_GOAL_LOCAL_PROVIDER": "ollama",
|
|
"CASAN_GOAL_REVIEWER_MAX_ATTEMPTS": "4",
|
|
"CASAN_GOAL_REVIEWER_DEADLINE_SEC": "30",
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, environment, clear=False), \
|
|
patch.object(MODULE, "call_account_model", account), patch.object(MODULE, "call_model", model):
|
|
path = job_file(directory)
|
|
ok, result, metadata, reason, reviewer = MODULE.run_reviewer_chain(
|
|
path, "review prompt", "claude", "openai:gpt-primary", "ollama:worker-model"
|
|
)
|
|
with open(path, encoding="utf-8") as handle:
|
|
job = json.load(handle)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual(result, "reviewed result")
|
|
self.assertEqual(metadata["output_tokens"], 7)
|
|
self.assertEqual(reason, "ok")
|
|
self.assertEqual(reviewer["provider"], "ollama")
|
|
self.assertEqual(calls, [
|
|
("account", "claude"),
|
|
("model", "openai-compatible:route-a", True),
|
|
("model", "openai-compatible:route-b", True),
|
|
("model", "ollama:worker-model", False),
|
|
])
|
|
self.assertEqual([row["status"] for row in job["reviewer_attempts"]], ["failed", "failed", "failed", "pass"])
|
|
self.assertEqual(job["stages"][0]["provider"], "ollama")
|
|
self.assertEqual(job["stages"][0]["model"], "ollama:worker-model")
|
|
|
|
def test_attempt_limit_is_bounded(self):
|
|
calls = []
|
|
|
|
def model(model, prompt, cloud, timeout):
|
|
calls.append(model)
|
|
return False, "", {}, "model_timeout"
|
|
|
|
environment = {
|
|
"CASAN_GOAL_OMNIROUTE_MODELS": "route-a,route-b,route-c",
|
|
"CASAN_GOAL_REVIEWER_MAX_ATTEMPTS": "2",
|
|
"CASAN_GOAL_REVIEWER_DEADLINE_SEC": "30",
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, environment, clear=False), patch.object(MODULE, "call_model", model):
|
|
path = job_file(directory)
|
|
ok, _, _, reason, _ = MODULE.run_reviewer_chain(path, "prompt", "", "openai:gpt", "ollama:local")
|
|
with open(path, encoding="utf-8") as handle:
|
|
job = json.load(handle)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertEqual(reason, "model_timeout")
|
|
self.assertEqual(len(calls), 2)
|
|
self.assertEqual(len(job["reviewer_attempts"]), 2)
|
|
|
|
def test_non_retryable_candidate_failure_advances_to_fallbacks(self):
|
|
calls = []
|
|
|
|
def model(model, prompt, cloud, timeout):
|
|
calls.append(model)
|
|
return False, "", {}, "endpoint_not_allowed"
|
|
|
|
environment = {
|
|
"CASAN_GOAL_OMNIROUTE_MODELS": "route-a",
|
|
"CASAN_GOAL_REVIEWER_MAX_ATTEMPTS": "5",
|
|
"CASAN_GOAL_REVIEWER_DEADLINE_SEC": "30",
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, environment, clear=False), patch.object(MODULE, "call_model", model):
|
|
path = job_file(directory)
|
|
ok, _, _, reason, _ = MODULE.run_reviewer_chain(path, "prompt", "", "openai:gpt", "ollama:local")
|
|
with open(path, encoding="utf-8") as handle:
|
|
job = json.load(handle)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertEqual(reason, "endpoint_not_allowed")
|
|
self.assertEqual(calls, ["openai:gpt", "openai-compatible:route-a", "ollama:local"])
|
|
self.assertFalse(job["reviewer_attempts"][0]["retryable"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|