open-swe/tests/test_jira_webhook_author.py

"""Tests for Jira webhook PR author linking and repo-mapping cascade."""

from __future__ import annotations

import asyncio
from typing import Any
from unittest.mock import AsyncMock, patch

from agent import webapp
from agent.webhooks import jira as jira_webhook


def _full_issue(*, title: str = "Fix the flaky test") -> dict:
    return {
        "key": "PROJ-42",
        "id": "10001",
        "title": title,
        "description": "Do the thing",
        "url": "https://seahaven.atlassian.net/browse/PROJ-42",
        "project_key": "PROJ",
    }


def _issue_data(*, account_id: str | None, email: str | None, name: str = "Jane") -> dict:
    # jira_webhook resolves account_id -> email once and attaches it to
    # comment_author before dispatch (see webapp.jira_webhook).
    return {
        "key": "PROJ-42",
        "project_key": "PROJ",
        "triggering_comment": "@openswe fix this",
        "triggering_comment_id": "10050",
        "comment_author": {"account_id": account_id, "email": email, "name": name},
    }


def _run_process(
    issue_data: dict,
    repo_config: dict[str, str],
    *,
    active_logins: set[str] | None = None,
) -> tuple[dict, dict, str | None]:
    captured: dict[str, Any] = {}
    active = {"jane"} if active_logins is None else active_logins

    async def fake_dispatch(
        thread_id, content, configurable, *, source, metadata=None, client=None
    ):
        captured["configurable"] = configurable
        return {"run_id": "run-1"}

    async def fake_upsert(
        thread_id,
        *,
        source,
        repo_config=None,
        github_login="",
        user_email="",
        title="",
        source_context=None,
    ):
        captured["upsert"] = {"github_login": github_login, "user_email": user_email}
        return None

    async def fake_resolve_login(email):
        captured["resolved_email"] = email
        return "jane" if email == "jane@example.com" else None

    with (
        patch.object(
            jira_webhook.webapp, "generate_thread_id_from_jira_issue", return_value="thread-1"
        ),
        patch.object(
            jira_webhook.webapp,
            "fetch_jira_issue_details",
            new_callable=AsyncMock,
            return_value=_full_issue(),
        ),
        patch.object(
            jira_webhook.webapp,
            "fetch_jira_issue_comments",
            new_callable=AsyncMock,
            return_value=[],
        ),
        patch.object(
            jira_webhook.webapp, "resolve_login_from_email_async", side_effect=fake_resolve_login
        ),
        patch.object(
            jira_webhook.webapp, "is_login_mapped", side_effect=lambda login: login in active
        ),
        patch.object(jira_webhook.webapp, "dispatch_agent_run", side_effect=fake_dispatch),
        patch.object(
            jira_webhook.webapp, "upsert_agent_thread_owner_metadata", side_effect=fake_upsert
        ),
        patch.object(jira_webhook.webapp, "post_jira_trace_comment", new_callable=AsyncMock),
    ):
        asyncio.run(jira_webhook.process_jira_issue(issue_data, repo_config))

    return (
        captured.get("configurable", {}),
        captured.get("upsert", {}),
        captured.get("resolved_email"),
    )


def test_jira_configurable_carries_github_login() -> None:
    configurable, _upsert, resolved_email = _run_process(
        _issue_data(account_id="acc-1", email="jane@example.com"),
        {"owner": "langchain-ai", "name": "open-swe"},
    )

    assert resolved_email == "jane@example.com"
    assert configurable["source"] == "jira"
    assert configurable["github_login"] == "jane"
    assert configurable["user_email"] == "jane@example.com"
    assert configurable["jira_issue"]["key"] == "PROJ-42"
    assert configurable["jira_issue"]["project_key"] == "PROJ"
    assert configurable["jira_issue"]["issue_number"] == "42"


def test_jira_upsert_tags_thread_with_login() -> None:
    _configurable, upsert, _email = _run_process(
        _issue_data(account_id="acc-1", email="jane@example.com"),
        {"owner": "langchain-ai", "name": "open-swe"},
    )

    assert upsert["github_login"] == "jane"
    assert upsert["user_email"] == "jane@example.com"


def test_jira_omits_login_when_unmapped() -> None:
    configurable, upsert, resolved_email = _run_process(
        _issue_data(account_id="acc-2", email="nobody@example.com"),
        {"owner": "langchain-ai", "name": "open-swe"},
    )

    assert resolved_email == "nobody@example.com"
    assert "github_login" not in configurable
    assert upsert["github_login"] == ""


def test_jira_omits_login_when_mapping_not_active() -> None:
    # A resolvable email whose mapping is pending/inactive must not be attributed.
    configurable, upsert, resolved_email = _run_process(
        _issue_data(account_id="acc-1", email="jane@example.com"),
        {"owner": "langchain-ai", "name": "open-swe"},
        active_logins=set(),
    )

    assert resolved_email == "jane@example.com"
    assert "github_login" not in configurable
    assert upsert["github_login"] == ""


def test_jira_omits_login_when_no_email_resolved() -> None:
    configurable, upsert, resolved_email = _run_process(
        _issue_data(account_id=None, email=None),
        {"owner": "langchain-ai", "name": "open-swe"},
    )

    assert resolved_email is None
    assert "github_login" not in configurable
    assert upsert["github_login"] == ""


def test_repo_cascade_uses_project_mapping(monkeypatch) -> None:
    monkeypatch.setattr(
        webapp, "JIRA_PROJECT_TO_REPO", {"PROJ": {"owner": "acme", "name": "widgets"}}
    )
    assert webapp.get_repo_config_from_jira_mapping("PROJ") == {"owner": "acme", "name": "widgets"}


def test_repo_cascade_falls_back_to_default_repo(monkeypatch) -> None:
    monkeypatch.setattr(webapp, "JIRA_PROJECT_TO_REPO", {})
    monkeypatch.setattr(webapp, "DEFAULT_REPO_OWNER", "langchain-ai")
    monkeypatch.setattr(webapp, "DEFAULT_REPO_NAME", "open-swe")
    assert webapp.get_repo_config_from_jira_mapping("UNKNOWN") == {
        "owner": "langchain-ai",
        "name": "open-swe",
    }


def test_repo_cascade_empty_without_default(monkeypatch) -> None:
    monkeypatch.setattr(webapp, "JIRA_PROJECT_TO_REPO", {})
    monkeypatch.setattr(webapp, "DEFAULT_REPO_NAME", "")
    assert webapp.get_repo_config_from_jira_mapping("UNKNOWN") == {}