cowork-local/tests/test_project_context_mcp_template.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

import pytest
from cowork_local.mcp_servers.project_context.foundation import (
    IdentityContext,
    ProjectContextRuntime,
)
from cowork_local.mcp_servers.project_context.registry import (
    TOOL_NAMES,
    tool_declarations,
)
from cowork_local.mcp_servers.project_context.runtime import require_supported_python
from cowork_local.mcp_servers.project_context.server import dispatch

EXPECTED_TOOLS = {
    "get_project_issue_context",
    "search_project_knowledge",
    "get_project_change_context",
}


@dataclass
class RecordingPolicy:
    allowed: bool
    calls: int = 0

    def decide(self, identity: IdentityContext, tool_name: str, project_id: str) -> bool:
        self.calls += 1
        return self.allowed


@dataclass
class RecordingResolver:
    provider: Any
    calls: int = 0

    def resolve(self, identity: IdentityContext, tool_name: str) -> Any:
        self.calls += 1
        return self.provider


@dataclass(frozen=True)
class FakeProvider:
    response: dict[str, Any]

    def get_issue_context(self, **_: Any) -> dict[str, Any]:
        return dict(self.response)

    def search_knowledge(self, **_: Any) -> dict[str, Any]:
        return dict(self.response)

    def get_change_context(self, **_: Any) -> dict[str, Any]:
        return dict(self.response)


@pytest.fixture
def identity() -> IdentityContext:
    return IdentityContext(
        actor_id="member-a",
        org_unit="fsg",
        customer="internal",
        project="cowork-local",
        granted_scopes=frozenset({"read"}),
    )


def runtime(identity: IdentityContext, response: dict[str, Any], *, allowed: bool = True):
    policy = RecordingPolicy(allowed=allowed)
    resolver = RecordingResolver(provider=FakeProvider(response))
    return ProjectContextRuntime(
        identity=identity,
        policy=policy,
        credential_resolver=resolver,
    ), policy, resolver


def source() -> dict[str, str]:
    return {
        "system": "gitea",
        "url": "http://example.test/gitea-admin/cowork-local/issues/1",
        "revision": "main@abc123",
        "retrieved_at": "2026-08-20T10:00:00Z",
    }


def test_template_exposes_exactly_three_provider_neutral_tools() -> None:
    # Importing the MCP SDK at module scope aborted collection for the ENTIRE
    # suite whenever the SDK was missing, so the guard lives here, inside the only
    # test that touches it. Guarding per-test rather than per-module keeps the
    # other cases -- pure-Python contract checks that need no SDK -- running
    # instead of silently skipping with it.
    #
    # ``mcp`` is in requirements.txt, so a correctly installed checkout runs this
    # test for real; the guard only covers an environment installed by hand.
    types = pytest.importorskip("mcp.types")

    assert set(TOOL_NAMES) == EXPECTED_TOOLS
    declarations = tool_declarations()
    assert {item["name"] for item in declarations} == EXPECTED_TOOLS
    assert all(item["inputSchema"]["additionalProperties"] is False for item in declarations)
    assert all(item["outputSchema"]["additionalProperties"] is False for item in declarations)
    assert all(types.Tool(**item).name in EXPECTED_TOOLS for item in declarations)


def test_runtime_fails_fast_below_python_311() -> None:
    with pytest.raises(RuntimeError, match="requires Python 3.11"):
        require_supported_python((3, 9, 0))


def test_denied_request_never_resolves_credentials_or_calls_provider(
    identity: IdentityContext,
) -> None:
    app, policy, resolver = runtime(identity, {}, allowed=False)

    result = dispatch(
        "get_project_issue_context",
        {"project_id": "other-project", "issue_key": "1"},
        app,
    )

    assert result.ok is False
    assert result.payload["error"]["code"] == "DENIED"
    assert policy.calls == 1
    assert resolver.calls == 0


def test_invalid_input_is_rejected_before_policy(identity: IdentityContext) -> None:
    app, policy, resolver = runtime(identity, {})

    result = dispatch("get_project_issue_context", {"project_id": "cowork-local"}, app)

    assert result.ok is False
    assert result.payload["error"]["code"] == "INVALID_INPUT"
    assert policy.calls == 0
    assert resolver.calls == 0


@pytest.mark.parametrize(
    ("tool_name", "arguments", "response"),
    [
        (
            "get_project_issue_context",
            {"project_id": "cowork-local", "issue_key": "1"},
            {
                "project_id": "cowork-local",
                "issue_key": "1",
                "title": "MCP pilot",
                "status": "open",
                "description": "Build verifiable project context.",
                "acceptance_criteria": ["Every result has a source."],
                "related": [],
                "source": source(),
                "truncated": False,
                "returned": 1,
                "remaining": 0,
                "next_cursor": None,
            },
        ),
        (
            "search_project_knowledge",
            {"project_id": "cowork-local", "query": "MCP setup"},
            {
                "project_id": "cowork-local",
                "query": "MCP setup",
                "items": [
                    {
                        "document_id": "README.md",
                        "chunk_id": "README.md#setup",
                        "title": "Setup",
                        "excerpt": "Install the approved dependencies.",
                        "score": 0.9,
                        "source": source(),
                    }
                ],
                "truncated": False,
                "returned": 1,
                "remaining": 0,
                "next_cursor": None,
            },
        ),
        (
            "get_project_change_context",
            {"project_id": "cowork-local", "change_id": "1"},
            {
                "project_id": "cowork-local",
                "change_id": "1",
                "change_type": "pull-request",
                "title": "Add MCP contract",
                "state": "merged",
                "summary": "Introduces the project context contract.",
                "authors": ["member-c"],
                "files": ["mcp/contract.yaml"],
                "commits": ["abc123"],
                "related_issues": ["1"],
                "source": source(),
                "truncated": False,
                "returned": 1,
                "remaining": 0,
                "next_cursor": None,
            },
        ),
    ],
)
def test_each_member_template_has_a_valid_success_path(
    identity: IdentityContext,
    tool_name: str,
    arguments: dict[str, Any],
    response: dict[str, Any],
) -> None:
    app, policy, resolver = runtime(identity, response)

    result = dispatch(tool_name, arguments, app)

    assert result.ok is True
    assert result.payload["project_id"] == "cowork-local"
    assert result.payload["correlation_id"]
    assert policy.calls == 1
    assert resolver.calls == 1


def test_provider_output_must_match_contract(identity: IdentityContext) -> None:
    app, _, _ = runtime(identity, {"project_id": "cowork-local"})

    result = dispatch(
        "get_project_issue_context",
        {"project_id": "cowork-local", "issue_key": "1"},
        app,
    )

    assert result.ok is False
    assert result.payload["error"]["code"] == "UPSTREAM_ERROR"


def test_unexpected_provider_error_does_not_leak_exception(identity: IdentityContext) -> None:
    class LeakingProvider:
        def get_issue_context(self, **_: Any) -> dict[str, Any]:
            raise RuntimeError("secret provider-token-value")

    policy = RecordingPolicy(allowed=True)
    resolver = RecordingResolver(provider=LeakingProvider())
    app = ProjectContextRuntime(identity=identity, policy=policy, credential_resolver=resolver)

    result = dispatch(
        "get_project_issue_context",
        {"project_id": "cowork-local", "issue_key": "1"},
        app,
    )

    assert result.ok is False
    assert result.payload["error"]["code"] == "UPSTREAM_ERROR"
    assert "secret" not in str(result.payload)