.pouch/tests/test_ack_omp_worker.py

from __future__ import annotations

import os
import sys
import unittest
from pathlib import Path
from unittest import mock


REPO_ROOT = Path(__file__).resolve().parents[1]
ACK_SCRIPTS = REPO_ROOT / "skills" / "ack" / "scripts"
sys.path.insert(0, str(ACK_SCRIPTS))

import launch_worker  # noqa: E402
import worker_profiles  # noqa: E402


def omp_profile(*, role: str = "developer", permission: str = "workspace-write", approval: str | None = None) -> dict:
    profile = {
        "role": role,
        "cli": "omp",
        "tier": "standard",
        "model": "opencode-go/gpt-5.6-luna",
        "reasoningEffort": "low",
        "permissionMode": permission,
    }
    if approval is not None:
        profile["approvalMode"] = approval
    return profile


def omp_orchestration(**profile_kwargs: object) -> dict:
    return {
        "profileVersion": 1,
        "mode": "orca",
        "allowedWorktrees": ["/repo/demo"],
        "modelAllowlist": {
            "omp": {
                "developer": {"standard": ["opencode-go/gpt-5.6-luna"]},
                "test": {"standard": ["opencode-go/gpt-5.6-luna"]},
            }
        },
        "profiles": {
            "omp-dev-standard": omp_profile(**profile_kwargs),
            "omp-test-standard": omp_profile(role="test"),
        },
        "defaults": {
            "developer": "omp-dev-standard",
            "test": "omp-test-standard",
        },
    }


class OmpProfileValidationTests(unittest.TestCase):
    def test_omp_profile_and_provider_model_allowlist_are_valid(self) -> None:
        self.assertEqual(worker_profiles.validate_orchestration(omp_orchestration()), [])

    def test_opencode_is_not_an_omp_worker_cli(self) -> None:
        routing = omp_orchestration()
        routing["profiles"]["omp-dev-standard"]["cli"] = "opencode"
        routing["modelAllowlist"]["opencode"] = routing["modelAllowlist"].pop("omp")
        errors = worker_profiles.validate_orchestration(routing)
        self.assertTrue(any("must be" in error and "omp" in error for error in errors))

    def test_test_cannot_use_strong_omp_profile(self) -> None:
        routing = omp_orchestration()
        routing["profiles"]["omp-test-standard"]["tier"] = "strong"
        routing["modelAllowlist"]["omp"]["test"]["strong"] = [
            "opencode-go/gpt-5.6-luna"
        ]
        errors = worker_profiles.validate_orchestration(routing)
        self.assertTrue(any("Test may only use standard" in error for error in errors))
        self.assertTrue(any("Test cannot define a strong allowlist" in error for error in errors))


class OmpApprovalModeTests(unittest.TestCase):
    def test_default_workspace_write_uses_yolo(self) -> None:
        argv = worker_profiles.render_worker_argv(
            omp_profile(), "/usr/local/bin/omp", "/repo/demo"
        )
        self.assertEqual(
            argv,
            [
                "/usr/local/bin/omp",
                "--model",
                "opencode-go/gpt-5.6-luna",
                "--thinking",
                "low",
                "--approval-mode",
                "yolo",
                "--cwd",
                "/repo/demo",
                "--no-session",
            ],
        )
        self.assertNotIn("--auto-approve", argv)
        self.assertNotIn("--plan-yolo", argv)

    def test_explicit_write_turns_approval_off(self) -> None:
        argv = worker_profiles.render_worker_argv(
            omp_profile(approval="write"), "/usr/local/bin/omp", "/repo/demo"
        )
        self.assertEqual(argv[argv.index("--approval-mode") + 1], "write")

    def test_explicit_always_ask_on_workspace_write(self) -> None:
        argv = worker_profiles.render_worker_argv(
            omp_profile(approval="always-ask"), "/usr/local/bin/omp", "/repo/demo"
        )
        self.assertEqual(argv[argv.index("--approval-mode") + 1], "always-ask")

    def test_read_only_defaults_to_always_ask(self) -> None:
        argv = worker_profiles.render_worker_argv(
            omp_profile(permission="read-only"), "/usr/local/bin/omp", "/repo/demo"
        )
        self.assertEqual(argv[argv.index("--approval-mode") + 1], "always-ask")

    def test_read_only_rejects_explicit_yolo(self) -> None:
        routing = omp_orchestration(permission="read-only", approval="yolo")
        errors = worker_profiles.validate_orchestration(routing)
        self.assertTrue(any("read-only requires always-ask" in error for error in errors))

    def test_approval_mode_is_only_valid_for_omp(self) -> None:
        routing = omp_orchestration()
        codex = routing["profiles"]["omp-dev-standard"].copy()
        codex["cli"] = "codex"
        codex["approvalMode"] = "yolo"
        routing["profiles"]["codex-dev-standard"] = codex
        routing["modelAllowlist"]["codex"] = {
            "developer": {"standard": ["opencode-go/gpt-5.6-luna"]}
        }
        errors = worker_profiles.validate_orchestration(routing)
        self.assertTrue(any("only valid for cli=omp" in error for error in errors))

    def test_invalid_approval_value_is_rejected(self) -> None:
        routing = omp_orchestration(approval="always-prompt")
        errors = worker_profiles.validate_orchestration(routing)
        self.assertTrue(any("must be yolo/write/always-ask" in error for error in errors))

    def test_omp_executable_identity_matches_only_omp(self) -> None:
        self.assertTrue(worker_profiles.executable_basename_matches_cli("/bin/omp", "omp"))
        self.assertFalse(
            worker_profiles.executable_basename_matches_cli("/bin/opencode", "omp")
        )


class OmpEnvironmentTests(unittest.TestCase):
    def test_environment_passes_only_omp_provider_credential(self) -> None:
        with mock.patch.dict(
            os.environ,
            {
                "OPENCODE_API_KEY": "omp-secret",
                "OPENAI_API_KEY": "must-not-pass",
                "CURSOR_API_KEY": "must-not-pass",
                "UNRELATED_SECRET": "must-not-pass",
            },
            clear=True,
        ):
            environment = launch_worker.worker_environment("omp")
        self.assertEqual(environment["OPENCODE_API_KEY"], "omp-secret")
        self.assertNotIn("OPENAI_API_KEY", environment)
        self.assertNotIn("CURSOR_API_KEY", environment)
        self.assertNotIn("UNRELATED_SECRET", environment)

    def test_unknown_cli_environment_fails_closed(self) -> None:
        with self.assertRaises(launch_worker.LaunchError):
            launch_worker.worker_environment("opencode")


if __name__ == "__main__":
    unittest.main()