231 lines
8.2 KiB
Python
231 lines
8.2 KiB
Python
"""End-to-end flow across BOTH Project Context MCP tools.
|
|
|
|
Issue -> get_project_issue_context -> requirement context
|
|
-> search_project_knowledge -> related project knowledge -> evidence
|
|
|
|
No LLM is involved: the "agent" is deterministic test code that takes the
|
|
requirement text tool #1 returned and feeds it to tool #2, which is exactly the
|
|
hand-off the two tools exist to support. Gitea is mocked; knowledge is a
|
|
synthetic workspace under tmp_path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import requests
|
|
from cowork_local.mcp_servers.project_context.foundation import (
|
|
IdentityContext,
|
|
ProjectContextRuntime,
|
|
)
|
|
from cowork_local.mcp_servers.project_context.runtime import (
|
|
ProjectProviderResolver,
|
|
ProjectScopePolicy,
|
|
)
|
|
from cowork_local.mcp_servers.project_context.server import dispatch
|
|
|
|
PROJECT = "cowork-local"
|
|
OTHER_PROJECT = "other-customer"
|
|
FAKE_TOKEN = "e2e-test-token" # noqa: S105 - test-only sentinel, never a real credential
|
|
|
|
ISSUE_BODY = """The login screen must lock an account after repeated failed attempts.
|
|
|
|
# Acceptance Criteria
|
|
|
|
- [ ] The account locks after five failed login attempts.
|
|
- [ ] An operator can clear the lock from the admin console.
|
|
|
|
# Definition of Done
|
|
|
|
- [ ] Release notes updated.
|
|
"""
|
|
|
|
|
|
@dataclass
|
|
class _Response:
|
|
status_code: int
|
|
payload: dict[str, Any]
|
|
|
|
def json(self) -> dict[str, Any]:
|
|
return self.payload
|
|
|
|
|
|
@pytest.fixture
|
|
def identity() -> IdentityContext:
|
|
return IdentityContext(
|
|
actor_id="agent-e2e",
|
|
org_unit="fsg",
|
|
customer="internal",
|
|
project=PROJECT,
|
|
granted_scopes=frozenset({"read"}),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def wired_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
"""Both providers wired: a mocked Gitea issue and a synthetic knowledge base."""
|
|
base = tmp_path / "workspaces"
|
|
(base / PROJECT).mkdir(parents=True)
|
|
(base / OTHER_PROJECT).mkdir(parents=True)
|
|
|
|
(base / PROJECT / "authentication-basic-design.md").write_text(
|
|
"# Authentication Basic Design\n"
|
|
"The account lock engages after five failed login attempts and is recorded "
|
|
"in the audit log.\n\n"
|
|
"# Unlock Procedure\n"
|
|
"An operator clears the account lock from the admin console.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(base / OTHER_PROJECT / "other-auth.md").write_text(
|
|
"# Other Customer Auth\n"
|
|
"This other-customer account lock policy uses failed login thresholds too.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("PROJECT_CONTEXT_KNOWLEDGE_ROOT", str(base))
|
|
monkeypatch.setenv("GITEA_BASE_URL", "http://gitea.test")
|
|
monkeypatch.setenv("GITEA_TOKEN", FAKE_TOKEN)
|
|
monkeypatch.setenv(
|
|
"PROJECT_CONTEXT_REPO_MAP", json.dumps({PROJECT: "gitea-admin/cowork-local"}),
|
|
)
|
|
|
|
def _fake_get(url: str, headers: dict[str, str] | None = None, timeout: float | None = None):
|
|
del headers, timeout
|
|
assert "/issues/7" in url
|
|
return _Response(
|
|
status_code=200,
|
|
payload={
|
|
"title": "Lock the account after repeated failed logins",
|
|
"state": "open",
|
|
"body": ISSUE_BODY,
|
|
"html_url": "http://gitea.test/gitea-admin/cowork-local/issues/7",
|
|
"updated_at": "2026-09-01T09:00:00Z",
|
|
},
|
|
)
|
|
|
|
monkeypatch.setattr(requests, "get", _fake_get)
|
|
return base
|
|
|
|
|
|
def production_runtime(identity: IdentityContext) -> ProjectContextRuntime:
|
|
"""The real policy and the real provider resolver — no injected doubles."""
|
|
return ProjectContextRuntime(
|
|
identity=identity,
|
|
policy=ProjectScopePolicy(),
|
|
credential_resolver=ProjectProviderResolver(),
|
|
)
|
|
|
|
|
|
def derive_query(issue_context: dict[str, Any]) -> str:
|
|
"""Stand-in for the agent: turn the requirement into a knowledge query."""
|
|
first_criterion = issue_context["acceptance_criteria"][0]
|
|
words = re.findall(r"[A-Za-z]+", first_criterion.casefold())
|
|
stopwords = {"the", "a", "an", "after", "can", "from", "is", "must", "and"}
|
|
return " ".join(word for word in words if word not in stopwords)
|
|
|
|
|
|
def test_issue_context_feeds_knowledge_search_with_evidence(
|
|
identity: IdentityContext, wired_environment: Path,
|
|
) -> None:
|
|
runtime = production_runtime(identity)
|
|
|
|
# ---- Step 1: Issue -> requirement context ---------------------------
|
|
issue_result = dispatch(
|
|
"get_project_issue_context",
|
|
{"project_id": PROJECT, "issue_key": "7"},
|
|
runtime,
|
|
)
|
|
assert issue_result.ok is True, issue_result.payload
|
|
issue = issue_result.payload
|
|
|
|
assert issue["title"] == "Lock the account after repeated failed logins"
|
|
assert issue["status"] == "open"
|
|
# Acceptance criteria are scoped to their own heading — Definition of Done
|
|
# items must not bleed in.
|
|
assert issue["acceptance_criteria"] == [
|
|
"The account locks after five failed login attempts.",
|
|
"An operator can clear the lock from the admin console.",
|
|
]
|
|
assert "Release notes updated." not in issue["acceptance_criteria"]
|
|
assert issue["source"]["url"].startswith("http://gitea.test/")
|
|
assert issue["source"]["revision"]
|
|
|
|
# ---- Step 2: requirement -> related project knowledge ---------------
|
|
query = derive_query(issue)
|
|
knowledge_result = dispatch(
|
|
"search_project_knowledge",
|
|
{"project_id": PROJECT, "query": query},
|
|
runtime,
|
|
)
|
|
assert knowledge_result.ok is True, knowledge_result.payload
|
|
knowledge = knowledge_result.payload
|
|
|
|
assert knowledge["items"], f"the design doc must be found for query {query!r}"
|
|
top = knowledge["items"][0]
|
|
assert top["document_id"] == "authentication-basic-design.md"
|
|
assert "account lock" in top["excerpt"].casefold()
|
|
|
|
# ---- Step 3: every answer carries openable evidence -----------------
|
|
assert top["source"]["system"] == "cowork-workspace"
|
|
assert top["source"]["url"].startswith("file://")
|
|
assert top["source"]["revision"].startswith("mtime:")
|
|
assert top["chunk_id"].startswith(top["document_id"])
|
|
|
|
# ---- The two tools stay inside the same project ---------------------
|
|
retrieved = json.dumps(knowledge["items"])
|
|
assert OTHER_PROJECT not in retrieved
|
|
assert "other-auth.md" not in retrieved
|
|
for item in knowledge["items"]:
|
|
assert f"/{PROJECT}/" in item["source"]["url"]
|
|
|
|
# ---- Both steps are independently traceable -------------------------
|
|
assert issue["correlation_id"] != knowledge["correlation_id"]
|
|
|
|
# ---- Neither step leaked the credential -----------------------------
|
|
combined = json.dumps(issue) + json.dumps(knowledge)
|
|
assert FAKE_TOKEN not in combined
|
|
|
|
|
|
def test_the_same_flow_is_denied_for_an_out_of_scope_project(
|
|
identity: IdentityContext, wired_environment: Path,
|
|
) -> None:
|
|
"""Both tools refuse the same out-of-scope project the same way."""
|
|
runtime = production_runtime(identity)
|
|
|
|
issue_result = dispatch(
|
|
"get_project_issue_context",
|
|
{"project_id": OTHER_PROJECT, "issue_key": "7"},
|
|
runtime,
|
|
)
|
|
knowledge_result = dispatch(
|
|
"search_project_knowledge",
|
|
{"project_id": OTHER_PROJECT, "query": "account lock"},
|
|
runtime,
|
|
)
|
|
|
|
assert issue_result.ok is False
|
|
assert knowledge_result.ok is False
|
|
assert issue_result.payload["error"]["code"] == "DENIED"
|
|
assert knowledge_result.payload["error"]["code"] == "DENIED"
|
|
|
|
|
|
def test_both_tools_are_advertised_as_read_only_context_tools() -> None:
|
|
"""The MVP surface is exactly two production-oriented read tools."""
|
|
from cowork_local.mcp_servers.project_context.registry import TOOLS_BY_NAME
|
|
|
|
for name in ("get_project_issue_context", "search_project_knowledge"):
|
|
tool = TOOLS_BY_NAME[name]
|
|
schema = tool.input_model.model_json_schema()
|
|
assert schema.get("additionalProperties") is False
|
|
# No write-shaped argument exists anywhere on the input contract.
|
|
for field in schema["properties"]:
|
|
assert not any(
|
|
verb in field
|
|
for verb in ("write", "update", "create", "delete", "comment", "body")
|
|
), f"{name}.{field} looks like a write surface"
|