mirror of
https://github.com/OpenHands/OpenHands.git
synced 2026-03-22 13:47:19 +08:00
feat(bitbucket): supports cloud and server APIs (#11052)
Co-authored-by: Ray Myers <ray.myers@gmail.com> Co-authored-by: Chris Bagwell <chris@cnpbagwell.com> Co-authored-by: CHANGE <joe.laverty@openhands.dev> Co-authored-by: Joe Laverty <jlav@users.noreply.github.com>
This commit is contained in:
Pierrick Hymbert
@@ -0,0 +1,258 @@
|
||||
"""Tests for BitbucketDCService core: init, headers, get_user, pagination, email."""
|
||||
|
||||
import base64
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from openhands.integrations.bitbucket_data_center.bitbucket_dc_service import (
|
||||
BitbucketDCService,
|
||||
)
|
||||
from openhands.integrations.service_types import AuthenticationError, User
|
||||
from openhands.server.types import AppMode
|
||||
|
||||
# ── init / BASE_URL ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_init_plain_domain():
|
||||
svc = BitbucketDCService(token=SecretStr('tok'), base_domain='host.example.com')
|
||||
assert svc.BASE_URL == 'https://host.example.com/rest/api/1.0'
|
||||
|
||||
|
||||
def test_init_no_domain():
|
||||
svc = BitbucketDCService(token=SecretStr('tok'), base_domain=None)
|
||||
assert svc.BASE_URL == ''
|
||||
|
||||
|
||||
# ── token wrapping ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_token_wraps_simple_token():
|
||||
svc = BitbucketDCService(token=SecretStr('mytoken'))
|
||||
assert svc.token.get_secret_value() == 'x-token-auth:mytoken'
|
||||
|
||||
|
||||
def test_token_preserves_colon_token():
|
||||
svc = BitbucketDCService(token=SecretStr('alice:secret'))
|
||||
assert svc.token.get_secret_value() == 'alice:secret'
|
||||
|
||||
|
||||
# ── user_id derivation ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_user_id_derived_from_username_password_token():
|
||||
svc = BitbucketDCService(token=SecretStr('alice:secret'))
|
||||
assert svc.user_id == 'alice'
|
||||
|
||||
|
||||
def test_user_id_not_derived_from_xtoken_auth_token():
|
||||
svc = BitbucketDCService(token=SecretStr('x-token-auth:mytoken'))
|
||||
assert svc.user_id is None
|
||||
|
||||
|
||||
def test_explicit_user_id_not_overridden():
|
||||
svc = BitbucketDCService(token=SecretStr('alice:secret'), user_id='bob')
|
||||
assert svc.user_id == 'bob'
|
||||
|
||||
|
||||
# ── _get_headers ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_headers_basic_auth():
|
||||
svc = BitbucketDCService(
|
||||
token=SecretStr('user:pass'), base_domain='host.example.com'
|
||||
)
|
||||
headers = await svc._get_headers()
|
||||
expected = 'Basic ' + base64.b64encode(b'user:pass').decode()
|
||||
assert headers['Authorization'] == expected
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_headers_xtoken_auth():
|
||||
svc = BitbucketDCService(
|
||||
token=SecretStr('plaintoken'), base_domain='host.example.com'
|
||||
)
|
||||
# plaintoken has no ':' so it gets wrapped as x-token-auth:plaintoken
|
||||
headers = await svc._get_headers()
|
||||
expected = 'Basic ' + base64.b64encode(b'x-token-auth:plaintoken').decode()
|
||||
assert headers['Authorization'] == expected
|
||||
|
||||
|
||||
# ── get_user ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_with_user_id():
|
||||
svc = BitbucketDCService(
|
||||
token=SecretStr('tok'),
|
||||
base_domain='host.example.com',
|
||||
user_id='jdoe',
|
||||
)
|
||||
mock_response = {
|
||||
'values': [
|
||||
{
|
||||
'id': 5,
|
||||
'slug': 'jdoe',
|
||||
'name': 'jdoe',
|
||||
'displayName': 'J Doe',
|
||||
'emailAddress': 'j@example.com',
|
||||
'avatarUrl': '',
|
||||
}
|
||||
]
|
||||
}
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
user = await svc.get_user()
|
||||
|
||||
assert user.id == '5'
|
||||
assert user.login == 'jdoe'
|
||||
assert user.name == 'J Doe'
|
||||
assert user.email == 'j@example.com'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_without_user_id():
|
||||
# x-token-auth tokens don't have a derivable username, so user_id stays None
|
||||
svc = BitbucketDCService(
|
||||
token=SecretStr('x-token-auth:mytoken'), base_domain='host.example.com'
|
||||
)
|
||||
with patch.object(svc, '_make_request') as mock_req:
|
||||
user = await svc.get_user()
|
||||
mock_req.assert_not_called()
|
||||
|
||||
assert isinstance(user, User)
|
||||
assert user.id == ''
|
||||
assert user.login == ''
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_raises_when_not_found():
|
||||
svc = BitbucketDCService(
|
||||
token=SecretStr('tok'),
|
||||
base_domain='host.example.com',
|
||||
user_id='jdoe',
|
||||
)
|
||||
mock_response = {'values': []}
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
with pytest.raises(AuthenticationError):
|
||||
await svc.get_user()
|
||||
|
||||
|
||||
# ── _resolve_primary_email ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_resolve_primary_email_selects_primary_confirmed():
|
||||
from openhands.integrations.bitbucket_data_center.service.base import (
|
||||
BitbucketDCMixinBase,
|
||||
)
|
||||
|
||||
emails = [
|
||||
{'email': 'secondary@example.com', 'is_primary': False, 'is_confirmed': True},
|
||||
{'email': 'primary@example.com', 'is_primary': True, 'is_confirmed': True},
|
||||
{
|
||||
'email': 'unconfirmed@example.com',
|
||||
'is_primary': False,
|
||||
'is_confirmed': False,
|
||||
},
|
||||
]
|
||||
result = BitbucketDCMixinBase._resolve_primary_email(emails)
|
||||
assert result == 'primary@example.com'
|
||||
|
||||
|
||||
def test_resolve_primary_email_returns_none_when_no_primary():
|
||||
from openhands.integrations.bitbucket_data_center.service.base import (
|
||||
BitbucketDCMixinBase,
|
||||
)
|
||||
|
||||
emails = [
|
||||
{'email': 'a@example.com', 'is_primary': False, 'is_confirmed': True},
|
||||
{'email': 'b@example.com', 'is_primary': False, 'is_confirmed': True},
|
||||
]
|
||||
result = BitbucketDCMixinBase._resolve_primary_email(emails)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_resolve_primary_email_returns_none_when_primary_not_confirmed():
|
||||
from openhands.integrations.bitbucket_data_center.service.base import (
|
||||
BitbucketDCMixinBase,
|
||||
)
|
||||
|
||||
emails = [
|
||||
{'email': 'primary@example.com', 'is_primary': True, 'is_confirmed': False},
|
||||
{'email': 'other@example.com', 'is_primary': False, 'is_confirmed': True},
|
||||
]
|
||||
result = BitbucketDCMixinBase._resolve_primary_email(emails)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_resolve_primary_email_returns_none_for_empty_list():
|
||||
from openhands.integrations.bitbucket_data_center.service.base import (
|
||||
BitbucketDCMixinBase,
|
||||
)
|
||||
|
||||
result = BitbucketDCMixinBase._resolve_primary_email([])
|
||||
assert result is None
|
||||
|
||||
|
||||
# ── get_user_emails ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_emails():
|
||||
svc = BitbucketDCService(token=SecretStr('tok'), base_domain='host.example.com')
|
||||
mock_response = {
|
||||
'values': [
|
||||
{'email': 'primary@example.com', 'is_primary': True, 'is_confirmed': True},
|
||||
{
|
||||
'email': 'secondary@example.com',
|
||||
'is_primary': False,
|
||||
'is_confirmed': True,
|
||||
},
|
||||
]
|
||||
}
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
emails = await svc.get_user_emails()
|
||||
|
||||
assert emails == mock_response['values']
|
||||
|
||||
|
||||
# ── pagination (get_all_repositories iterates projects) ──────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pagination_iterates_projects():
|
||||
svc = BitbucketDCService(token=SecretStr('tok'), base_domain='host.example.com')
|
||||
|
||||
def _repo_dict(key='PROJ', slug='myrepo'):
|
||||
return {'id': 1, 'slug': slug, 'project': {'key': key}, 'public': False}
|
||||
|
||||
async def fake_fetch(url, params, max_items):
|
||||
if '/projects' in url and '/repos' not in url:
|
||||
return [{'key': 'PROJ1'}, {'key': 'PROJ2'}]
|
||||
if 'PROJ1' in url:
|
||||
return [_repo_dict('PROJ1', 'repo1')]
|
||||
if 'PROJ2' in url:
|
||||
return [_repo_dict('PROJ2', 'repo2')]
|
||||
return []
|
||||
|
||||
with patch.object(svc, '_fetch_paginated_data', side_effect=fake_fetch):
|
||||
repos = await svc.get_all_repositories('name', AppMode.SAAS)
|
||||
|
||||
full_names = {r.full_name for r in repos}
|
||||
assert 'PROJ1/repo1' in full_names
|
||||
assert 'PROJ2/repo2' in full_names
|
||||
|
||||
|
||||
# ── verify_access ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verify_access_makes_request():
|
||||
svc = BitbucketDCService(token=SecretStr('tok'), base_domain='host.example.com')
|
||||
with patch.object(svc, '_make_request', return_value=({}, {})) as mock_req:
|
||||
await svc.verify_access()
|
||||
|
||||
mock_req.assert_called_once()
|
||||
call_url = mock_req.call_args[0][0]
|
||||
assert call_url.endswith('/repos')
|
||||
@@ -0,0 +1,139 @@
|
||||
"""Tests for BitbucketDCBranchesMixin: get_paginated_branches, search_branches, get_branches."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from openhands.integrations.bitbucket_data_center.bitbucket_dc_service import (
|
||||
BitbucketDCService,
|
||||
)
|
||||
from openhands.integrations.service_types import Branch, PaginatedBranchesResponse
|
||||
|
||||
|
||||
def make_service():
|
||||
return BitbucketDCService(token=SecretStr('tok'), base_domain='host.example.com')
|
||||
|
||||
|
||||
def _dc_branch(display_id='main', commit='abc123'):
|
||||
return {'displayId': display_id, 'latestCommit': commit}
|
||||
|
||||
|
||||
# ── get_paginated_branches ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_branches_parses_display_id_and_commit():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [
|
||||
_dc_branch('main', 'abc'),
|
||||
_dc_branch('feature/x', 'def'),
|
||||
],
|
||||
'isLastPage': True,
|
||||
'size': 2,
|
||||
}
|
||||
|
||||
with patch.object(
|
||||
svc, '_make_request', return_value=(mock_response, {})
|
||||
) as mock_req:
|
||||
res = await svc.get_paginated_branches('PROJ/myrepo', page=1, per_page=30)
|
||||
|
||||
# Verify the URL uses the DC format
|
||||
call_url = mock_req.call_args[0][0]
|
||||
assert '/projects/PROJ/repos/myrepo/branches' in call_url
|
||||
|
||||
assert isinstance(res, PaginatedBranchesResponse)
|
||||
assert res.branches == [
|
||||
Branch(name='main', commit_sha='abc', protected=False, last_push_date=None),
|
||||
Branch(
|
||||
name='feature/x', commit_sha='def', protected=False, last_push_date=None
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_branches_has_next_page():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [_dc_branch()],
|
||||
'isLastPage': False,
|
||||
'nextPageStart': 30,
|
||||
'size': 100,
|
||||
}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
res = await svc.get_paginated_branches('PROJ/myrepo', page=1, per_page=30)
|
||||
|
||||
assert res.has_next_page is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_branches_last_page():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [_dc_branch()],
|
||||
'isLastPage': True,
|
||||
'size': 1,
|
||||
}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
res = await svc.get_paginated_branches('PROJ/myrepo', page=1, per_page=30)
|
||||
|
||||
assert res.has_next_page is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_branches_total_count():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [_dc_branch()],
|
||||
'isLastPage': True,
|
||||
'size': 42,
|
||||
}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
res = await svc.get_paginated_branches('PROJ/myrepo', page=1, per_page=30)
|
||||
|
||||
assert res.total_count == 42
|
||||
|
||||
|
||||
# ── search_branches ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_branches_uses_filter_text():
|
||||
svc = make_service()
|
||||
mock_response = {'values': [_dc_branch('feature/my-thing', 'sha1')]}
|
||||
|
||||
with patch.object(
|
||||
svc, '_make_request', return_value=(mock_response, {})
|
||||
) as mock_req:
|
||||
branches = await svc.search_branches(
|
||||
'PROJ/myrepo', query='my-thing', per_page=15
|
||||
)
|
||||
|
||||
call_url, call_params = mock_req.call_args[0]
|
||||
assert 'filterText' in call_params
|
||||
assert call_params['filterText'] == 'my-thing'
|
||||
assert 'q' not in call_params
|
||||
assert len(branches) == 1
|
||||
assert branches[0].name == 'feature/my-thing'
|
||||
|
||||
|
||||
# ── get_branches (all pages via _fetch_paginated_data) ───────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_branches_returns_all_pages():
|
||||
svc = make_service()
|
||||
|
||||
async def fake_fetch(url, params, max_items):
|
||||
return [_dc_branch('main', 'a'), _dc_branch('dev', 'b')]
|
||||
|
||||
with patch.object(svc, '_fetch_paginated_data', side_effect=fake_fetch):
|
||||
branches = await svc.get_branches('PROJ/myrepo')
|
||||
|
||||
assert len(branches) == 2
|
||||
assert branches[0].name == 'main'
|
||||
assert branches[1].name == 'dev'
|
||||
@@ -0,0 +1,138 @@
|
||||
"""Tests for BitbucketDCPRsMixin: create_pr, get_pr_details, is_pr_open."""
|
||||
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from openhands.integrations.bitbucket_data_center.bitbucket_dc_service import (
|
||||
BitbucketDCService,
|
||||
)
|
||||
|
||||
|
||||
def make_service():
|
||||
return BitbucketDCService(token=SecretStr('tok'), base_domain='host.example.com')
|
||||
|
||||
|
||||
# ── create_pr ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_pr_payload_structure():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'id': 1,
|
||||
'links': {'self': [{'href': 'https://host.example.com/pr/1'}]},
|
||||
}
|
||||
|
||||
with patch.object(
|
||||
svc, '_make_request', return_value=(mock_response, {})
|
||||
) as mock_req:
|
||||
await svc.create_pr('PROJ/myrepo', 'feature', 'main', 'My PR')
|
||||
|
||||
# The payload is passed as the 'params' positional arg
|
||||
payload = mock_req.call_args[1].get('params') or mock_req.call_args[0][1]
|
||||
assert payload['fromRef']['id'] == 'refs/heads/feature'
|
||||
assert payload['toRef']['id'] == 'refs/heads/main'
|
||||
assert payload['fromRef']['repository']['slug'] == 'myrepo'
|
||||
assert payload['fromRef']['repository']['project']['key'] == 'PROJ'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_pr_returns_href():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'id': 5,
|
||||
'links': {'self': [{'href': 'https://host.example.com/pr/5'}]},
|
||||
}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
url = await svc.create_pr('PROJ/myrepo', 'feature', 'main', 'My PR')
|
||||
|
||||
assert url == 'https://host.example.com/pr/5'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_pr_html_link_dict():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'id': 5,
|
||||
'links': {'html': {'href': 'https://host.example.com/pr/5/html'}},
|
||||
}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
url = await svc.create_pr('PROJ/myrepo', 'feature', 'main', 'My PR')
|
||||
|
||||
assert url == 'https://host.example.com/pr/5/html'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_pr_no_link_returns_empty_string():
|
||||
svc = make_service()
|
||||
mock_response = {'id': 5, 'links': {}}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(mock_response, {})):
|
||||
url = await svc.create_pr('PROJ/myrepo', 'feature', 'main', 'My PR')
|
||||
|
||||
assert url == ''
|
||||
|
||||
|
||||
# ── get_pr_details ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pr_details_returns_raw_data():
|
||||
svc = make_service()
|
||||
mock_data = {'id': 3, 'state': 'OPEN', 'title': 'A PR'}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(mock_data, {})):
|
||||
result = await svc.get_pr_details('PROJ/myrepo', 3)
|
||||
|
||||
assert result == mock_data
|
||||
|
||||
|
||||
# ── is_pr_open ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_is_pr_open_returns_true():
|
||||
svc = make_service()
|
||||
|
||||
with patch.object(
|
||||
svc, 'get_pr_details', new=AsyncMock(return_value={'state': 'OPEN'})
|
||||
):
|
||||
assert await svc.is_pr_open('PROJ/myrepo', 1) is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_is_pr_open_returns_false_for_merged():
|
||||
svc = make_service()
|
||||
|
||||
with patch.object(
|
||||
svc, 'get_pr_details', new=AsyncMock(return_value={'state': 'MERGED'})
|
||||
):
|
||||
assert await svc.is_pr_open('PROJ/myrepo', 1) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_is_pr_open_returns_false_for_declined():
|
||||
svc = make_service()
|
||||
|
||||
with patch.object(
|
||||
svc, 'get_pr_details', new=AsyncMock(return_value={'state': 'DECLINED'})
|
||||
):
|
||||
assert await svc.is_pr_open('PROJ/myrepo', 1) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_is_pr_open_returns_true_on_exception():
|
||||
"""Current implementation catches all exceptions and returns True."""
|
||||
svc = make_service()
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'get_pr_details',
|
||||
new=AsyncMock(side_effect=Exception('Some error')),
|
||||
):
|
||||
result = await svc.is_pr_open('PROJ/myrepo', 999)
|
||||
assert result is True
|
||||
@@ -0,0 +1,355 @@
|
||||
"""Tests for BitbucketDCReposMixin: URL parsing, get_paginated_repos, get_all_repositories."""
|
||||
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from openhands.integrations.bitbucket_data_center.bitbucket_dc_service import (
|
||||
BitbucketDCService,
|
||||
)
|
||||
from openhands.server.types import AppMode
|
||||
|
||||
|
||||
def make_service():
|
||||
return BitbucketDCService(token=SecretStr('tok'), base_domain='host.example.com')
|
||||
|
||||
|
||||
def _repo_dict(key='PROJ', slug='myrepo', name='My Repository'):
|
||||
return {
|
||||
'id': 1,
|
||||
'slug': slug,
|
||||
'name': name,
|
||||
'project': {'key': key},
|
||||
'public': False,
|
||||
}
|
||||
|
||||
|
||||
# ── search_repositories URL parsing ──────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_repositories_projects_url():
|
||||
svc = make_service()
|
||||
query = 'https://host.example.com/projects/PROJ/repos/myrepo'
|
||||
|
||||
mock_repo_data = _repo_dict('PROJ', 'myrepo')
|
||||
mock_response = {'id': 1, **mock_repo_data}
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
side_effect=[
|
||||
(mock_response, {}),
|
||||
(mock_default_branch, {}),
|
||||
],
|
||||
):
|
||||
repos = await svc.search_repositories(
|
||||
query, 25, 'name', 'asc', True, AppMode.SAAS
|
||||
)
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/myrepo'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_repositories_projects_url_with_extra_segments():
|
||||
svc = make_service()
|
||||
# URL with extra segments after repo name
|
||||
query = 'https://host.example.com/projects/PROJ/repos/myrepo/browse/src/main.py'
|
||||
|
||||
mock_repo_data = _repo_dict('PROJ', 'myrepo')
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
side_effect=[
|
||||
(mock_repo_data, {}),
|
||||
(mock_default_branch, {}),
|
||||
],
|
||||
):
|
||||
repos = await svc.search_repositories(
|
||||
query, 25, 'name', 'asc', True, AppMode.SAAS
|
||||
)
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/myrepo'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_repositories_invalid_url():
|
||||
svc = make_service()
|
||||
with patch.object(svc, '_make_request') as mock_req:
|
||||
repos = await svc.search_repositories(
|
||||
'not-a-valid-url', 25, 'name', 'asc', True, AppMode.SAAS
|
||||
)
|
||||
assert repos == []
|
||||
mock_req.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_repositories_insufficient_path_segments():
|
||||
svc = make_service()
|
||||
# URL with only one path segment (just a project, no repo)
|
||||
with patch.object(svc, '_make_request') as mock_req:
|
||||
repos = await svc.search_repositories(
|
||||
'https://host.example.com/projects/PROJ',
|
||||
25,
|
||||
'name',
|
||||
'asc',
|
||||
True,
|
||||
AppMode.SAAS,
|
||||
)
|
||||
assert repos == []
|
||||
mock_req.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_repositories_slash_query():
|
||||
svc = make_service()
|
||||
query = 'PROJ/myrepo'
|
||||
|
||||
mock_repo = _repo_dict('PROJ', slug='myrepo', name='My Repository')
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_fetch_paginated_data',
|
||||
new=AsyncMock(return_value=[mock_repo]),
|
||||
) as mock_fetch:
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
new=AsyncMock(return_value=(mock_default_branch, {})),
|
||||
):
|
||||
repos = await svc.search_repositories(
|
||||
query, 25, 'name', 'asc', False, AppMode.SAAS
|
||||
)
|
||||
|
||||
mock_fetch.assert_called_once_with(
|
||||
'https://host.example.com/rest/api/1.0/projects/PROJ/repos',
|
||||
{'limit': 25},
|
||||
1000,
|
||||
)
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/myrepo'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_repositories_slash_query_filters_by_name():
|
||||
"""Filter matches the human-readable name when slug doesn't match."""
|
||||
svc = make_service()
|
||||
matching = _repo_dict('PROJ', slug='proj-alpha', name='My Repository')
|
||||
non_matching = _repo_dict('PROJ', slug='proj-beta', name='Other Repo')
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_fetch_paginated_data',
|
||||
new=AsyncMock(return_value=[matching, non_matching]),
|
||||
):
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
new=AsyncMock(return_value=(mock_default_branch, {})),
|
||||
):
|
||||
repos = await svc.search_repositories(
|
||||
'PROJ/my repository', 25, 'name', 'asc', False, AppMode.SAAS
|
||||
)
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/proj-alpha'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_repositories_slash_query_filters_by_slug():
|
||||
"""Filter matches the slug when the human-readable name doesn't match."""
|
||||
svc = make_service()
|
||||
matching = _repo_dict('PROJ', slug='my-repo', name='My Repository')
|
||||
non_matching = _repo_dict('PROJ', slug='other-repo', name='Other Repository')
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_fetch_paginated_data',
|
||||
new=AsyncMock(return_value=[matching, non_matching]),
|
||||
):
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
new=AsyncMock(return_value=(mock_default_branch, {})),
|
||||
):
|
||||
repos = await svc.search_repositories(
|
||||
'PROJ/my-repo', 25, 'name', 'asc', False, AppMode.SAAS
|
||||
)
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/my-repo'
|
||||
|
||||
|
||||
# ── get_paginated_repos ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_repos_parses_values():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [_repo_dict()],
|
||||
'isLastPage': True,
|
||||
}
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
side_effect=[(mock_response, {}), (mock_default_branch, {})],
|
||||
):
|
||||
repos = await svc.get_paginated_repos(1, 25, 'name', 'PROJ')
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/myrepo'
|
||||
assert repos[0].link_header == ''
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_repos_has_next_page():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [_repo_dict()],
|
||||
'isLastPage': False,
|
||||
'nextPageStart': 25,
|
||||
}
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
side_effect=[(mock_response, {}), (mock_default_branch, {})],
|
||||
):
|
||||
repos = await svc.get_paginated_repos(1, 25, 'name', 'PROJ')
|
||||
|
||||
assert len(repos) == 1
|
||||
assert 'rel="next"' in repos[0].link_header
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_repos_last_page():
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [_repo_dict()],
|
||||
'isLastPage': True,
|
||||
}
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
side_effect=[(mock_response, {}), (mock_default_branch, {})],
|
||||
):
|
||||
repos = await svc.get_paginated_repos(1, 25, 'name', 'PROJ')
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].link_header == ''
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_repos_filters_by_slug():
|
||||
"""Query matches slug when name doesn't contain the search term."""
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [
|
||||
_repo_dict('PROJ', slug='my-repo', name='My Repository'),
|
||||
_repo_dict('PROJ', slug='other-repo', name='Other Repository'),
|
||||
],
|
||||
'isLastPage': True,
|
||||
}
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
side_effect=[(mock_response, {}), (mock_default_branch, {})],
|
||||
):
|
||||
repos = await svc.get_paginated_repos(1, 25, 'name', 'PROJ', query='my-repo')
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/my-repo'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_paginated_repos_filters_by_name():
|
||||
"""Query matches human-readable name when slug doesn't contain the search term."""
|
||||
svc = make_service()
|
||||
mock_response = {
|
||||
'values': [
|
||||
_repo_dict('PROJ', slug='proj-alpha', name='My Repository'),
|
||||
_repo_dict('PROJ', slug='proj-beta', name='Other Repository'),
|
||||
],
|
||||
'isLastPage': True,
|
||||
}
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
|
||||
with patch.object(
|
||||
svc,
|
||||
'_make_request',
|
||||
side_effect=[(mock_response, {}), (mock_default_branch, {})],
|
||||
):
|
||||
repos = await svc.get_paginated_repos(
|
||||
1, 25, 'name', 'PROJ', query='my repository'
|
||||
)
|
||||
|
||||
assert len(repos) == 1
|
||||
assert repos[0].full_name == 'PROJ/proj-alpha'
|
||||
|
||||
|
||||
# ── get_all_repositories ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_all_repositories_iterates_projects():
|
||||
svc = make_service()
|
||||
|
||||
async def fake_fetch(url, params, max_items):
|
||||
if '/projects' in url and '/repos' not in url:
|
||||
return [{'key': 'PROJ1'}, {'key': 'PROJ2'}]
|
||||
if 'PROJ1' in url:
|
||||
return [_repo_dict('PROJ1', 'repo1')]
|
||||
if 'PROJ2' in url:
|
||||
return [_repo_dict('PROJ2', 'repo2')]
|
||||
return []
|
||||
|
||||
mock_default_branch = {'displayId': 'main'}
|
||||
with patch.object(svc, '_fetch_paginated_data', side_effect=fake_fetch):
|
||||
with patch.object(svc, '_make_request', return_value=(mock_default_branch, {})):
|
||||
repos = await svc.get_all_repositories('name', AppMode.SAAS)
|
||||
|
||||
full_names = {r.full_name for r in repos}
|
||||
assert 'PROJ1/repo1' in full_names
|
||||
assert 'PROJ2/repo2' in full_names
|
||||
|
||||
|
||||
# ── get_installations ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_installations_returns_project_keys():
|
||||
svc = make_service()
|
||||
|
||||
async def fake_fetch(url, params, max_items):
|
||||
return [{'key': 'PROJ1'}, {'key': 'PROJ2'}, {'name': 'no-key'}]
|
||||
|
||||
with patch.object(svc, '_fetch_paginated_data', side_effect=fake_fetch):
|
||||
keys = await svc.get_installations()
|
||||
|
||||
assert keys == ['PROJ1', 'PROJ2']
|
||||
|
||||
|
||||
# ── helper ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def _make_parsed_repo(svc, repo_dict):
|
||||
"""Helper to create a parsed Repository from a repo dict (with mocked default branch)."""
|
||||
with patch.object(svc, '_make_request', return_value=({'displayId': 'main'}, {})):
|
||||
return await svc._parse_repository(repo_dict)
|
||||
@@ -0,0 +1,179 @@
|
||||
"""Tests for BitbucketDCResolverMixin: get_pr_title_and_body, get_pr_comments, _process_raw_comments."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from openhands.integrations.bitbucket_data_center.bitbucket_dc_service import (
|
||||
BitbucketDCService,
|
||||
)
|
||||
from openhands.integrations.service_types import Comment
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def svc():
|
||||
return BitbucketDCService(
|
||||
token=SecretStr('user:pass'), base_domain='host.example.com'
|
||||
)
|
||||
|
||||
|
||||
# ── get_pr_title_and_body ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pr_title_and_body(svc):
|
||||
mock_response = {'title': 'Fix the bug', 'description': 'Detailed description'}
|
||||
with patch.object(
|
||||
svc, '_make_request', return_value=(mock_response, {})
|
||||
) as mock_req:
|
||||
title, body = await svc.get_pr_title_and_body('PROJ', 'myrepo', 42)
|
||||
|
||||
assert title == 'Fix the bug'
|
||||
assert body == 'Detailed description'
|
||||
called_url = mock_req.call_args[0][0]
|
||||
assert '/projects/PROJ/repos/myrepo/pull-requests/42' in called_url
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pr_title_and_body_missing_fields(svc):
|
||||
with patch.object(svc, '_make_request', return_value=({}, {})):
|
||||
title, body = await svc.get_pr_title_and_body('PROJ', 'myrepo', 1)
|
||||
|
||||
assert title == ''
|
||||
assert body == ''
|
||||
|
||||
|
||||
# ── get_pr_comments ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pr_comments_returns_comments(svc):
|
||||
activities = {
|
||||
'values': [
|
||||
{
|
||||
'action': 'COMMENTED',
|
||||
'comment': {
|
||||
'id': 10,
|
||||
'text': 'Looks good!',
|
||||
'author': {'slug': 'alice', 'name': 'Alice'},
|
||||
'createdDate': 1_700_000_000_000,
|
||||
'updatedDate': 1_700_000_000_000,
|
||||
},
|
||||
},
|
||||
{
|
||||
'action': 'APPROVED', # should be ignored
|
||||
'comment': {},
|
||||
},
|
||||
{
|
||||
'action': 'COMMENTED',
|
||||
'comment': {
|
||||
'id': 11,
|
||||
'text': 'Please fix tests',
|
||||
'author': {'slug': 'bob', 'name': 'Bob'},
|
||||
'createdDate': 1_700_000_001_000,
|
||||
'updatedDate': 1_700_000_001_000,
|
||||
},
|
||||
},
|
||||
],
|
||||
'isLastPage': True,
|
||||
}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(activities, {})):
|
||||
comments = await svc.get_pr_comments('PROJ', 'myrepo', 42, max_comments=10)
|
||||
|
||||
assert len(comments) == 2
|
||||
assert all(isinstance(c, Comment) for c in comments)
|
||||
assert comments[0].author == 'alice'
|
||||
assert comments[0].body == 'Looks good!'
|
||||
assert comments[1].author == 'bob'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pr_comments_respects_max(svc):
|
||||
activities = {
|
||||
'values': [
|
||||
{
|
||||
'action': 'COMMENTED',
|
||||
'comment': {
|
||||
'id': i,
|
||||
'text': f'comment {i}',
|
||||
'author': {'slug': f'user{i}'},
|
||||
'createdDate': 1_700_000_000_000 + i * 1000,
|
||||
'updatedDate': 1_700_000_000_000 + i * 1000,
|
||||
},
|
||||
}
|
||||
for i in range(10)
|
||||
],
|
||||
'isLastPage': True,
|
||||
}
|
||||
|
||||
with patch.object(svc, '_make_request', return_value=(activities, {})):
|
||||
comments = await svc.get_pr_comments('PROJ', 'myrepo', 1, max_comments=3)
|
||||
|
||||
assert len(comments) == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pr_comments_empty(svc):
|
||||
with patch.object(
|
||||
svc, '_make_request', return_value=({'values': [], 'isLastPage': True}, {})
|
||||
):
|
||||
comments = await svc.get_pr_comments('PROJ', 'myrepo', 1)
|
||||
|
||||
assert comments == []
|
||||
|
||||
|
||||
# ── _process_raw_comments ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_process_raw_comments_sorts_by_date(svc):
|
||||
raw = [
|
||||
{
|
||||
'id': 2,
|
||||
'text': 'second',
|
||||
'author': {'slug': 'bob'},
|
||||
'createdDate': 1_700_000_002_000,
|
||||
'updatedDate': 1_700_000_002_000,
|
||||
},
|
||||
{
|
||||
'id': 1,
|
||||
'text': 'first',
|
||||
'author': {'slug': 'alice'},
|
||||
'createdDate': 1_700_000_001_000,
|
||||
'updatedDate': 1_700_000_001_000,
|
||||
},
|
||||
]
|
||||
comments = svc._process_raw_comments(raw, max_comments=10)
|
||||
assert comments[0].id == '1'
|
||||
assert comments[1].id == '2'
|
||||
|
||||
|
||||
def test_process_raw_comments_missing_timestamps(svc):
|
||||
raw = [{'id': 5, 'text': 'no dates', 'author': {'slug': 'eve'}}]
|
||||
comments = svc._process_raw_comments(raw)
|
||||
assert len(comments) == 1
|
||||
assert comments[0].id == '5'
|
||||
|
||||
|
||||
# ── MRO check ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_mro_includes_resolver_mixin_and_base_git_service():
|
||||
from openhands.integrations.bitbucket_data_center.service.resolver import (
|
||||
BitbucketDCResolverMixin,
|
||||
)
|
||||
from openhands.integrations.service_types import BaseGitService
|
||||
|
||||
mro_names = [cls.__name__ for cls in BitbucketDCService.__mro__]
|
||||
assert 'BitbucketDCResolverMixin' in mro_names
|
||||
assert 'BaseGitService' in mro_names
|
||||
|
||||
# Resolver mixin should appear before BaseGitService
|
||||
assert mro_names.index('BitbucketDCResolverMixin') < mro_names.index(
|
||||
'BaseGitService'
|
||||
)
|
||||
|
||||
# Verify instances
|
||||
assert issubclass(BitbucketDCService, BitbucketDCResolverMixin)
|
||||
assert issubclass(BitbucketDCService, BaseGitService)
|
||||
@@ -0,0 +1,357 @@
|
||||
import base64
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from openhands.core.config import LLMConfig
|
||||
from openhands.integrations.service_types import ProviderType
|
||||
from openhands.resolver.interfaces.bitbucket_data_center import (
|
||||
BitbucketDCIssueHandler,
|
||||
BitbucketDCPRHandler,
|
||||
)
|
||||
from openhands.resolver.interfaces.issue_definitions import (
|
||||
ServiceContextIssue,
|
||||
ServiceContextPR,
|
||||
)
|
||||
from openhands.resolver.issue_handler_factory import IssueHandlerFactory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def handler():
|
||||
return BitbucketDCIssueHandler(
|
||||
owner='PROJ',
|
||||
repo='my-repo',
|
||||
token='user:secret',
|
||||
base_domain='bitbucket.example.com',
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def llm_config():
|
||||
return LLMConfig(model='test-model', api_key=SecretStr('test-key'))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# URL / attribute tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_init_sets_correct_urls(handler):
|
||||
assert handler.base_url == 'https://bitbucket.example.com/rest/api/1.0'
|
||||
assert (
|
||||
handler.download_url
|
||||
== 'https://bitbucket.example.com/rest/api/latest/projects/PROJ/repos/my-repo/archive?format=zip'
|
||||
)
|
||||
assert handler.clone_url == 'https://bitbucket.example.com/scm/proj/my-repo.git'
|
||||
|
||||
|
||||
def test_get_headers_returns_basic_auth(handler):
|
||||
expected = base64.b64encode(b'user:secret').decode()
|
||||
headers = handler.get_headers()
|
||||
assert headers['Authorization'] == f'Basic {expected}'
|
||||
assert headers['Accept'] == 'application/json'
|
||||
|
||||
|
||||
def test_get_headers_bare_token_with_username():
|
||||
h = BitbucketDCIssueHandler(
|
||||
owner='PROJ',
|
||||
repo='my-repo',
|
||||
token='mytoken',
|
||||
username='myuser',
|
||||
base_domain='dc.example.com',
|
||||
)
|
||||
expected = base64.b64encode(b'myuser:mytoken').decode()
|
||||
assert h.headers['Authorization'] == f'Basic {expected}'
|
||||
|
||||
|
||||
def test_get_repo_url(handler):
|
||||
assert (
|
||||
handler.get_repo_url()
|
||||
== 'https://bitbucket.example.com/projects/PROJ/repos/my-repo'
|
||||
)
|
||||
|
||||
|
||||
def test_get_issue_url_returns_pr_url(handler):
|
||||
assert (
|
||||
handler.get_issue_url(42)
|
||||
== 'https://bitbucket.example.com/projects/PROJ/repos/my-repo/pull-requests/42'
|
||||
)
|
||||
|
||||
|
||||
def test_get_branch_url(handler):
|
||||
assert (
|
||||
handler.get_branch_url('feature/x')
|
||||
== 'https://bitbucket.example.com/projects/PROJ/repos/my-repo/browse?at=refs/heads/feature/x'
|
||||
)
|
||||
|
||||
|
||||
def test_get_authorize_url_with_colon_token(handler):
|
||||
url = handler.get_authorize_url()
|
||||
assert url == 'https://user:secret@bitbucket.example.com/'
|
||||
|
||||
|
||||
def test_get_authorize_url_with_username_and_bare_token():
|
||||
h = BitbucketDCIssueHandler(
|
||||
owner='PROJ',
|
||||
repo='my-repo',
|
||||
token='baretoken',
|
||||
username='john',
|
||||
base_domain='dc.example.com',
|
||||
)
|
||||
assert h.get_authorize_url() == 'https://john:baretoken@dc.example.com/'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_compare_url (requires get_default_branch_name)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_compare_url(handler):
|
||||
with patch.object(handler, 'get_default_branch_name', return_value='main'):
|
||||
url = handler.get_compare_url('feature/fix')
|
||||
assert url == (
|
||||
'https://bitbucket.example.com/projects/PROJ/repos/my-repo/compare/commits'
|
||||
'?sourceBranch=refs/heads/feature/fix&targetBranch=refs/heads/main'
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# API methods (mock httpx)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_issue_fetches_pr_endpoint(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
'id': 7,
|
||||
'title': 'Fix the thing',
|
||||
'description': 'Some body',
|
||||
'fromRef': {'displayId': 'feature/fix'},
|
||||
'toRef': {'displayId': 'main'},
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.get = AsyncMock(return_value=mock_response)
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=None)
|
||||
|
||||
with patch('httpx.AsyncClient', return_value=mock_client):
|
||||
issue = await handler.get_issue(7)
|
||||
|
||||
expected_url = (
|
||||
'https://bitbucket.example.com/rest/api/1.0'
|
||||
'/projects/PROJ/repos/my-repo/pull-requests/7'
|
||||
)
|
||||
mock_client.get.assert_called_once_with(expected_url, headers=handler.headers)
|
||||
assert issue.number == 7
|
||||
assert issue.title == 'Fix the thing'
|
||||
assert issue.body == 'Some body'
|
||||
assert issue.head_branch == 'feature/fix'
|
||||
assert issue.base_branch == 'main'
|
||||
|
||||
|
||||
def test_create_pr_uses_from_to_ref(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
'id': 3,
|
||||
'links': {
|
||||
'self': [
|
||||
{
|
||||
'href': 'https://bitbucket.example.com/projects/PROJ/repos/my-repo/pull-requests/3'
|
||||
}
|
||||
]
|
||||
},
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.post', return_value=mock_response) as mock_post:
|
||||
url = handler.create_pr('Title', 'Body', 'feature/src', 'main')
|
||||
|
||||
_, kwargs = mock_post.call_args
|
||||
payload = kwargs['json']
|
||||
assert payload['fromRef']['id'] == 'refs/heads/feature/src'
|
||||
assert payload['toRef']['id'] == 'refs/heads/main'
|
||||
assert payload['fromRef']['repository']['slug'] == 'my-repo'
|
||||
assert payload['fromRef']['repository']['project']['key'] == 'PROJ'
|
||||
assert (
|
||||
url
|
||||
== 'https://bitbucket.example.com/projects/PROJ/repos/my-repo/pull-requests/3'
|
||||
)
|
||||
|
||||
|
||||
def test_create_pull_request_returns_html_url_and_number(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
'id': 5,
|
||||
'links': {
|
||||
'self': [
|
||||
{
|
||||
'href': 'https://bitbucket.example.com/projects/PROJ/repos/my-repo/pull-requests/5'
|
||||
}
|
||||
]
|
||||
},
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.post', return_value=mock_response):
|
||||
result = handler.create_pull_request(
|
||||
{
|
||||
'title': 'My PR',
|
||||
'description': 'desc',
|
||||
'source_branch': 'feature/x',
|
||||
'target_branch': 'main',
|
||||
}
|
||||
)
|
||||
|
||||
assert result['number'] == 5
|
||||
assert result['html_url'] == (
|
||||
'https://bitbucket.example.com/projects/PROJ/repos/my-repo/pull-requests/5'
|
||||
)
|
||||
|
||||
|
||||
def test_send_comment_msg_uses_text_field(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.post', return_value=mock_response) as mock_post:
|
||||
handler.send_comment_msg(7, 'Hello from OpenHands')
|
||||
|
||||
_, kwargs = mock_post.call_args
|
||||
assert kwargs['json'] == {'text': 'Hello from OpenHands'}
|
||||
|
||||
|
||||
def test_reply_to_comment_posts_with_parent_id(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.post', return_value=mock_response) as mock_post:
|
||||
handler.reply_to_comment(7, '42', 'My reply')
|
||||
|
||||
_, kwargs = mock_post.call_args
|
||||
assert kwargs['json']['text'] == 'My reply'
|
||||
assert kwargs['json']['parent'] == {'id': 42}
|
||||
|
||||
|
||||
def test_branch_exists_true(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
'values': [{'displayId': 'feature/fix', 'id': 'refs/heads/feature/fix'}]
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.get', return_value=mock_response):
|
||||
assert handler.branch_exists('feature/fix') is True
|
||||
|
||||
|
||||
def test_branch_exists_false(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {'values': []}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.get', return_value=mock_response):
|
||||
assert handler.branch_exists('nonexistent') is False
|
||||
|
||||
|
||||
def test_branch_exists_no_match(handler):
|
||||
mock_response = MagicMock()
|
||||
# filterText returns similar but not exact match
|
||||
mock_response.json.return_value = {
|
||||
'values': [{'displayId': 'feature/fix-extended'}]
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.get', return_value=mock_response):
|
||||
assert handler.branch_exists('feature/fix') is False
|
||||
|
||||
|
||||
def test_get_default_branch_name_reads_display_id(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
'defaultBranch': {'displayId': 'main', 'id': 'refs/heads/main'}
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.get', return_value=mock_response):
|
||||
assert handler.get_default_branch_name() == 'main'
|
||||
|
||||
|
||||
def test_get_default_branch_name_strips_refs_heads_prefix(handler):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
'defaultBranch': {'displayId': 'refs/heads/develop'}
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch('httpx.get', return_value=mock_response):
|
||||
assert handler.get_default_branch_name() == 'develop'
|
||||
|
||||
|
||||
def test_get_default_branch_name_fallback_to_master(handler):
|
||||
import httpx as httpx_module
|
||||
|
||||
with patch('httpx.get', side_effect=httpx_module.HTTPError('connection refused')):
|
||||
assert handler.get_default_branch_name() == 'master'
|
||||
|
||||
|
||||
def test_get_context_returns_early(handler):
|
||||
"""get_context_from_external_issues_references should return closing_issues without API calls."""
|
||||
closing_issues = ['issue body 1']
|
||||
with patch('httpx.get') as mock_get:
|
||||
result = handler.get_context_from_external_issues_references(
|
||||
closing_issues=closing_issues,
|
||||
closing_issue_numbers=[1],
|
||||
issue_body='some body',
|
||||
review_comments=None,
|
||||
review_threads=[],
|
||||
thread_comments=None,
|
||||
)
|
||||
mock_get.assert_not_called()
|
||||
assert result == ['issue body 1']
|
||||
|
||||
|
||||
def test_download_issues_returns_empty(handler):
|
||||
result = handler.download_issues()
|
||||
assert result == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Factory integration tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_factory_creates_dc_issue_handler(llm_config):
|
||||
factory = IssueHandlerFactory(
|
||||
owner='PROJ',
|
||||
repo='my-repo',
|
||||
token='user:secret',
|
||||
username='user',
|
||||
platform=ProviderType.BITBUCKET_DATA_CENTER,
|
||||
base_domain='bitbucket.example.com',
|
||||
issue_type='issue',
|
||||
llm_config=llm_config,
|
||||
)
|
||||
ctx = factory.create()
|
||||
assert isinstance(ctx, ServiceContextIssue)
|
||||
assert isinstance(ctx._strategy, BitbucketDCIssueHandler)
|
||||
assert ctx._strategy.owner == 'PROJ'
|
||||
assert ctx._strategy.repo == 'my-repo'
|
||||
assert ctx._strategy.base_domain == 'bitbucket.example.com'
|
||||
|
||||
|
||||
def test_factory_creates_dc_pr_handler(llm_config):
|
||||
factory = IssueHandlerFactory(
|
||||
owner='PROJ',
|
||||
repo='my-repo',
|
||||
token='user:secret',
|
||||
username='user',
|
||||
platform=ProviderType.BITBUCKET_DATA_CENTER,
|
||||
base_domain='bitbucket.example.com',
|
||||
issue_type='pr',
|
||||
llm_config=llm_config,
|
||||
)
|
||||
ctx = factory.create()
|
||||
assert isinstance(ctx, ServiceContextPR)
|
||||
assert isinstance(ctx._strategy, BitbucketDCPRHandler)
|
||||
Reference in New Issue
Block a user