112 lines
6.4 KiB
Python
112 lines
6.4 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__), "..", "..", ".."))
|
|
|
|
|
|
def load_module(name, relative):
|
|
spec = importlib.util.spec_from_file_location(name, os.path.join(ROOT, relative))
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
ORCHESTRATOR = load_module("goal_orchestrator_patch", "packages/casan-harness/scripts/bash/goal-orchestrator.py")
|
|
EXECUTOR = load_module("goal_patch_executor", "packages/casan-harness/scripts/bash/goal-patch-executor.py")
|
|
|
|
|
|
class Result:
|
|
def __init__(self, code=0, stdout="", stderr=""):
|
|
self.returncode = code
|
|
self.stdout = stdout
|
|
self.stderr = stderr
|
|
|
|
|
|
class GoalPatchWorkflowTests(unittest.TestCase):
|
|
def test_vietnamese_completion_goal_is_write_intent(self):
|
|
self.assertTrue(ORCHESTRATOR.requests_side_effect("Hoàn thành component KeyResultDetail với form update progress đầy đủ"))
|
|
|
|
def test_extract_patch_requires_unified_diff(self):
|
|
with self.assertRaisesRegex(ValueError, "goal_patch_missing"):
|
|
ORCHESTRATOR.extract_patch("implementation plan only")
|
|
value = ORCHESTRATOR.extract_patch("```diff\ndiff --git a/apps/okr/frontend/a.ts b/apps/okr/frontend/a.ts\n--- a/apps/okr/frontend/a.ts\n+++ b/apps/okr/frontend/a.ts\n@@ -1 +1 @@\n-a\n+b\n```")
|
|
self.assertTrue(value.startswith("diff --git"))
|
|
|
|
def test_write_contract_rejects_a_truncated_diff_before_h3(self):
|
|
truncated = "diff --git a/apps/okr/frontend/a.ts b/apps/okr/frontend/a.ts\n--- a/apps/okr/frontend/a.ts\n+++ b/apps/okr/frontend/a.ts\n@@ -1 +1 @@\n-a\n"
|
|
with self.assertRaisesRegex(ValueError, "goal_patch_check_failed"):
|
|
ORCHESTRATOR.validate_write_output(truncated)
|
|
|
|
def test_invalid_write_output_gets_one_bounded_repair_attempt(self):
|
|
repaired = "```diff\ndiff --git a/apps/okr/frontend/a.ts b/apps/okr/frontend/a.ts\n--- a/apps/okr/frontend/a.ts\n+++ b/apps/okr/frontend/a.ts\n@@ -1 +1 @@\n-a\n+b\n```"
|
|
with patch.dict(os.environ, {"CASAN_GOAL_PATCH_REPAIR_ATTEMPTS": "1"}, clear=False), \
|
|
patch.object(ORCHESTRATOR, "call_model", return_value=(True, repaired, {"output_tokens": 9}, "ok")) as call:
|
|
ok, value, usage, reason = ORCHESTRATOR.repair_write_output("ollama:test", "original prompt", "implementation plan")
|
|
self.assertTrue(ok)
|
|
self.assertEqual(reason, "ok")
|
|
self.assertEqual(usage["output_tokens"], 9)
|
|
self.assertIn("diff --git", value)
|
|
self.assertEqual(call.call_count, 1)
|
|
|
|
def test_write_output_repair_can_be_disabled(self):
|
|
with patch.dict(os.environ, {"CASAN_GOAL_PATCH_REPAIR_ATTEMPTS": "0"}, clear=False), \
|
|
patch.object(ORCHESTRATOR, "call_model") as call:
|
|
ok, _, _, reason = ORCHESTRATOR.repair_write_output("ollama:test", "original prompt", "implementation plan")
|
|
self.assertFalse(ok)
|
|
self.assertEqual(reason, "goal_patch_missing")
|
|
call.assert_not_called()
|
|
|
|
def test_repair_prefers_direct_cloud_over_gateway_and_local(self):
|
|
repaired = "diff --git a/apps/okr/frontend/a.ts b/apps/okr/frontend/a.ts\n--- a/apps/okr/frontend/a.ts\n+++ b/apps/okr/frontend/a.ts\n@@ -1 +1 @@\n-a\n+b\n"
|
|
with (
|
|
patch.dict(os.environ, {"CASAN_GOAL_PATCH_REPAIR_ATTEMPTS": "1", "CASAN_GOAL_PATCH_REPAIR_MODELS": "openai:gpt-4o-mini,anthropic:claude,openai-compatible:auto/coding,ollama:ornith"}, clear=False),
|
|
patch.object(ORCHESTRATOR, "call_model", return_value=(True, repaired, {}, "ok")) as call,
|
|
):
|
|
ok, _, usage, reason = ORCHESTRATOR.repair_write_output("ollama:ornith", "original", "plan")
|
|
self.assertTrue(ok)
|
|
self.assertEqual(reason, "ok")
|
|
self.assertEqual(usage["repair_model"], "openai:gpt-4o-mini")
|
|
self.assertEqual(call.call_args.args[0], "openai:gpt-4o-mini")
|
|
self.assertEqual(call.call_args.kwargs["max_output_tokens"], ORCHESTRATOR.patch_output_tokens())
|
|
|
|
def test_patch_outside_workspace_is_denied(self):
|
|
job = {"workspace": {"context_roots": ["apps/okr/frontend"]}}
|
|
content = "diff --git a/package.json b/package.json\n--- a/package.json\n+++ b/package.json\n@@ -1 +1 @@\n-a\n+b\n"
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = os.path.join(directory, "job.json")
|
|
with self.assertRaisesRegex(ValueError, "goal_patch_outside_workspace"):
|
|
ORCHESTRATOR.validate_and_store_patch(path, job, content)
|
|
|
|
def test_patch_rename_source_outside_workspace_is_denied(self):
|
|
job = {"workspace": {"context_roots": ["apps/okr/frontend"]}}
|
|
content = "diff --git a/package.json b/apps/okr/frontend/package.json\nsimilarity index 100%\nrename from package.json\nrename to apps/okr/frontend/package.json\n"
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
with self.assertRaisesRegex(ValueError, "goal_patch_outside_workspace"):
|
|
ORCHESTRATOR.validate_and_store_patch(os.path.join(directory, "job.json"), job, content)
|
|
|
|
def test_executor_requires_approved_matching_proposal(self):
|
|
job = {"id": "goal-1", "approval": {"id": "AP-1"}, "patch_artifact": {"sha256": "abc"}}
|
|
pending = {"proposals": [{"id": "AP-1", "status": "pending", "action": "goal.workspace.execute", "payload": {"goal_id": "goal-1", "patch_sha256": "abc"}}]}
|
|
with patch.object(EXECUTOR, "run", return_value=Result(stdout=json.dumps(pending))):
|
|
with self.assertRaisesRegex(PermissionError, "GOAL_APPLY_APPROVAL_REQUIRED"):
|
|
EXECUTOR.verify_approval(job)
|
|
|
|
mismatched = {"proposals": [{"id": "AP-1", "status": "approved", "action": "goal.workspace.execute", "payload": {"goal_id": "goal-1", "patch_sha256": "different"}}]}
|
|
with patch.object(EXECUTOR, "run", return_value=Result(stdout=json.dumps(mismatched))):
|
|
with self.assertRaisesRegex(PermissionError, "GOAL_APPLY_PATCH_HASH_MISMATCH"):
|
|
EXECUTOR.verify_approval(job)
|
|
|
|
def test_frontend_patch_runs_build_and_tests(self):
|
|
commands = EXECUTOR.verification_commands(["apps/okr/frontend/src/pages/KeyResultDetail.tsx"])
|
|
self.assertIn(["npm", "run", "build", "-w", "@ainative-okr/frontend"], commands)
|
|
self.assertIn(["npm", "test", "-w", "@ainative-okr/frontend"], commands)
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|