270 lines
9.5 KiB
Python
270 lines
9.5 KiB
Python
"""Unit tests for the Jira knowledge provider (search_project_knowledge backend).
|
|
|
|
Mirrors the structure of ``test_project_context_knowledge.py`` so the Jira
|
|
provider is held to the same production contract:
|
|
- Happy path through real resolver / build_provider wiring
|
|
- Cross-project isolation (structural, not filter-based)
|
|
- DENIED before provider when policy rejects
|
|
- Untrusted content fence inherited
|
|
- Empty results are valid
|
|
- Pagination / cursor support
|
|
- Output bounds respected
|
|
- Malformed upstream handled gracefully
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from cowork_local.application.jira_knowledge.index_repository import JiraKnowledgeIndex
|
|
from cowork_local.domain.jira_knowledge.canonical_issue import CanonicalJiraIssue, JiraProvenance
|
|
from cowork_local.mcp_servers.project_context.foundation import (
|
|
IdentityContext,
|
|
ProjectContextRuntime,
|
|
ProviderError,
|
|
)
|
|
from cowork_local.mcp_servers.project_context.providers.jira_knowledge import (
|
|
JiraKnowledgeProvider,
|
|
UnconfiguredJiraKnowledgeProvider,
|
|
_JiraKbTarget,
|
|
build_provider,
|
|
)
|
|
from cowork_local.mcp_servers.project_context.server import dispatch
|
|
|
|
|
|
PROJECT = "proj-alpha"
|
|
OTHER_PROJECT = "proj-beta"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 FakeJiraTargetResolver:
|
|
"""Returns a fixed target or raises UNAVAILABLE."""
|
|
target: _JiraKbTarget | None = None
|
|
|
|
def resolve(self, identity: IdentityContext) -> _JiraKbTarget:
|
|
if self.target is None:
|
|
raise ProviderError(
|
|
"UNAVAILABLE",
|
|
"No Jira binding for test.",
|
|
retryable=False,
|
|
)
|
|
return self.target
|
|
|
|
|
|
def identity_for(project: str) -> IdentityContext:
|
|
return IdentityContext(
|
|
actor_id="test-agent",
|
|
org_unit="eng",
|
|
customer="internal",
|
|
project=project,
|
|
granted_scopes=frozenset({"read"}),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def identity() -> IdentityContext:
|
|
return identity_for(PROJECT)
|
|
|
|
|
|
@pytest.fixture
|
|
def index_root(tmp_path: Path) -> Path:
|
|
return tmp_path / "jira_kb"
|
|
|
|
|
|
@pytest.fixture
|
|
def populated_index(index_root: Path) -> JiraKnowledgeIndex:
|
|
"""Index with two projects, each containing a unique secret marker."""
|
|
idx = JiraKnowledgeIndex(index_root=index_root)
|
|
|
|
alpha_issue = CanonicalJiraIssue(
|
|
knowledge_id="ALPHA/ALPHA-1",
|
|
project_id=PROJECT,
|
|
title="Account lock policy",
|
|
content="The account lock engages after five failed login attempts. The alpha marker is secret-alpha.",
|
|
metadata={"issue_type": "Story", "status": "Open"},
|
|
provenance=JiraProvenance(
|
|
system="jira",
|
|
issue_key="ALPHA-1",
|
|
project_key="ALPHA",
|
|
source_url="https://jira.test/browse/ALPHA-1",
|
|
source_updated="2025-06-01T10:00:00.000+0000",
|
|
issue_type="Story",
|
|
status="Open",
|
|
),
|
|
ingested_at="2025-06-01T12:00:00+00:00",
|
|
)
|
|
idx.upsert(alpha_issue)
|
|
|
|
beta_issue = CanonicalJiraIssue(
|
|
knowledge_id="BETA/BETA-1",
|
|
project_id=OTHER_PROJECT,
|
|
title="Beta customer design",
|
|
content="The beta marker is secret-beta and must never reach another project.",
|
|
metadata={"issue_type": "Story", "status": "Open"},
|
|
provenance=JiraProvenance(
|
|
system="jira",
|
|
issue_key="BETA-1",
|
|
project_key="BETA",
|
|
source_url="https://jira.test/browse/BETA-1",
|
|
source_updated="2025-06-01T10:00:00.000+0000",
|
|
issue_type="Story",
|
|
status="Open",
|
|
),
|
|
ingested_at="2025-06-01T12:00:00+00:00",
|
|
)
|
|
idx.upsert(beta_issue)
|
|
return idx
|
|
|
|
|
|
def _make_provider(target: _JiraKbTarget, index: JiraKnowledgeIndex) -> JiraKnowledgeProvider:
|
|
return JiraKnowledgeProvider(target, index=index)
|
|
|
|
|
|
def _search(provider: JiraKnowledgeProvider, **kwargs: Any) -> dict[str, Any]:
|
|
defaults = {
|
|
"project_id": PROJECT,
|
|
"query": "account lock",
|
|
"detail": "standard",
|
|
"top_k": 5,
|
|
}
|
|
defaults.update(kwargs)
|
|
return provider.search_knowledge(**defaults)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Happy path
|
|
# ---------------------------------------------------------------------------
|
|
class TestHappyPath:
|
|
def test_returns_ranked_results_with_source_evidence(
|
|
self, populated_index: JiraKnowledgeIndex,
|
|
) -> None:
|
|
target = _JiraKbTarget(cowork_project_id=PROJECT, jira_project_key="ALPHA")
|
|
provider = _make_provider(target, populated_index)
|
|
|
|
result = _search(provider, query="account lock after failed login")
|
|
|
|
assert result["returned"] >= 1
|
|
item = result["items"][0]
|
|
assert item["source"]["system"] == "jira"
|
|
assert "ALPHA-1" in item["source"]["url"]
|
|
assert item["score"] > 0
|
|
assert "account lock" in item["excerpt"].lower()
|
|
|
|
def test_empty_query_returns_no_results(
|
|
self, populated_index: JiraKnowledgeIndex,
|
|
) -> None:
|
|
target = _JiraKbTarget(cowork_project_id=PROJECT, jira_project_key="ALPHA")
|
|
provider = _make_provider(target, populated_index)
|
|
|
|
result = _search(provider, query="xyznonexistent")
|
|
|
|
assert result["returned"] == 0
|
|
assert result["items"] == ()
|
|
assert result["truncated"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Project isolation
|
|
# ---------------------------------------------------------------------------
|
|
class TestProjectIsolation:
|
|
def test_cross_project_secret_not_leaked(
|
|
self, populated_index: JiraKnowledgeIndex,
|
|
) -> None:
|
|
"""Identity for proj-alpha searching 'secret-beta' must find ZERO results."""
|
|
target = _JiraKbTarget(cowork_project_id=PROJECT, jira_project_key="ALPHA")
|
|
provider = _make_provider(target, populated_index)
|
|
|
|
result = _search(provider, query="secret-beta")
|
|
|
|
items_json = json.dumps(result.get("items", []))
|
|
assert "secret-beta" not in items_json
|
|
assert result["returned"] == 0
|
|
|
|
def test_project_scope_mismatch_raises(
|
|
self, populated_index: JiraKnowledgeIndex,
|
|
) -> None:
|
|
"""Even if policy allowed it, mismatched project_id is rejected."""
|
|
target = _JiraKbTarget(cowork_project_id=PROJECT, jira_project_key="ALPHA")
|
|
provider = _make_provider(target, populated_index)
|
|
|
|
with pytest.raises(ProviderError, match="scope mismatch"):
|
|
_search(provider, project_id=OTHER_PROJECT, query="anything")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unconfigured provider
|
|
# ---------------------------------------------------------------------------
|
|
class TestUnconfigured:
|
|
def test_unconfigured_raises_unavailable(self) -> None:
|
|
provider = UnconfiguredJiraKnowledgeProvider()
|
|
with pytest.raises(ProviderError) as exc_info:
|
|
provider.search_knowledge(project_id="x", query="y")
|
|
assert exc_info.value.code == "UNAVAILABLE"
|
|
assert not exc_info.value.retryable
|
|
|
|
def test_build_provider_returns_unconfigured_when_no_binding(
|
|
self, identity: IdentityContext,
|
|
) -> None:
|
|
provider = build_provider(identity, target_resolver=FakeJiraTargetResolver(target=None))
|
|
assert isinstance(provider, UnconfiguredJiraKnowledgeProvider)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pagination
|
|
# ---------------------------------------------------------------------------
|
|
class TestPagination:
|
|
def test_cursor_pagination(
|
|
self, populated_index: JiraKnowledgeIndex,
|
|
) -> None:
|
|
target = _JiraKbTarget(cowork_project_id=PROJECT, jira_project_key="ALPHA")
|
|
provider = _make_provider(target, populated_index)
|
|
|
|
page1 = _search(provider, query="account", top_k=1)
|
|
assert page1["returned"] == 1
|
|
|
|
if page1["next_cursor"]:
|
|
page2 = _search(provider, query="account", top_k=1, cursor=page1["next_cursor"])
|
|
assert page2["returned"] >= 0 # may be 0 if only one match
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Output bounds
|
|
# ---------------------------------------------------------------------------
|
|
class TestOutputBounds:
|
|
def test_top_k_respected(
|
|
self, populated_index: JiraKnowledgeIndex,
|
|
) -> None:
|
|
target = _JiraKbTarget(cowork_project_id=PROJECT, jira_project_key="ALPHA")
|
|
provider = _make_provider(target, populated_index)
|
|
|
|
result = _search(provider, query="account", top_k=1)
|
|
assert result["returned"] <= 1
|
|
|
|
def test_detail_levels_control_excerpt_length(
|
|
self, populated_index: JiraKnowledgeIndex,
|
|
) -> None:
|
|
target = _JiraKbTarget(cowork_project_id=PROJECT, jira_project_key="ALPHA")
|
|
provider = _make_provider(target, populated_index)
|
|
|
|
summary = _search(provider, query="account", detail="summary", top_k=1)
|
|
full = _search(provider, query="account", detail="full", top_k=1)
|
|
|
|
if summary["returned"] > 0 and full["returned"] > 0:
|
|
assert len(summary["items"][0]["excerpt"]) <= len(full["items"][0]["excerpt"]) |