mirror of
https://github.com/All-Hands-AI/OpenHands.git
synced 2026-04-29 03:00:45 -04:00
Compare commits
6 Commits
openhands/
...
fix-bitbuc
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5769c111ab | ||
|
|
18155338f7 | ||
|
|
548b18c625 | ||
|
|
9eee0de458 | ||
|
|
e3da7178a6 | ||
|
|
5b0ed11ca3 |
@@ -11,6 +11,31 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
Mixin for BitBucket repository-related operations
|
||||
"""
|
||||
|
||||
def _get_bitbucket_sort_param(self, sort: str, order: str = 'desc') -> str:
|
||||
"""Map sort parameter to Bitbucket API compatible values with order.
|
||||
|
||||
Args:
|
||||
sort: The sort field ('pushed', 'updated', 'created', 'full_name')
|
||||
order: The sort order ('asc' or 'desc', defaults to 'desc')
|
||||
|
||||
Returns:
|
||||
Bitbucket API compatible sort parameter with order prefix
|
||||
"""
|
||||
# Map sort parameter to Bitbucket API compatible values
|
||||
if sort == 'pushed':
|
||||
bitbucket_sort = 'updated_on' # Bitbucket doesn't support 'pushed'
|
||||
elif sort == 'updated':
|
||||
bitbucket_sort = 'updated_on'
|
||||
elif sort == 'created':
|
||||
bitbucket_sort = 'created_on'
|
||||
elif sort == 'full_name':
|
||||
bitbucket_sort = 'name' # Bitbucket uses 'name' not 'full_name'
|
||||
else:
|
||||
bitbucket_sort = 'updated_on' # Default to most recently updated
|
||||
|
||||
# Apply order - Bitbucket uses '-' prefix for descending order
|
||||
return f'-{bitbucket_sort}' if order.lower() == 'desc' else bitbucket_sort
|
||||
|
||||
async def search_repositories(
|
||||
self, query: str, per_page: int, sort: str, order: str, public: bool
|
||||
) -> list[Repository]:
|
||||
@@ -37,7 +62,7 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
if '/' in query:
|
||||
workspace_slug, repo_query = query.split('/', 1)
|
||||
return await self.get_paginated_repos(
|
||||
1, per_page, sort, workspace_slug, repo_query
|
||||
1, per_page, sort, workspace_slug, repo_query, order
|
||||
)
|
||||
|
||||
all_installations = await self.get_installations()
|
||||
@@ -50,7 +75,7 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
# Get repositories where query matches workspace name
|
||||
try:
|
||||
repos = await self.get_paginated_repos(
|
||||
1, per_page, sort, workspace_slug
|
||||
1, per_page, sort, workspace_slug, None, order
|
||||
)
|
||||
repositories.extend(repos)
|
||||
except Exception:
|
||||
@@ -60,7 +85,7 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
# Get repositories in all workspaces where query matches repo name
|
||||
try:
|
||||
repos = await self.get_paginated_repos(
|
||||
1, per_page, sort, workspace_slug, query
|
||||
1, per_page, sort, workspace_slug, query, order
|
||||
)
|
||||
repositories.extend(repos)
|
||||
except Exception:
|
||||
@@ -97,6 +122,7 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
sort: str,
|
||||
installation_id: str | None,
|
||||
query: str | None = None,
|
||||
order: str = 'desc',
|
||||
) -> list[Repository]:
|
||||
"""Get paginated repositories for a specific workspace.
|
||||
|
||||
@@ -105,6 +131,8 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
per_page: The number of repositories per page
|
||||
sort: The sort field ('pushed', 'updated', 'created', 'full_name')
|
||||
installation_id: The workspace slug to fetch repositories from (as int, will be converted to string)
|
||||
query: Optional query string to filter repositories by name
|
||||
order: The sort order ('asc' or 'desc', defaults to 'desc')
|
||||
|
||||
Returns:
|
||||
A list of Repository objects
|
||||
@@ -116,20 +144,7 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
workspace_slug = installation_id
|
||||
workspace_repos_url = f'{self.BASE_URL}/repositories/{workspace_slug}'
|
||||
|
||||
# Map sort parameter to Bitbucket API compatible values
|
||||
bitbucket_sort = sort
|
||||
if sort == 'pushed':
|
||||
# Bitbucket doesn't support 'pushed', use 'updated_on' instead
|
||||
bitbucket_sort = '-updated_on' # Use negative prefix for descending order
|
||||
elif sort == 'updated':
|
||||
bitbucket_sort = '-updated_on'
|
||||
elif sort == 'created':
|
||||
bitbucket_sort = '-created_on'
|
||||
elif sort == 'full_name':
|
||||
bitbucket_sort = 'name' # Bitbucket uses 'name' not 'full_name'
|
||||
else:
|
||||
# Default to most recently updated first
|
||||
bitbucket_sort = '-updated_on'
|
||||
bitbucket_sort = self._get_bitbucket_sort_param(sort, order)
|
||||
|
||||
params = {
|
||||
'pagelen': per_page,
|
||||
@@ -173,7 +188,7 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
return repositories
|
||||
|
||||
async def get_all_repositories(
|
||||
self, sort: str, app_mode: AppMode
|
||||
self, sort: str, app_mode: AppMode, order: str = 'desc'
|
||||
) -> list[Repository]:
|
||||
"""Get repositories for the authenticated user using workspaces endpoint.
|
||||
|
||||
@@ -198,23 +213,7 @@ class BitBucketReposMixin(BitBucketMixinBase):
|
||||
# Get repositories for this workspace with pagination
|
||||
workspace_repos_url = f'{self.BASE_URL}/repositories/{workspace_slug}'
|
||||
|
||||
# Map sort parameter to Bitbucket API compatible values and ensure descending order
|
||||
# to show most recently changed repos at the top
|
||||
bitbucket_sort = sort
|
||||
if sort == 'pushed':
|
||||
# Bitbucket doesn't support 'pushed', use 'updated_on' instead
|
||||
bitbucket_sort = (
|
||||
'-updated_on' # Use negative prefix for descending order
|
||||
)
|
||||
elif sort == 'updated':
|
||||
bitbucket_sort = '-updated_on'
|
||||
elif sort == 'created':
|
||||
bitbucket_sort = '-created_on'
|
||||
elif sort == 'full_name':
|
||||
bitbucket_sort = 'name' # Bitbucket uses 'name' not 'full_name'
|
||||
else:
|
||||
# Default to most recently updated first
|
||||
bitbucket_sort = '-updated_on'
|
||||
bitbucket_sort = self._get_bitbucket_sort_param(sort, order)
|
||||
|
||||
params = {
|
||||
'pagelen': PER_PAGE,
|
||||
|
||||
@@ -92,6 +92,7 @@ class GitHubReposMixin(GitHubMixinBase):
|
||||
sort: str,
|
||||
installation_id: str | None,
|
||||
query: str | None = None,
|
||||
order: str = 'desc',
|
||||
):
|
||||
params = {'page': str(page), 'per_page': str(per_page)}
|
||||
if installation_id:
|
||||
@@ -101,6 +102,7 @@ class GitHubReposMixin(GitHubMixinBase):
|
||||
else:
|
||||
url = f'{self.BASE_URL}/user/repos'
|
||||
params['sort'] = sort
|
||||
params['direction'] = order # GitHub uses 'direction' for asc/desc
|
||||
response, headers = await self._make_request(url, params)
|
||||
|
||||
next_link: str = headers.get('Link', '')
|
||||
@@ -109,7 +111,7 @@ class GitHubReposMixin(GitHubMixinBase):
|
||||
]
|
||||
|
||||
async def get_all_repositories(
|
||||
self, sort: str, app_mode: AppMode
|
||||
self, sort: str, app_mode: AppMode, order: str = 'desc'
|
||||
) -> list[Repository]:
|
||||
MAX_REPOS = 1000
|
||||
PER_PAGE = 100 # Maximum allowed by GitHub API
|
||||
@@ -141,7 +143,7 @@ class GitHubReposMixin(GitHubMixinBase):
|
||||
all_repos.sort(key=self.parse_pushed_at_date, reverse=True)
|
||||
else:
|
||||
# Original behavior for non-SaaS mode
|
||||
params = {'per_page': str(PER_PAGE), 'sort': sort}
|
||||
params = {'per_page': str(PER_PAGE), 'sort': sort, 'direction': order}
|
||||
url = f'{self.BASE_URL}/user/repos'
|
||||
|
||||
# Fetch user repositories
|
||||
|
||||
@@ -85,7 +85,7 @@ class GitLabReposMixin(GitLabMixinBase):
|
||||
repository = await self.get_repository_details_from_repo_name(repo_path)
|
||||
return [repository]
|
||||
|
||||
return await self.get_paginated_repos(1, per_page, sort, None, query)
|
||||
return await self.get_paginated_repos(1, per_page, sort, None, query, order)
|
||||
|
||||
async def get_paginated_repos(
|
||||
self,
|
||||
@@ -94,6 +94,7 @@ class GitLabReposMixin(GitLabMixinBase):
|
||||
sort: str,
|
||||
installation_id: str | None,
|
||||
query: str | None = None,
|
||||
order: str = 'desc',
|
||||
) -> list[Repository]:
|
||||
url = f'{self.BASE_URL}/projects'
|
||||
order_by = {
|
||||
@@ -107,7 +108,7 @@ class GitLabReposMixin(GitLabMixinBase):
|
||||
'page': str(page),
|
||||
'per_page': str(per_page),
|
||||
'order_by': order_by,
|
||||
'sort': 'desc', # GitLab uses sort for direction (asc/desc)
|
||||
'sort': order, # GitLab uses sort for direction (asc/desc)
|
||||
'membership': True, # Include projects user is a member of
|
||||
}
|
||||
|
||||
@@ -124,7 +125,7 @@ class GitLabReposMixin(GitLabMixinBase):
|
||||
return repos
|
||||
|
||||
async def get_all_repositories(
|
||||
self, sort: str, app_mode: AppMode
|
||||
self, sort: str, app_mode: AppMode, order: str = 'desc'
|
||||
) -> list[Repository]:
|
||||
MAX_REPOS = 1000
|
||||
PER_PAGE = 100 # Maximum allowed by GitLab API
|
||||
@@ -145,7 +146,7 @@ class GitLabReposMixin(GitLabMixinBase):
|
||||
'page': str(page),
|
||||
'per_page': str(PER_PAGE),
|
||||
'order_by': order_by,
|
||||
'sort': 'desc', # GitLab uses sort for direction (asc/desc)
|
||||
'sort': order, # GitLab uses sort for direction (asc/desc)
|
||||
'membership': 1, # Use 1 instead of True
|
||||
}
|
||||
response, headers = await self._make_request(url, params)
|
||||
|
||||
@@ -502,7 +502,7 @@ class GitService(Protocol):
|
||||
...
|
||||
|
||||
async def get_all_repositories(
|
||||
self, sort: str, app_mode: AppMode
|
||||
self, sort: str, app_mode: AppMode, order: str = 'desc'
|
||||
) -> list[Repository]:
|
||||
"""Get repositories for the authenticated user"""
|
||||
...
|
||||
@@ -514,6 +514,7 @@ class GitService(Protocol):
|
||||
sort: str,
|
||||
installation_id: str | None,
|
||||
query: str | None = None,
|
||||
order: str = 'desc',
|
||||
) -> list[Repository]:
|
||||
"""Get a page of repositories for the authenticated user"""
|
||||
...
|
||||
|
||||
@@ -341,13 +341,19 @@ class TestBitbucketProviderDomain(unittest.TestCase):
|
||||
)
|
||||
|
||||
# Verify that run_action was called at least once (for git clone)
|
||||
self.assertTrue(mock_run_action.called)
|
||||
mock_run_action.assert_called()
|
||||
|
||||
# Verify that the domain used was 'bitbucket.org'
|
||||
# Extract the command from the first call to run_action
|
||||
args, _ = mock_run_action.call_args
|
||||
action = args[0]
|
||||
self.assertIn('bitbucket.org', action.command)
|
||||
# Check that at least one call contains 'bitbucket.org' in the action command
|
||||
calls_with_bitbucket = [
|
||||
call_args
|
||||
for call_args in mock_run_action.call_args_list
|
||||
if 'bitbucket.org' in call_args[0][0].command
|
||||
]
|
||||
self.assertTrue(
|
||||
len(calls_with_bitbucket) > 0,
|
||||
"Expected at least one call with 'bitbucket.org' in the command",
|
||||
)
|
||||
|
||||
|
||||
# Provider Token Validation Tests
|
||||
@@ -417,12 +423,10 @@ async def test_check_provider_tokens_with_only_bitbucket():
|
||||
):
|
||||
result = await check_provider_tokens(post_model, None)
|
||||
|
||||
# Verify that validate_provider_token was called only once (for Bitbucket)
|
||||
assert mock_validate.call_count == 1
|
||||
|
||||
# Verify that the token passed to validate_provider_token was the Bitbucket token
|
||||
args, kwargs = mock_validate.call_args
|
||||
assert args[0].get_secret_value() == 'username:app_password'
|
||||
# Verify that validate_provider_token was called only once with the Bitbucket token and host
|
||||
expected_token = provider_tokens[ProviderType.BITBUCKET].token
|
||||
expected_host = provider_tokens[ProviderType.BITBUCKET].host
|
||||
mock_validate.assert_called_once_with(expected_token, expected_host)
|
||||
|
||||
# Verify that no error message was returned
|
||||
assert result == ''
|
||||
@@ -450,13 +454,20 @@ async def test_bitbucket_sort_parameter_mapping():
|
||||
# Verify that the second call used 'updated_on' instead of 'pushed'
|
||||
assert mock_request.call_count == 2
|
||||
|
||||
# Check the second call (repositories call)
|
||||
second_call_args = mock_request.call_args_list[1]
|
||||
url, params = second_call_args[0]
|
||||
# Verify the second call was made with the correct parameters
|
||||
# We can't use assert_called_with directly because we need to check partial URL and params
|
||||
# But we can verify the call structure more explicitly
|
||||
calls = mock_request.call_args_list
|
||||
assert len(calls) == 2, f'Expected 2 calls, got {len(calls)}'
|
||||
|
||||
# Verify the sort parameter was mapped correctly (with descending order)
|
||||
assert params['sort'] == '-updated_on'
|
||||
assert 'repositories/test-workspace' in url
|
||||
# Check the second call (repositories call) contains the mapped sort parameter
|
||||
second_call_url, second_call_params = calls[1][0]
|
||||
assert second_call_params['sort'] == '-updated_on', (
|
||||
f"Expected sort parameter '-updated_on', got {second_call_params.get('sort')}"
|
||||
)
|
||||
assert 'repositories/test-workspace' in second_call_url, (
|
||||
f"Expected URL to contain 'repositories/test-workspace', got {second_call_url}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -741,23 +752,31 @@ def test_initialize_repository_for_runtime_with_bitbucket_token(
|
||||
|
||||
# Verify that call_async_from_sync was called with the correct arguments
|
||||
mock_call_async_from_sync.assert_called_once()
|
||||
args, kwargs = mock_call_async_from_sync.call_args
|
||||
|
||||
# Check that the function called was clone_or_init_repo
|
||||
assert args[0] == mock_runtime.clone_or_init_repo
|
||||
# Extract call arguments to verify the provider tokens were set correctly
|
||||
call_args = mock_call_async_from_sync.call_args
|
||||
assert call_args[0][0] == mock_runtime.clone_or_init_repo, (
|
||||
'Expected first argument to be clone_or_init_repo method'
|
||||
)
|
||||
|
||||
# Check that provider tokens were passed correctly
|
||||
provider_tokens = args[2] # Third argument is immutable_provider_tokens
|
||||
assert provider_tokens is not None
|
||||
assert ProviderType.BITBUCKET in provider_tokens
|
||||
provider_tokens = call_args[0][2] # Third argument is immutable_provider_tokens
|
||||
assert provider_tokens is not None, 'Provider tokens should not be None'
|
||||
assert ProviderType.BITBUCKET in provider_tokens, (
|
||||
'BITBUCKET provider should be in provider_tokens'
|
||||
)
|
||||
assert (
|
||||
provider_tokens[ProviderType.BITBUCKET].token.get_secret_value()
|
||||
== 'username:app_password'
|
||||
), (
|
||||
f"Expected BITBUCKET token to be 'username:app_password', got {provider_tokens[ProviderType.BITBUCKET].token.get_secret_value()}"
|
||||
)
|
||||
|
||||
# Check that the repository was passed correctly
|
||||
assert args[3] == 'all-hands-ai/test-repo' # selected_repository
|
||||
assert args[4] is None # selected_branch
|
||||
assert call_args[0][3] == 'all-hands-ai/test-repo', (
|
||||
"Expected selected_repository to be 'all-hands-ai/test-repo'"
|
||||
)
|
||||
assert call_args[0][4] is None, 'Expected selected_branch to be None'
|
||||
|
||||
|
||||
@patch('openhands.core.setup.call_async_from_sync')
|
||||
@@ -797,29 +816,41 @@ def test_initialize_repository_for_runtime_with_multiple_tokens(
|
||||
|
||||
# Verify that call_async_from_sync was called
|
||||
mock_call_async_from_sync.assert_called_once()
|
||||
args, kwargs = mock_call_async_from_sync.call_args
|
||||
|
||||
# Check that provider tokens were passed correctly
|
||||
provider_tokens = args[2] # Third argument is immutable_provider_tokens
|
||||
assert provider_tokens is not None
|
||||
# Extract call arguments to verify the provider tokens were set correctly
|
||||
call_args = mock_call_async_from_sync.call_args
|
||||
provider_tokens = call_args[0][2] # Third argument is immutable_provider_tokens
|
||||
assert provider_tokens is not None, 'Provider tokens should not be None'
|
||||
|
||||
# Verify all three provider types are present
|
||||
assert ProviderType.GITHUB in provider_tokens
|
||||
assert ProviderType.GITLAB in provider_tokens
|
||||
assert ProviderType.BITBUCKET in provider_tokens
|
||||
assert ProviderType.GITHUB in provider_tokens, (
|
||||
'GITHUB provider should be in provider_tokens'
|
||||
)
|
||||
assert ProviderType.GITLAB in provider_tokens, (
|
||||
'GITLAB provider should be in provider_tokens'
|
||||
)
|
||||
assert ProviderType.BITBUCKET in provider_tokens, (
|
||||
'BITBUCKET provider should be in provider_tokens'
|
||||
)
|
||||
|
||||
# Verify token values
|
||||
assert (
|
||||
provider_tokens[ProviderType.GITHUB].token.get_secret_value()
|
||||
== 'github_token_123'
|
||||
), (
|
||||
f"Expected GITHUB token to be 'github_token_123', got {provider_tokens[ProviderType.GITHUB].token.get_secret_value()}"
|
||||
)
|
||||
assert (
|
||||
provider_tokens[ProviderType.GITLAB].token.get_secret_value()
|
||||
== 'gitlab_token_456'
|
||||
), (
|
||||
f"Expected GITLAB token to be 'gitlab_token_456', got {provider_tokens[ProviderType.GITLAB].token.get_secret_value()}"
|
||||
)
|
||||
assert (
|
||||
provider_tokens[ProviderType.BITBUCKET].token.get_secret_value()
|
||||
== 'username:bitbucket_app_password'
|
||||
), (
|
||||
f"Expected BITBUCKET token to be 'username:bitbucket_app_password', got {provider_tokens[ProviderType.BITBUCKET].token.get_secret_value()}"
|
||||
)
|
||||
|
||||
|
||||
@@ -861,13 +892,272 @@ def test_initialize_repository_for_runtime_without_bitbucket_token(
|
||||
|
||||
# Verify that call_async_from_sync was called
|
||||
mock_call_async_from_sync.assert_called_once()
|
||||
args, kwargs = mock_call_async_from_sync.call_args
|
||||
|
||||
# Check that provider tokens were passed correctly
|
||||
provider_tokens = args[2] # Third argument is immutable_provider_tokens
|
||||
assert provider_tokens is not None
|
||||
# Extract call arguments to verify the provider tokens were set correctly
|
||||
call_args = mock_call_async_from_sync.call_args
|
||||
provider_tokens = call_args[0][2] # Third argument is immutable_provider_tokens
|
||||
assert provider_tokens is not None, 'Provider tokens should not be None'
|
||||
|
||||
# Verify only GitHub and GitLab are present, not Bitbucket
|
||||
assert ProviderType.GITHUB in provider_tokens
|
||||
assert ProviderType.GITLAB in provider_tokens
|
||||
assert ProviderType.BITBUCKET not in provider_tokens
|
||||
assert ProviderType.GITHUB in provider_tokens, (
|
||||
'GITHUB provider should be in provider_tokens'
|
||||
)
|
||||
assert ProviderType.GITLAB in provider_tokens, (
|
||||
'GITLAB provider should be in provider_tokens'
|
||||
)
|
||||
assert ProviderType.BITBUCKET not in provider_tokens, (
|
||||
'BITBUCKET provider should not be in provider_tokens'
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bitbucket_order_parameter_honored():
|
||||
"""Test that the Bitbucket service correctly honors the order parameter (asc/desc)."""
|
||||
# Create a service instance
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Mock the _make_request method to avoid actual API calls
|
||||
with patch.object(service, '_make_request') as mock_request:
|
||||
# Mock response for repositories
|
||||
mock_request.return_value = ({'values': []}, {})
|
||||
|
||||
# Test ascending order
|
||||
await service.get_paginated_repos(
|
||||
page=1,
|
||||
per_page=10,
|
||||
sort='updated',
|
||||
installation_id='test-workspace',
|
||||
query=None,
|
||||
order='asc',
|
||||
)
|
||||
|
||||
# Verify the call was made with ascending order (no '-' prefix)
|
||||
mock_request.assert_called_once()
|
||||
call_args = mock_request.call_args
|
||||
url, params = call_args[0]
|
||||
assert params['sort'] == 'updated_on', (
|
||||
f"Expected sort parameter 'updated_on', got {params.get('sort')}"
|
||||
)
|
||||
assert 'repositories/test-workspace' in url, (
|
||||
f"Expected URL to contain 'repositories/test-workspace', got {url}"
|
||||
)
|
||||
|
||||
# Reset mock for next test
|
||||
mock_request.reset_mock()
|
||||
|
||||
# Test descending order
|
||||
await service.get_paginated_repos(
|
||||
page=1,
|
||||
per_page=10,
|
||||
sort='updated',
|
||||
installation_id='test-workspace',
|
||||
query=None,
|
||||
order='desc',
|
||||
)
|
||||
|
||||
# Verify the call was made with descending order ('-' prefix)
|
||||
mock_request.assert_called_once()
|
||||
call_args = mock_request.call_args
|
||||
url, params = call_args[0]
|
||||
assert params['sort'] == '-updated_on', (
|
||||
f"Expected sort parameter '-updated_on', got {params.get('sort')}"
|
||||
)
|
||||
assert 'repositories/test-workspace' in url, (
|
||||
f"Expected URL to contain 'repositories/test-workspace', got {url}"
|
||||
)
|
||||
|
||||
# Reset mock for next test
|
||||
mock_request.reset_mock()
|
||||
|
||||
# Test default order (should be descending)
|
||||
await service.get_paginated_repos(
|
||||
page=1,
|
||||
per_page=10,
|
||||
sort='created',
|
||||
installation_id='test-workspace',
|
||||
query=None,
|
||||
# order parameter omitted, should default to 'desc'
|
||||
)
|
||||
|
||||
# Verify the call was made with descending order by default
|
||||
mock_request.assert_called_once()
|
||||
call_args = mock_request.call_args
|
||||
url, params = call_args[0]
|
||||
assert params['sort'] == '-created_on', (
|
||||
f"Expected sort parameter '-created_on', got {params.get('sort')}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bitbucket_search_repositories_passes_order():
|
||||
"""Test that search_repositories correctly passes the order parameter to get_paginated_repos."""
|
||||
# Create a service instance
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Mock the get_installations method
|
||||
with patch.object(service, 'get_installations') as mock_installations:
|
||||
mock_installations.return_value = ['test-workspace']
|
||||
|
||||
# Mock the get_paginated_repos method to capture calls
|
||||
with patch.object(service, 'get_paginated_repos') as mock_paginated_repos:
|
||||
mock_paginated_repos.return_value = []
|
||||
|
||||
# Test ascending order
|
||||
await service.search_repositories(
|
||||
query='test-repo',
|
||||
per_page=10,
|
||||
sort='updated',
|
||||
order='asc',
|
||||
public=False,
|
||||
)
|
||||
|
||||
# Verify get_paginated_repos was called with the correct order parameter
|
||||
mock_paginated_repos.assert_called()
|
||||
|
||||
# Check that at least one call included the 'asc' order parameter
|
||||
calls_with_asc = [
|
||||
call_args
|
||||
for call_args in mock_paginated_repos.call_args_list
|
||||
if len(call_args[0]) > 5
|
||||
and call_args[0][5] == 'asc' # order is the 6th positional argument
|
||||
]
|
||||
assert len(calls_with_asc) > 0, (
|
||||
"Expected at least one call with 'asc' order parameter"
|
||||
)
|
||||
|
||||
# Reset mock for next test
|
||||
mock_paginated_repos.reset_mock()
|
||||
|
||||
# Test descending order
|
||||
await service.search_repositories(
|
||||
query='test-repo',
|
||||
per_page=10,
|
||||
sort='created',
|
||||
order='desc',
|
||||
public=False,
|
||||
)
|
||||
|
||||
# Verify get_paginated_repos was called with the correct order parameter
|
||||
mock_paginated_repos.assert_called()
|
||||
|
||||
# Check that at least one call included the 'desc' order parameter
|
||||
calls_with_desc = [
|
||||
call_args
|
||||
for call_args in mock_paginated_repos.call_args_list
|
||||
if len(call_args[0]) > 5
|
||||
and call_args[0][5] == 'desc' # order is the 6th positional argument
|
||||
]
|
||||
assert len(calls_with_desc) > 0, (
|
||||
"Expected at least one call with 'desc' order parameter"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bitbucket_sort_param_all_sort_types():
|
||||
"""Test _get_bitbucket_sort_param with all supported sort types and default desc order."""
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Test all sort types with default desc order
|
||||
assert service._get_bitbucket_sort_param('pushed') == '-updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated') == '-updated_on'
|
||||
assert service._get_bitbucket_sort_param('created') == '-created_on'
|
||||
assert service._get_bitbucket_sort_param('full_name') == '-name'
|
||||
assert service._get_bitbucket_sort_param('unknown_sort') == '-updated_on' # default
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bitbucket_sort_param_asc_order():
|
||||
"""Test _get_bitbucket_sort_param with ascending order for all sort types."""
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Test all sort types with ascending order
|
||||
assert service._get_bitbucket_sort_param('pushed', 'asc') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'asc') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('created', 'asc') == 'created_on'
|
||||
assert service._get_bitbucket_sort_param('full_name', 'asc') == 'name'
|
||||
assert (
|
||||
service._get_bitbucket_sort_param('unknown_sort', 'asc') == 'updated_on'
|
||||
) # default
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bitbucket_sort_param_desc_order():
|
||||
"""Test _get_bitbucket_sort_param with explicit descending order for all sort types."""
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Test all sort types with explicit descending order
|
||||
assert service._get_bitbucket_sort_param('pushed', 'desc') == '-updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'desc') == '-updated_on'
|
||||
assert service._get_bitbucket_sort_param('created', 'desc') == '-created_on'
|
||||
assert service._get_bitbucket_sort_param('full_name', 'desc') == '-name'
|
||||
assert (
|
||||
service._get_bitbucket_sort_param('unknown_sort', 'desc') == '-updated_on'
|
||||
) # default
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bitbucket_sort_param_case_insensitive_order():
|
||||
"""Test _get_bitbucket_sort_param with case-insensitive order parameters."""
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Test case insensitive order parameters
|
||||
assert service._get_bitbucket_sort_param('updated', 'DESC') == '-updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'Desc') == '-updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'ASC') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'Asc') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'asc') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'desc') == '-updated_on'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bitbucket_sort_param_invalid_order():
|
||||
"""Test _get_bitbucket_sort_param with invalid order parameters defaults to asc (no prefix)."""
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Test invalid order parameters - should default to asc behavior (no prefix)
|
||||
# Only 'desc' (case insensitive) gets the '-' prefix, everything else is treated as asc
|
||||
assert service._get_bitbucket_sort_param('updated', 'invalid') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', '') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('updated', 'random') == 'updated_on'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bitbucket_sort_param_edge_cases():
|
||||
"""Test _get_bitbucket_sort_param with edge cases and boundary conditions."""
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Test empty sort parameter - should default to updated_on
|
||||
assert service._get_bitbucket_sort_param('', 'asc') == 'updated_on'
|
||||
assert service._get_bitbucket_sort_param('', 'desc') == '-updated_on'
|
||||
|
||||
# Test None-like values (if they could be passed)
|
||||
assert (
|
||||
service._get_bitbucket_sort_param('None', 'asc') == 'updated_on'
|
||||
) # treated as unknown
|
||||
|
||||
# Test whitespace handling
|
||||
assert (
|
||||
service._get_bitbucket_sort_param(' updated ', 'asc') == 'updated_on'
|
||||
) # treated as unknown due to spaces
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bitbucket_sort_param_mapping_correctness():
|
||||
"""Test that _get_bitbucket_sort_param correctly maps to Bitbucket API field names."""
|
||||
service = BitBucketService(token=SecretStr('test-token'))
|
||||
|
||||
# Verify the mapping is correct for Bitbucket API
|
||||
# 'pushed' -> 'updated_on' (Bitbucket doesn't have pushed_at)
|
||||
assert 'updated_on' in service._get_bitbucket_sort_param('pushed', 'asc')
|
||||
|
||||
# 'updated' -> 'updated_on'
|
||||
assert 'updated_on' in service._get_bitbucket_sort_param('updated', 'asc')
|
||||
|
||||
# 'created' -> 'created_on'
|
||||
assert 'created_on' in service._get_bitbucket_sort_param('created', 'asc')
|
||||
|
||||
# 'full_name' -> 'name' (Bitbucket uses 'name' field)
|
||||
assert service._get_bitbucket_sort_param('full_name', 'asc') == 'name'
|
||||
|
||||
# Default case -> 'updated_on'
|
||||
assert 'updated_on' in service._get_bitbucket_sort_param('anything_else', 'asc')
|
||||
|
||||
Reference in New Issue
Block a user