From 98f4a1ed771d1dfe11628c5edf06891b2f1be2dd Mon Sep 17 00:00:00 2001 From: taiph9 Date: Tue, 25 Aug 2026 22:54:56 +0900 Subject: [PATCH] create feature mcp-issue-context --- .../project_context/providers/issue.py | 374 +++++++++- tests/test_project_context_issue.py | 646 ++++++++++++++++++ 2 files changed, 1018 insertions(+), 2 deletions(-) create mode 100644 tests/test_project_context_issue.py diff --git a/mcp_servers/project_context/providers/issue.py b/mcp_servers/project_context/providers/issue.py index 665db9b..c2e551b 100644 --- a/mcp_servers/project_context/providers/issue.py +++ b/mcp_servers/project_context/providers/issue.py @@ -1,11 +1,124 @@ -"""Provider boundary owned with get_project_issue_context.""" +"""Provider boundary owned with get_project_issue_context. + +Read-only adapter that maps a Gitea issue/pull-request onto the neutral +``IssueContextOutput`` schema declared in ``tools/issue_context.py``. Nothing +here changes the shared runtime, the registry, or the tool schema — this +module only fills in ``build_provider`` so ``runtime.py``'s +``PROVIDER_FACTORIES["get_project_issue_context"]`` resolves to a real, +read-only Gitea client instead of :class:`UnconfiguredIssueProvider`. + +Design notes (documented here so a reviewer does not have to guess): + +* Credentials (``GITEA_TOKEN``) and the base URL are read INSIDE + ``build_provider``/``_resolve_target`` only — never at import time — because + ``build_provider`` is only ever invoked by + ``ProjectProviderResolver.resolve`` (see ``runtime.py``), which + ``server.py::dispatch`` calls strictly AFTER ``policy.decide`` returns + ``True``. Reading env vars at module scope would read them before the + policy check ever runs. +* The provider is bound to exactly one project/repository at construction + time (``identity.project`` -> ``PROJECT_CONTEXT_REPO_MAP``). The + ``project_id`` argument received per call is only used to echo it back in + the response and as a defensive equality check — it is never used to pick + which repository to query. This means a caller can never redirect the + provider to an arbitrary repository by tampering with ``project_id``. +* ``revision`` pins to the issue's own ``updated_at`` timestamp. Gitea issues + (unlike commits/PRs) have no natural commit SHA of their own, so the most + meaningful, verifiable "version marker" Gitea offers for an issue is its + last-modified time: given the same URL and the same ``revision`` string, a + reviewer can confirm whether the issue has changed since this was fetched. +* ``related`` items are derived from BARE ``#`` cross-references + found in the issue body — never from a ``#`` that is only the + label text of an explicit Markdown link (``[other-repo PR #4](http://.../ + other-repo/pulls/4)``). A real issue body regularly links to a pull + request or an issue in a DIFFERENT repository this way; re-guessing that + as "issue #4 in THIS repo" would silently point at the wrong resource + entirely, which is worse than omitting it — the author already gave the + correct URL right there, so extraction strips whole ``[label](url)`` + spans, then any remaining bare URLs, before scanning for mentions. Titles + for the mentions that DO survive are NOT fetched with an extra API call + (kept deliberately simple and bounded) — each related item's title is a + generic, honest placeholder ("Referenced item #"), and its ``url`` + always resolves to a real, openable page in this same repository. This + is a known, documented simplification, not a bug. +* ``acceptance_criteria`` looks for the issue's own "Acceptance Criteria" + Markdown heading (``#`` through ``######``, case-insensitive) and only + collects checklist items (``- [ ]``/``- [x]``) from inside that section, + up to the next heading of equal-or-shallower depth. This matters because + a real work-item body commonly has a SEPARATE "Definition of Done" + section that also uses checklist syntax — without scoping to the + matching heading, those unrelated items would be silently folded into + "acceptance criteria" and misrepresent the issue. When no such heading + exists at all (ad-hoc issues with no fixed template), this falls back to + scanning the whole body, so those issues still get a best-effort result + instead of an always-empty list. +* ``truncated``/``returned``/``remaining``/``next_cursor`` describe pagination + over the ``related`` list only. ``description`` has its own independent, + generous hard safety cap (so the response stays bounded even for a + pathological issue body) and is never silently cut without a visible + notice appended. +* The regexes that derive ``acceptance_criteria``/``related`` never scan more + than ``_MAX_SCAN_CHARS`` of the raw body, and always run on a copy with + ``http(s)://`` links stripped first — otherwise a `#42` inside a URL + fragment (e.g. a doc anchor) would be mis-detected as a cross-reference to + issue #42, and a pathologically large body would cost unbounded regex work + on every call regardless of the display cap on ``description``. +""" from __future__ import annotations +import json +import os +import re +from dataclasses import dataclass +from datetime import datetime, timezone from typing import Any, Protocol +import requests + from ..foundation import IdentityContext, ProviderError +# ---- tunables (documented, not hardcoded secrets) ------------------------- +_REQUEST_TIMEOUT_SECONDS = 10 +_STANDARD_RELATED_PAGE_SIZE = 20 +_FULL_RELATED_PAGE_SIZE = 100 +_SUMMARY_DESCRIPTION_CHARS = 280 +_MAX_DESCRIPTION_CHARS = 20_000 +_MAX_SCAN_CHARS = 200_000 # hard cap on regex work, independent of the display cap above +_TRUNCATION_NOTICE = "\n\n[description truncated: exceeds the display size limit]" + +_ISSUE_KEY_PATTERN = re.compile(r"^[1-9][0-9]*$") +_CHECKLIST_PATTERN = re.compile(r"^[-*]\s+\[[ xX]\]\s+(.+)$", re.MULTILINE) +_MENTION_PATTERN = re.compile(r"(?` that is only the link's label text (often a cross-repo or +# pull-request reference) is never re-guessed as a same-repo issue mention. +_MARKDOWN_LINK_PATTERN = re.compile(r"\[[^\]]*\]\([^)]*\)") +# ATX heading line, e.g. "# Acceptance Criteria" / "## Acceptance Criteria". +_HEADING_PATTERN = re.compile(r"^(#{1,6})[ \t]+(.+?)\s*$", re.MULTILINE) +_ACCEPTANCE_HEADING_NAMES = ("acceptance criteria",) + + +def _extract_heading_section(text: str, heading_names: tuple[str, ...]) -> str | None: + """Return the body of the first ATX heading whose title case-insensitively + matches one of ``heading_names``, up to the next heading of equal or + shallower depth (or the end of ``text``). Returns ``None`` when no such + heading exists, so the caller can fall back to the whole body.""" + wanted = {name.strip().lower() for name in heading_names} + headings = list(_HEADING_PATTERN.finditer(text)) + for index, match in enumerate(headings): + if match.group(2).strip().lower() not in wanted: + continue + level = len(match.group(1)) + end = len(text) + for later in headings[index + 1 :]: + if len(later.group(1)) <= level: + end = later.start() + break + return text[match.end() : end] + return None + class IssueProvider(Protocol): def get_issue_context(self, **arguments: Any) -> dict[str, Any]: ... @@ -20,6 +133,263 @@ class UnconfiguredIssueProvider: ) +@dataclass(frozen=True) +class _GiteaRepoTarget: + base_url: str + owner: str + repo: str + token: str + project_id: str + + +def _load_repo_map() -> dict[str, str]: + raw = os.environ.get("PROJECT_CONTEXT_REPO_MAP", "").strip() + if not raw: + return {} + try: + parsed = json.loads(raw) + except json.JSONDecodeError as exc: + raise ProviderError( + "UNAVAILABLE", + "PROJECT_CONTEXT_REPO_MAP is not valid JSON.", + retryable=False, + ) from exc + if not isinstance(parsed, dict) or not all( + isinstance(k, str) and isinstance(v, str) for k, v in parsed.items() + ): + raise ProviderError( + "UNAVAILABLE", + "PROJECT_CONTEXT_REPO_MAP must be a JSON object of project_id -> 'owner/repo'.", + retryable=False, + ) + return parsed + + +def _resolve_target(identity: IdentityContext) -> _GiteaRepoTarget: + """Read-only credential/target resolution. Only ever called AFTER the + runtime's policy has already granted ALLOW for this identity/tool.""" + base_url = os.environ.get("GITEA_BASE_URL", "").strip().rstrip("/") + token = os.environ.get("GITEA_TOKEN", "").strip() + if not base_url or not token: + raise ProviderError( + "UNAVAILABLE", + "GITEA_BASE_URL/GITEA_TOKEN are not configured for this environment.", + retryable=False, + ) + repo_map = _load_repo_map() + slug = repo_map.get(identity.project, "") + if not slug or "/" not in slug: + raise ProviderError( + "UNAVAILABLE", + "This project is not mapped to an approved Gitea repository.", + retryable=False, + ) + owner, _, repo = slug.partition("/") + if not owner or not repo: + raise ProviderError( + "UNAVAILABLE", + "This project's Gitea repository mapping is malformed.", + retryable=False, + ) + return _GiteaRepoTarget( + base_url=base_url, owner=owner, repo=repo, token=token, project_id=identity.project, + ) + + def build_provider(identity: IdentityContext) -> IssueProvider: """Replace only this factory when wiring the approved read-only issue adapter.""" - return UnconfiguredIssueProvider() + return GiteaIssueProvider(_resolve_target(identity)) + + +class GiteaIssueProvider: + """Read-only adapter mapping one Gitea issue/PR onto the neutral schema.""" + + def __init__(self, target: _GiteaRepoTarget) -> None: + self._target = target + + def get_issue_context( + self, + *, + project_id: str, + issue_key: str, + detail: str, + cursor: str | None, + **_: Any, + ) -> dict[str, Any]: + if project_id != self._target.project_id: + # Defense in depth: the runtime's policy already guarantees this + # can never happen (DENIED would have fired first), but the + # provider never trusts caller-supplied routing regardless. + raise ProviderError( + "INTERNAL", + "Resolved provider does not match the requested project.", + retryable=False, + ) + if not _ISSUE_KEY_PATTERN.match(issue_key): + raise ProviderError( + "INVALID_INPUT", + "issue_key must be a positive work item number.", + retryable=False, + ) + offset = self._decode_cursor(cursor) + + payload = self._fetch_issue(issue_key) + + title = str(payload.get("title") or "") + raw_state = str(payload.get("state") or "") + status = raw_state if raw_state in {"open", "closed"} else "unknown" + body = str(payload.get("body") or "") + description = self._build_description(body, detail) + # Bounded regardless of the actual body size: caps worst-case regex + # cost, independently of `description`'s own display-only cap. + scan_text = body[:_MAX_SCAN_CHARS] + acceptance_section = _extract_heading_section(scan_text, _ACCEPTANCE_HEADING_NAMES) + acceptance_criteria = tuple( + _CHECKLIST_PATTERN.findall(acceptance_section if acceptance_section is not None else scan_text) + ) + related_all = self._extract_related(scan_text, issue_key) + + related_page, returned, remaining, truncated, next_cursor = self._paginate_related( + related_all, detail, offset, + ) + + html_url = str( + payload.get("html_url") + or f"{self._target.base_url}/{self._target.owner}/{self._target.repo}/issues/{issue_key}" + ) + updated_at = str(payload.get("updated_at") or "") + retrieved_at = datetime.now(timezone.utc).isoformat() + + return { + "project_id": project_id, + "issue_key": issue_key, + "title": title, + "status": status, + "description": description, + "acceptance_criteria": acceptance_criteria, + "related": related_page, + "source": { + "system": "gitea", + "url": html_url, + "revision": f"issue-updated:{updated_at or retrieved_at}", + "retrieved_at": retrieved_at, + }, + "truncated": truncated, + "returned": returned, + "remaining": remaining, + "next_cursor": next_cursor, + } + + # ---- internals --------------------------------------------------- + def _decode_cursor(self, cursor: str | None) -> int: + if cursor is None: + return 0 + try: + offset = int(cursor) + except ValueError as exc: + raise ProviderError("INVALID_INPUT", "cursor is not valid.", retryable=False) from exc + if offset < 0: + raise ProviderError("INVALID_INPUT", "cursor is not valid.", retryable=False) + return offset + + def _build_description(self, body: str, detail: str) -> str: + text = body.strip() + if detail == "summary": + return text.split("\n\n", 1)[0][:_SUMMARY_DESCRIPTION_CHARS] + if len(text) > _MAX_DESCRIPTION_CHARS: + return text[:_MAX_DESCRIPTION_CHARS] + _TRUNCATION_NOTICE + return text + + def _extract_related(self, body: str, issue_key: str) -> tuple[dict[str, str], ...]: + # Strip whole `[label](url)` spans FIRST (as one unit) so a `#` + # that only appears as a Markdown link's label — often a cross-repo or + # pull-request reference with its own, possibly different, URL right + # there — is never re-guessed as "issue # in this repo". + text_without_links = _MARKDOWN_LINK_PATTERN.sub(" ", body) + # Then strip any remaining bare URLs so a doc-anchor link like + # ".../guide#42" is never mistaken for a cross-reference to issue #42. + text_without_urls = _URL_PATTERN.sub(" ", text_without_links) + numbers = sorted({int(n) for n in _MENTION_PATTERN.findall(text_without_urls) if n != issue_key}) + return tuple( + { + "item_id": str(number), + "relation": "mentioned", + "title": f"Referenced item #{number}", + "url": f"{self._target.base_url}/{self._target.owner}/{self._target.repo}/issues/{number}", + } + for number in numbers + ) + + def _paginate_related( + self, + related_all: tuple[dict[str, str], ...], + detail: str, + offset: int, + ) -> tuple[tuple[dict[str, str], ...], int, int, bool, str | None]: + if detail == "summary": + # Summary mode intentionally omits related items outright; it is + # not a size-limit truncation, so callers who need them must + # call again with detail="standard"/"full". + remaining = len(related_all) + return (), 0, remaining, remaining > 0, None + + page_size = _FULL_RELATED_PAGE_SIZE if detail == "full" else _STANDARD_RELATED_PAGE_SIZE + page = related_all[offset : offset + page_size] + remaining = max(0, len(related_all) - (offset + page_size)) + truncated = remaining > 0 + next_cursor = str(offset + page_size) if truncated else None + return page, len(page), remaining, truncated, next_cursor + + def _fetch_issue(self, issue_key: str) -> dict[str, Any]: + url = ( + f"{self._target.base_url}/api/v1/repos/{self._target.owner}/" + f"{self._target.repo}/issues/{issue_key}" + ) + headers = {"Authorization": f"token {self._target.token}"} + try: + response = requests.get(url, headers=headers, timeout=_REQUEST_TIMEOUT_SECONDS) + except requests.exceptions.Timeout as exc: + raise ProviderError( + "UPSTREAM_TIMEOUT", "The Gitea request timed out.", retryable=True, + ) from exc + except requests.exceptions.RequestException as exc: + # Never surface str(exc) — it can embed the request URL/host and, + # in some transport errors, request headers. + raise ProviderError( + "UPSTREAM_ERROR", "The Gitea request failed.", retryable=True, + ) from exc + + if response.status_code == 404: + raise ProviderError( + "NOT_FOUND", + "The work item was not found or is not accessible.", + retryable=False, + ) + if response.status_code == 429: + raise ProviderError("RATE_LIMITED", "Gitea rate-limited this request.", retryable=True) + if response.status_code in (401, 403): + raise ProviderError( + "UPSTREAM_ERROR", + "The read-only Gitea credential could not access the repository.", + retryable=False, + ) + if response.status_code >= 500: + raise ProviderError("UPSTREAM_ERROR", "Gitea returned a server error.", retryable=True) + if response.status_code != 200: + raise ProviderError( + "UPSTREAM_ERROR", "Gitea returned an unexpected response.", retryable=False, + ) + + try: + data = response.json() + except ValueError as exc: + raise ProviderError( + "UPSTREAM_ERROR", + "Gitea returned a response that could not be parsed.", + retryable=False, + ) from exc + if not isinstance(data, dict): + raise ProviderError( + "UPSTREAM_ERROR", "Gitea returned an unexpected response shape.", retryable=False, + ) + return data diff --git a/tests/test_project_context_issue.py b/tests/test_project_context_issue.py new file mode 100644 index 0000000..bfc6672 --- /dev/null +++ b/tests/test_project_context_issue.py @@ -0,0 +1,646 @@ +"""Member A's own test suite for get_project_issue_context. + +Every test mocks the Gitea transport (``requests.get``) and never touches a +real network call or a real credential — per Issue #3 / MCP Contract v2: +unit tests must not call Gitea for real or use a real token. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import pytest +import requests + +from cowork_local.mcp_servers.project_context.foundation import ( + IdentityContext, + ProjectContextRuntime, + ProviderError, +) +from cowork_local.mcp_servers.project_context.providers.issue import ( + GiteaIssueProvider, + UnconfiguredIssueProvider, + _GiteaRepoTarget, +) +from cowork_local.mcp_servers.project_context.runtime import ProjectProviderResolver +from cowork_local.mcp_servers.project_context.server import dispatch + +FAKE_TOKEN = "super-secret-token-value" # noqa: S105 - test-only sentinel, never a real credential + + +# --------------------------------------------------------------------------- +# Shared fixtures / test doubles +# --------------------------------------------------------------------------- +@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 + + +class _FakeResponse: + def __init__(self, status_code: int, json_body: Any = "__missing__") -> None: + self.status_code = status_code + self._json_body = json_body + + def json(self) -> Any: + if self._json_body == "__missing__": + raise ValueError("no json body") + return self._json_body + + +class _FakeTransport: + """Drop-in replacement for ``requests.get`` that queues canned results + and records every call it received (url/headers/timeout).""" + + def __init__(self, queue: list[Any]) -> None: + self._queue = list(queue) + self.calls: list[dict[str, Any]] = [] + + def __call__(self, url: str, headers: dict[str, str] | None = None, timeout: float | None = None): + self.calls.append({"url": url, "headers": headers, "timeout": timeout}) + item = self._queue.pop(0) + if isinstance(item, BaseException): + raise item + return item + + +@pytest.fixture +def identity() -> IdentityContext: + return IdentityContext( + actor_id="member-a", + org_unit="fsg", + customer="internal", + project="cowork-local", + granted_scopes=frozenset({"read"}), + ) + + +def _target(**overrides: Any) -> _GiteaRepoTarget: + base = dict( + base_url="http://example.test", + owner="gitea-admin", + repo="cowork-local", + token=FAKE_TOKEN, + project_id="cowork-local", + ) + base.update(overrides) + return _GiteaRepoTarget(**base) + + +def _issue_payload(**overrides: Any) -> dict[str, Any]: + payload = { + "title": "MCP pilot", + "state": "open", + "body": "Build verifiable project context.\n\n- [ ] Every result has a source.", + "html_url": "http://example.test/gitea-admin/cowork-local/issues/1", + "updated_at": "2026-08-20T10:00:00Z", + } + payload.update(overrides) + return payload + + +def _runtime( + identity: IdentityContext, provider: Any, *, allowed: bool = True, +) -> tuple[ProjectContextRuntime, RecordingPolicy, RecordingResolver]: + policy = RecordingPolicy(allowed=allowed) + resolver = RecordingResolver(provider=provider) + return ( + ProjectContextRuntime(identity=identity, policy=policy, credential_resolver=resolver), + policy, + resolver, + ) + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- +def test_happy_path_returns_full_schema_with_openable_source( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + transport = _FakeTransport([_FakeResponse(200, _issue_payload())]) + monkeypatch.setattr(requests, "get", transport) + provider = GiteaIssueProvider(_target()) + app, policy, resolver = _runtime(identity, provider) + + result = dispatch( + "get_project_issue_context", + {"project_id": "cowork-local", "issue_key": "1", "detail": "standard"}, + app, + ) + + assert result.ok is True + assert policy.calls == 1 + assert resolver.calls == 1 + assert result.payload["project_id"] == "cowork-local" + assert result.payload["issue_key"] == "1" + assert result.payload["title"] == "MCP pilot" + assert result.payload["status"] == "open" + assert result.payload["acceptance_criteria"] == ["Every result has a source."] + assert result.payload["correlation_id"] + source = result.payload["source"] + assert source["system"] == "gitea" + assert source["url"].startswith("http://example.test/gitea-admin/cowork-local/issues/1") + assert source["revision"] == "issue-updated:2026-08-20T10:00:00Z" + assert source["retrieved_at"] + # exactly one Gitea call was made, to the expected REST path + assert len(transport.calls) == 1 + assert transport.calls[0]["url"].endswith("/api/v1/repos/gitea-admin/cowork-local/issues/1") + assert transport.calls[0]["headers"] == {"Authorization": f"token {FAKE_TOKEN}"} + + +def test_source_fields_are_all_present_and_well_formed( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, _issue_payload())])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch( + "get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app, + ) + + source = result.payload["source"] + assert source["url"].startswith("http") + assert isinstance(source["revision"], str) and source["revision"] + assert "T" in source["retrieved_at"] # ISO-8601 timestamp, not a placeholder + + +# --------------------------------------------------------------------------- +# Invalid input (before any policy/provider call) +# --------------------------------------------------------------------------- +def test_invalid_input_is_rejected_before_policy_or_provider(identity: IdentityContext) -> None: + app, policy, resolver = _runtime(identity, UnconfiguredIssueProvider()) + + 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 + + +# --------------------------------------------------------------------------- +# DENIED — zero upstream calls, security-critical +# --------------------------------------------------------------------------- +def test_denied_project_never_resolves_credentials_or_calls_gitea( + identity: IdentityContext, +) -> None: + # No transport is patched at all: if the provider were ever reached it + # would hit the real `requests.get` and fail loudly, so this test also + # proves "zero upstream calls" by construction, not just by call count. + app, policy, resolver = _runtime(identity, UnconfiguredIssueProvider(), allowed=False) + + result = dispatch( + "get_project_issue_context", + {"project_id": "some-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_permission_decision_lives_outside_the_tool( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Acceptance criterion: swapping ONLY the policy must change the + outcome, proving `tools/issue_context.py` contains no permission logic + of its own.""" + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, _issue_payload())])) + provider = GiteaIssueProvider(_target()) + arguments = {"project_id": "cowork-local", "issue_key": "1"} + + allowed_app, _, _ = _runtime(identity, provider, allowed=True) + denied_app, _, _ = _runtime(identity, provider, allowed=False) + + allowed_result = dispatch("get_project_issue_context", arguments, allowed_app) + denied_result = dispatch("get_project_issue_context", arguments, denied_app) + + assert allowed_result.ok is True + assert denied_result.ok is False + assert denied_result.payload["error"]["code"] == "DENIED" + + +# --------------------------------------------------------------------------- +# Boundary / failure — distinct, non-leaking error codes +# --------------------------------------------------------------------------- +def test_not_found_issue_maps_to_not_found( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(404)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch( + "get_project_issue_context", {"project_id": "cowork-local", "issue_key": "999999"}, app, + ) + + assert result.ok is False + assert result.payload["error"]["code"] == "NOT_FOUND" + assert result.payload["error"]["suggested_action"] + + +def test_provider_raises_provider_error_directly_for_not_found( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Unit-level check on the provider class itself (not only through + dispatch): the raised exception must carry the right `.code`/`.retryable` + for the runtime to map correctly.""" + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(404)])) + provider = GiteaIssueProvider(_target()) + + with pytest.raises(ProviderError) as exc_info: + provider.get_issue_context( + project_id="cowork-local", issue_key="1", detail="standard", cursor=None, + ) + + assert exc_info.value.code == "NOT_FOUND" + assert exc_info.value.retryable is False + + +def test_upstream_timeout_maps_to_upstream_timeout( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(requests, "get", _FakeTransport([requests.exceptions.Timeout("slow")])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + 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_TIMEOUT" + assert result.payload["error"]["retryable"] is True + + +@pytest.mark.parametrize( + ("status_code", "expected_code"), + [(500, "UPSTREAM_ERROR"), (503, "UPSTREAM_ERROR"), (429, "RATE_LIMITED"), + (401, "UPSTREAM_ERROR"), (403, "UPSTREAM_ERROR")], +) +def test_upstream_status_codes_map_to_distinct_error_codes( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, status_code: int, expected_code: str, +) -> None: + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(status_code)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + assert result.ok is False + assert result.payload["error"]["code"] == expected_code + + +def test_malformed_gitea_response_maps_to_upstream_error( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, json_body="__missing__")])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + 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_provider_output_schema_mismatch_maps_to_upstream_error(identity: IdentityContext) -> None: + class BrokenProvider: + def get_issue_context(self, **_: Any) -> dict[str, Any]: + return {"project_id": "cowork-local"} # missing every other required field + + app, _, _ = _runtime(identity, BrokenProvider()) + + 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" + + +# --------------------------------------------------------------------------- +# Reject before any network call +# --------------------------------------------------------------------------- +def test_invalid_issue_key_format_rejected_before_network_call( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + transport = _FakeTransport([]) # empty queue: a real call would raise IndexError + monkeypatch.setattr(requests, "get", transport) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch( + "get_project_issue_context", {"project_id": "cowork-local", "issue_key": "not-a-number"}, app, + ) + + assert result.ok is False + assert result.payload["error"]["code"] == "INVALID_INPUT" + assert transport.calls == [] + + +def test_invalid_cursor_rejected_before_network_call( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + transport = _FakeTransport([]) + monkeypatch.setattr(requests, "get", transport) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch( + "get_project_issue_context", + {"project_id": "cowork-local", "issue_key": "1", "cursor": "not-a-number"}, + app, + ) + + assert result.ok is False + assert result.payload["error"]["code"] == "INVALID_INPUT" + assert transport.calls == [] + + +# --------------------------------------------------------------------------- +# Fail-closed configuration (build_provider itself, via the real resolver) +# --------------------------------------------------------------------------- +def test_missing_gitea_env_vars_returns_unavailable_with_no_network_call( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("GITEA_BASE_URL", raising=False) + monkeypatch.delenv("GITEA_TOKEN", raising=False) + monkeypatch.delenv("PROJECT_CONTEXT_REPO_MAP", raising=False) + + def _fail_if_called(*_args: Any, **_kwargs: Any) -> Any: + raise AssertionError("Gitea must not be called when the provider is unconfigured") + + monkeypatch.setattr(requests, "get", _fail_if_called) + policy = RecordingPolicy(allowed=True) + app = ProjectContextRuntime(identity=identity, policy=policy, credential_resolver=ProjectProviderResolver()) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + assert result.ok is False + assert result.payload["error"]["code"] == "UNAVAILABLE" + + +def test_project_without_repo_mapping_returns_unavailable( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GITEA_BASE_URL", "http://example.test") + monkeypatch.setenv("GITEA_TOKEN", FAKE_TOKEN) + monkeypatch.setenv("PROJECT_CONTEXT_REPO_MAP", '{"some-other-project": "gitea-admin/other"}') + policy = RecordingPolicy(allowed=True) + app = ProjectContextRuntime(identity=identity, policy=policy, credential_resolver=ProjectProviderResolver()) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + assert result.ok is False + assert result.payload["error"]["code"] == "UNAVAILABLE" + + +@pytest.mark.parametrize( + "raw_map", + [ + "{not valid json", # malformed JSON + '["cowork-local", "gitea-admin/cowork-local"]', # valid JSON, wrong shape (array) + '{"cowork-local": 123}', # valid JSON object, non-string value + ], +) +def test_malformed_repo_map_returns_unavailable_with_no_network_call( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, raw_map: str, +) -> None: + monkeypatch.setenv("GITEA_BASE_URL", "http://example.test") + monkeypatch.setenv("GITEA_TOKEN", FAKE_TOKEN) + monkeypatch.setenv("PROJECT_CONTEXT_REPO_MAP", raw_map) + + def _fail_if_called(*_args: Any, **_kwargs: Any) -> Any: + raise AssertionError("Gitea must not be called when the repo map is malformed") + + monkeypatch.setattr(requests, "get", _fail_if_called) + policy = RecordingPolicy(allowed=True) + app = ProjectContextRuntime(identity=identity, policy=policy, credential_resolver=ProjectProviderResolver()) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + assert result.ok is False + assert result.payload["error"]["code"] == "UNAVAILABLE" + + +# --------------------------------------------------------------------------- +# Truncation + cursor pagination over `related` +# --------------------------------------------------------------------------- +def test_truncation_and_cursor_paginate_related_items( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + mentions = " ".join(f"#{n}" for n in range(2, 27)) # 25 distinct related items + payload = _issue_payload(body=f"See also {mentions}.") + transport = _FakeTransport([_FakeResponse(200, payload), _FakeResponse(200, payload)]) + monkeypatch.setattr(requests, "get", transport) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + first = dispatch( + "get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app, + ) + assert first.ok is True + assert first.payload["returned"] == 20 + assert first.payload["remaining"] == 5 + assert first.payload["truncated"] is True + assert first.payload["next_cursor"] == "20" + assert len(first.payload["related"]) == 20 + assert first.payload["related"][0]["url"].startswith("http://example.test/") + + second = dispatch( + "get_project_issue_context", + {"project_id": "cowork-local", "issue_key": "1", "cursor": first.payload["next_cursor"]}, + app, + ) + assert second.ok is True + assert second.payload["returned"] == 5 + assert second.payload["remaining"] == 0 + assert second.payload["truncated"] is False + assert second.payload["next_cursor"] is None + + +def test_full_detail_uses_a_larger_related_page_than_standard( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression guard: `detail='full'` must genuinely page differently + from `detail='standard'` (100 vs 20) — this was previously unverified.""" + mentions = " ".join(f"#{n}" for n in range(2, 32)) # 30 distinct related items + payload = _issue_payload(body=f"See also {mentions}.") + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, payload)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch( + "get_project_issue_context", + {"project_id": "cowork-local", "issue_key": "1", "detail": "full"}, + app, + ) + + assert result.ok is True + assert result.payload["returned"] == 30 + assert result.payload["remaining"] == 0 + assert result.payload["truncated"] is False + assert result.payload["next_cursor"] is None + + +def test_url_fragment_is_not_mistaken_for_a_related_issue( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression guard: a doc-anchor link like '.../guide#42' must not be + reported as a related item pointing to issue #42, while a plain '#7' + text mention elsewhere in the same body still must be.""" + body = "See http://example.test/gitea-admin/cowork-local/wiki/guide#42 and also #7 directly." + payload = _issue_payload(body=body) + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, payload)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + related_ids = {item["item_id"] for item in result.payload["related"]} + assert related_ids == {"7"} + + +def test_related_excludes_number_that_is_only_a_markdown_link_label( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression guard (found via a real Gitea issue during manual smoke + testing): a Markdown link whose LABEL happens to contain '#' — + e.g. a cross-repository pull-request reference — must not be re-guessed + as a same-repo issue mention, because that silently points at the wrong + resource. A plain '#9' mention elsewhere in the same body must still be + picked up.""" + body = ( + "See [other-repo PR #4](http://example.test/other-repo/pulls/4) " + "and also #9 directly." + ) + payload = _issue_payload(body=body) + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, payload)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + related_ids = {item["item_id"] for item in result.payload["related"]} + assert related_ids == {"9"} + + +def test_acceptance_criteria_is_scoped_to_its_own_heading_not_definition_of_done( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression guard (found via a real Gitea issue during manual smoke + testing): a body with a SEPARATE 'Definition of Done' checklist section + must not have those items folded into acceptance_criteria.""" + body = ( + "# Acceptance Criteria\n\n" + "- [ ] Real acceptance item one.\n" + "- [ ] Real acceptance item two.\n\n" + "# Definition of Done\n\n" + "- [ ] Unrelated DoD item one.\n" + "- [ ] Unrelated DoD item two.\n" + ) + payload = _issue_payload(body=body) + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, payload)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + assert result.payload["acceptance_criteria"] == [ + "Real acceptance item one.", + "Real acceptance item two.", + ] + + +def test_acceptance_criteria_falls_back_to_whole_body_without_a_heading( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + """An issue with no 'Acceptance Criteria' heading at all (no fixed + template) must still get a best-effort result from the whole body, + rather than always coming back empty.""" + body = "Ad-hoc issue, no headings.\n\n- [ ] Just do the thing.\n" + payload = _issue_payload(body=body) + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, payload)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + assert result.payload["acceptance_criteria"] == ["Just do the thing."] + + +def test_summary_detail_omits_related_and_shortens_description( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + long_paragraph = "First paragraph. " * 40 # > 280 chars + payload = _issue_payload(body=f"{long_paragraph}\n\nSecond paragraph mentions #2.") + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(200, payload)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch( + "get_project_issue_context", + {"project_id": "cowork-local", "issue_key": "1", "detail": "summary"}, + app, + ) + + assert result.ok is True + assert len(result.payload["description"]) <= 280 + assert result.payload["related"] == [] + assert result.payload["returned"] == 0 + assert result.payload["remaining"] == 1 + assert result.payload["truncated"] is True + assert result.payload["next_cursor"] is None + + +# --------------------------------------------------------------------------- +# No credential/exception leakage +# --------------------------------------------------------------------------- +def test_unexpected_transport_error_does_not_leak_credential_or_raw_exception( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + leaking_exception = requests.exceptions.ConnectionError( + f"connect failed for token={FAKE_TOKEN} at internal-host:5432" + ) + monkeypatch.setattr(requests, "get", _FakeTransport([leaking_exception])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + 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" + payload_text = str(result.payload) + assert FAKE_TOKEN not in payload_text + assert "internal-host" not in payload_text + + +def test_not_found_message_does_not_distinguish_missing_from_inaccessible( + identity: IdentityContext, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Security requirement: a denial/miss must not reveal whether the + underlying resource exists — the safe_message must stay generic.""" + monkeypatch.setattr(requests, "get", _FakeTransport([_FakeResponse(404)])) + provider = GiteaIssueProvider(_target()) + app, _, _ = _runtime(identity, provider) + + result = dispatch("get_project_issue_context", {"project_id": "cowork-local", "issue_key": "1"}, app) + + message = result.payload["error"]["message"].lower() + assert "not found or is not accessible" in message + assert "does not exist" not in message