mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 22:03:14 +00:00
Adam Moussa
ae1f883b4c
Applies the plan's C5 step: git mv every test per the domain-reorg
move-map (movemap-m50.txt) into tests/{agent,analyzer,auth,dashboard,
github,middleware,models,reviewer,sandbox,slack,tools,webhooks}/, plus
the 13 fork-only placements from the scoping report §2c (Atlassian
webhook tests -> tests/webhooks/, test_atlassian_connect.py and
test_auth_error_leak.py -> tests/auth/, jira/confluence util tests ->
tests/tools/, test_repo_binding_isolation.py -> tests/sandbox/,
bot-identity/autofix tests -> tests/github/).
Path-only move: the only content edits are parents[1] -> parents[2]
fixes in test_e2b_integration.py and test_daytona_integration.py,
required because their __file__-relative ROOT path gained one more
directory level in the move.
Monkeypatch retargets for these files were already completed in C4;
none remained outstanding here.
187 lines
7.4 KiB
Python
187 lines
7.4 KiB
Python
"""Tests for agent.utils.repo and Linear webhook repo override behavior."""
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from agent.utils.repo import extract_repo_from_text
|
|
from agent.utils.slack import extract_channel_description_text
|
|
|
|
|
|
class TestExtractRepoFromText:
|
|
def test_repo_colon_with_org(self) -> None:
|
|
result = extract_repo_from_text("please use repo:my-org/my-repo")
|
|
assert result == {"owner": "my-org", "name": "my-repo"}
|
|
|
|
def test_repo_space_with_org(self) -> None:
|
|
result = extract_repo_from_text("please use repo langchain-ai/langchainjs")
|
|
assert result == {"owner": "langchain-ai", "name": "langchainjs"}
|
|
|
|
def test_repo_colon_name_only_uses_default_owner(self) -> None:
|
|
result = extract_repo_from_text("fix bug in repo:langchainplus")
|
|
assert result == {"owner": "langchain-ai", "name": "langchainplus"}
|
|
|
|
def test_repo_space_name_only_uses_default_owner(self) -> None:
|
|
result = extract_repo_from_text("fix bug in repo open-swe")
|
|
assert result == {"owner": "langchain-ai", "name": "open-swe"}
|
|
|
|
def test_repo_name_only_custom_default_owner(self) -> None:
|
|
result = extract_repo_from_text("repo:my-repo", default_owner="custom-org")
|
|
assert result == {"owner": "custom-org", "name": "my-repo"}
|
|
|
|
def test_github_url(self) -> None:
|
|
result = extract_repo_from_text(
|
|
"check https://github.com/langchain-ai/langgraph-api please"
|
|
)
|
|
assert result == {"owner": "langchain-ai", "name": "langgraph-api"}
|
|
|
|
def test_explicit_repo_beats_github_url(self) -> None:
|
|
result = extract_repo_from_text(
|
|
"see https://github.com/langchain-ai/langgraph-api but use repo:my-org/my-repo"
|
|
)
|
|
assert result == {"owner": "my-org", "name": "my-repo"}
|
|
|
|
def test_no_repo_returns_none(self) -> None:
|
|
result = extract_repo_from_text("please fix the bug")
|
|
assert result is None
|
|
|
|
def test_empty_string_returns_none(self) -> None:
|
|
result = extract_repo_from_text("")
|
|
assert result is None
|
|
|
|
def test_trailing_slash_stripped(self) -> None:
|
|
result = extract_repo_from_text("repo:my-org/my-repo/")
|
|
assert result == {"owner": "my-org", "name": "my-repo"}
|
|
|
|
|
|
class TestExtractChannelDescriptionText:
|
|
def test_combines_topic_and_purpose(self) -> None:
|
|
channel = {
|
|
"topic": {"value": "repo:my-org/my-repo"},
|
|
"purpose": {"value": "Team channel"},
|
|
}
|
|
assert extract_channel_description_text(channel) == "repo:my-org/my-repo\nTeam channel"
|
|
|
|
def test_handles_missing_sections(self) -> None:
|
|
assert extract_channel_description_text({"topic": {"value": "hi"}}) == "hi"
|
|
|
|
def test_empty_for_none(self) -> None:
|
|
assert extract_channel_description_text(None) == ""
|
|
|
|
def test_empty_for_blank_values(self) -> None:
|
|
channel = {"topic": {"value": " "}, "purpose": {"value": ""}}
|
|
assert extract_channel_description_text(channel) == ""
|
|
|
|
def test_repo_token_extractable_from_description(self) -> None:
|
|
channel = {"topic": {"value": "Use repo:langchain-ai/open-swe here"}, "purpose": {}}
|
|
description = extract_channel_description_text(channel)
|
|
assert extract_repo_from_text(description) == {
|
|
"owner": "langchain-ai",
|
|
"name": "open-swe",
|
|
}
|
|
|
|
|
|
class TestLinearWebhookRepoOverride:
|
|
"""Test that the Linear webhook handler checks comment body for repo config first."""
|
|
|
|
@pytest.fixture()
|
|
def _base_payload(self) -> dict:
|
|
return {
|
|
"type": "Comment",
|
|
"action": "create",
|
|
"data": {
|
|
"id": "comment-123",
|
|
"body": "@openswe please fix this repo:custom-org/custom-repo",
|
|
"issue": {
|
|
"id": "issue-456",
|
|
"title": "Test issue",
|
|
},
|
|
"user": {"id": "user-1", "name": "Test User", "email": "test@test.com"},
|
|
},
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_comment_repo_overrides_team_mapping(self, _base_payload: dict) -> None:
|
|
from agent.webhooks.linear_routes import linear_webhook
|
|
|
|
with (
|
|
patch("agent.webhooks.common.verify_linear_signature", return_value=True),
|
|
patch(
|
|
"agent.webhooks.common.fetch_linear_issue_details",
|
|
new_callable=AsyncMock,
|
|
return_value={
|
|
"id": "issue-456",
|
|
"title": "Test issue",
|
|
"identifier": "TEST-1",
|
|
"url": "https://linear.app/test/issue/TEST-1",
|
|
"team": {"id": "t1", "name": "Some Team", "key": "ST"},
|
|
"project": {"id": "p1", "name": "Some Project"},
|
|
"comments": {"nodes": []},
|
|
},
|
|
),
|
|
patch("agent.webhooks.common._is_repo_allowed", return_value=True),
|
|
patch("agent.webhooks.common.BackgroundTasks"),
|
|
):
|
|
mock_request = AsyncMock()
|
|
mock_request.body.return_value = json.dumps(_base_payload).encode()
|
|
mock_request.headers = {"Linear-Signature": "valid"}
|
|
|
|
bg_tasks = AsyncMock()
|
|
result = await linear_webhook(mock_request, bg_tasks)
|
|
|
|
assert result["status"] == "accepted"
|
|
assert "custom-org/custom-repo" in result["message"]
|
|
|
|
call_args = bg_tasks.add_task.call_args
|
|
repo_config = call_args[0][2]
|
|
assert repo_config == {"owner": "custom-org", "name": "custom-repo"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_falls_back_to_team_mapping_when_no_repo_in_comment(self) -> None:
|
|
from agent.webhooks.linear_routes import linear_webhook
|
|
|
|
payload = {
|
|
"type": "Comment",
|
|
"action": "create",
|
|
"data": {
|
|
"id": "comment-123",
|
|
"body": "@openswe please fix this bug",
|
|
"issue": {
|
|
"id": "issue-456",
|
|
"title": "Test issue",
|
|
},
|
|
"user": {"id": "user-1", "name": "Test User", "email": "test@test.com"},
|
|
},
|
|
}
|
|
|
|
with (
|
|
patch("agent.webhooks.common.verify_linear_signature", return_value=True),
|
|
patch(
|
|
"agent.webhooks.common.fetch_linear_issue_details",
|
|
new_callable=AsyncMock,
|
|
return_value={
|
|
"id": "issue-456",
|
|
"title": "Test issue",
|
|
"identifier": "TEST-1",
|
|
"url": "https://linear.app/test/issue/TEST-1",
|
|
"team": {"id": "t1", "name": "Open SWE", "key": "OS"},
|
|
"project": None,
|
|
"comments": {"nodes": []},
|
|
},
|
|
),
|
|
patch("agent.webhooks.common._is_repo_allowed", return_value=True),
|
|
):
|
|
mock_request = AsyncMock()
|
|
mock_request.body.return_value = json.dumps(payload).encode()
|
|
mock_request.headers = {"Linear-Signature": "valid"}
|
|
|
|
bg_tasks = AsyncMock()
|
|
result = await linear_webhook(mock_request, bg_tasks)
|
|
|
|
assert result["status"] == "accepted"
|
|
assert "langchain-ai/open-swe" in result["message"]
|
|
|
|
call_args = bg_tasks.add_task.call_args
|
|
repo_config = call_args[0][2]
|
|
assert repo_config == {"owner": "langchain-ai", "name": "open-swe"}
|