wehub-resource-sync
158 lines
5.5 KiB
Python
158 lines
5.5 KiB
Python
"""Tests for the Codex elicitation protocol adapters.
|
|
|
|
These are pure-function tests — no HTTP or runtime needed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from omnigent.errors import OmnigentError
|
|
from omnigent.server.routes._codex_elicitation import (
|
|
_codex_command_preview,
|
|
_execpolicy_amendment,
|
|
_json_preview,
|
|
_string_list_answer,
|
|
parse_codex_elicitation_request,
|
|
)
|
|
|
|
# ── parse_codex_elicitation_request ──────────────────────────────────
|
|
|
|
|
|
class TestParseCodexElicitationRequest:
|
|
"""Tests for the top-level request parser."""
|
|
|
|
def test_missing_method_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="non-empty method"):
|
|
parse_codex_elicitation_request({"id": 1, "params": {}})
|
|
|
|
def test_empty_method_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="non-empty method"):
|
|
parse_codex_elicitation_request({"id": 1, "method": "", "params": {}})
|
|
|
|
def test_non_dict_params_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="params must be an object"):
|
|
parse_codex_elicitation_request(
|
|
{"id": 1, "method": "mcpServer/elicitation/request", "params": "bad"}
|
|
)
|
|
|
|
def test_missing_id_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="string or integer id"):
|
|
parse_codex_elicitation_request(
|
|
{"method": "mcpServer/elicitation/request", "params": {}}
|
|
)
|
|
|
|
def test_unsupported_method_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="Unsupported"):
|
|
parse_codex_elicitation_request({"id": 1, "method": "unknown/method", "params": {}})
|
|
|
|
def test_valid_mcp_form_request(self) -> None:
|
|
req = parse_codex_elicitation_request(
|
|
{
|
|
"id": 1,
|
|
"method": "mcpServer/elicitation/request",
|
|
"params": {
|
|
"mode": "form",
|
|
"message": "Need input",
|
|
"requestedSchema": {"type": "object"},
|
|
},
|
|
}
|
|
)
|
|
assert req.method == "mcpServer/elicitation/request"
|
|
assert req.params.mode == "form"
|
|
|
|
def test_valid_command_approval(self) -> None:
|
|
req = parse_codex_elicitation_request(
|
|
{
|
|
"id": 2,
|
|
"method": "item/commandExecution/requestApproval",
|
|
"params": {"command": "npm test"},
|
|
}
|
|
)
|
|
assert req.method == "item/commandExecution/requestApproval"
|
|
|
|
|
|
# ── _string_list_answer ──────────────────────────────────────────────
|
|
|
|
|
|
class TestStringListAnswer:
|
|
"""Tests for answer normalization."""
|
|
|
|
def test_string_input(self) -> None:
|
|
assert _string_list_answer("React") == ["React"]
|
|
|
|
def test_empty_string(self) -> None:
|
|
assert _string_list_answer("") == []
|
|
|
|
def test_list_input(self) -> None:
|
|
assert _string_list_answer(["a", "b"]) == ["a", "b"]
|
|
|
|
def test_list_with_non_strings(self) -> None:
|
|
assert _string_list_answer(["a", 123, "b"]) == ["a", "b"]
|
|
|
|
def test_none_input(self) -> None:
|
|
assert _string_list_answer(None) == []
|
|
|
|
def test_numeric_input(self) -> None:
|
|
assert _string_list_answer(42) == ["42"]
|
|
|
|
|
|
# ── _codex_command_preview ───────────────────────────────────────────
|
|
|
|
|
|
class TestCodexCommandPreview:
|
|
"""Tests for command preview extraction."""
|
|
|
|
def test_string_command(self) -> None:
|
|
assert _codex_command_preview({"command": "npm test"}) == "npm test"
|
|
|
|
def test_list_command(self) -> None:
|
|
assert _codex_command_preview({"command": ["npm", "test"]}) == "npm test"
|
|
|
|
def test_empty_command(self) -> None:
|
|
assert _codex_command_preview({"command": ""}) is None
|
|
|
|
def test_missing_command(self) -> None:
|
|
assert _codex_command_preview({}) is None
|
|
|
|
|
|
# ── _json_preview ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestJsonPreview:
|
|
"""Tests for the bounded preview function."""
|
|
|
|
def test_simple_object(self) -> None:
|
|
result = _json_preview({"key": "value"})
|
|
assert '"key"' in result
|
|
|
|
def test_truncated(self) -> None:
|
|
big = {"k": "x" * 2000}
|
|
result = _json_preview(big)
|
|
assert len(result) <= 1024
|
|
|
|
|
|
# ── _execpolicy_amendment ────────────────────────────────────────────
|
|
|
|
|
|
class TestExecpolicyAmendment:
|
|
"""Tests for execpolicy amendment validation."""
|
|
|
|
def test_none_returns_none(self) -> None:
|
|
assert _execpolicy_amendment(None) is None
|
|
|
|
def test_valid_list(self) -> None:
|
|
assert _execpolicy_amendment(["pytest", "-v"]) == ["pytest", "-v"]
|
|
|
|
def test_empty_list_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="non-empty list"):
|
|
_execpolicy_amendment([])
|
|
|
|
def test_non_list_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="non-empty list"):
|
|
_execpolicy_amendment("pytest")
|
|
|
|
def test_list_with_non_strings_raises(self) -> None:
|
|
with pytest.raises(OmnigentError, match="non-empty list"):
|
|
_execpolicy_amendment(["pytest", 42])
|