Skip to content

Add comprehensive test coverage for cli.pipeline_step module #795

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Add comprehensive test coverage for cli.pipeline_step module #795

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bb0a378
Initial plan
Copilot Oct 1, 2025
0c1f609
Add comprehensive test coverage for cli.pipeline_step module
Copilot Oct 1, 2025
a79570e
Merge branch 'master' into copilot/fix-871114d0-6e37-49bc-95b2-07d822…
nitinbhojwani Oct 23, 2025
73eb1ee
Merge branch 'master' into copilot/fix-871114d0-6e37-49bc-95b2-07d822…
nitinbhojwani Oct 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 0 additions & 169 deletions tests/cli/test_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1457,173 +1457,4 @@ def test_list_command_default_parameters(self, mock_validate):
mock_user_instance.list_pipelines.assert_called_once_with(page_no=1, per_page=16)


class TestPipelineStepListCommand:
"""Test cases for the pipeline step list CLI command."""

@patch('clarifai.cli.pipeline_step.validate_context')
@patch('clarifai.cli.pipeline_step.User')
@patch('clarifai.cli.pipeline_step.display_co_resources')
def test_list_command_success_no_app_id(self, mock_display, mock_user_class, mock_validate):
"""Test that list command works without app_id (lists across all apps)."""
# Setup mocks
mock_validate.return_value = None
mock_user_instance = Mock()
mock_user_class.return_value = mock_user_instance
mock_user_instance.list_pipeline_steps.return_value = [
{
'id': 'step1',
'user_id': 'user1',
'app_id': 'app1',
'pipeline_id': 'pipe1',
'description': 'Test step 1',
},
{
'id': 'step2',
'user_id': 'user1',
'app_id': 'app2',
'pipeline_id': 'pipe2',
'description': 'Test step 2',
},
]

# Setup context
runner = CliRunner()
ctx_obj = Mock()
ctx_obj.current.user_id = 'test-user'
ctx_obj.current.pat = 'test-pat'
ctx_obj.current.api_base = 'https://api.clarifai.com'

# Import here to avoid circular imports in testing
from clarifai.cli.pipeline_step import list as list_command

result = runner.invoke(
list_command,
['--page_no', '1', '--per_page', '10'],
obj=ctx_obj,
)

assert result.exit_code == 0
mock_validate.assert_called_once()
mock_user_class.assert_called_once_with(
user_id='test-user', pat='test-pat', base_url='https://api.clarifai.com'
)
mock_user_instance.list_pipeline_steps.assert_called_once_with(page_no=1, per_page=10)
mock_display.assert_called_once()

@patch('clarifai.cli.pipeline_step.validate_context')
@patch('clarifai.cli.pipeline_step.App')
@patch('clarifai.cli.pipeline_step.display_co_resources')
def test_list_command_success_with_app_id(self, mock_display, mock_app_class, mock_validate):
"""Test that list command works with app_id (lists within specific app)."""
# Setup mocks
mock_validate.return_value = None
mock_app_instance = Mock()
mock_app_class.return_value = mock_app_instance
mock_app_instance.list_pipeline_steps.return_value = [
{
'id': 'step1',
'user_id': 'user1',
'app_id': 'app1',
'pipeline_id': 'pipe1',
'description': 'Test step 1',
},
]

# Setup context
runner = CliRunner()
ctx_obj = Mock()
ctx_obj.current.user_id = 'test-user'
ctx_obj.current.pat = 'test-pat'
ctx_obj.current.api_base = 'https://api.clarifai.com'

# Import here to avoid circular imports in testing
from clarifai.cli.pipeline_step import list as list_command

result = runner.invoke(
list_command,
['--app_id', 'test-app', '--page_no', '1', '--per_page', '5'],
obj=ctx_obj,
)

assert result.exit_code == 0
mock_validate.assert_called_once()
mock_app_class.assert_called_once_with(
app_id='test-app',
user_id='test-user',
pat='test-pat',
base_url='https://api.clarifai.com',
)
mock_app_instance.list_pipeline_steps.assert_called_once_with(
pipeline_id=None, page_no=1, per_page=5
)
mock_display.assert_called_once()

@patch('clarifai.cli.pipeline_step.validate_context')
@patch('clarifai.cli.pipeline_step.App')
@patch('clarifai.cli.pipeline_step.display_co_resources')
def test_list_command_success_with_pipeline_id(
self, mock_display, mock_app_class, mock_validate
):
"""Test that list command works with both app_id and pipeline_id."""
# Setup mocks
mock_validate.return_value = None
mock_app_instance = Mock()
mock_app_class.return_value = mock_app_instance
mock_app_instance.list_pipeline_steps.return_value = [
{
'id': 'step1',
'user_id': 'user1',
'app_id': 'app1',
'pipeline_id': 'pipe1',
'description': 'Test step 1',
},
]

# Setup context
runner = CliRunner()
ctx_obj = Mock()
ctx_obj.current.user_id = 'test-user'
ctx_obj.current.pat = 'test-pat'
ctx_obj.current.api_base = 'https://api.clarifai.com'

# Import here to avoid circular imports in testing
from clarifai.cli.pipeline_step import list as list_command

result = runner.invoke(
list_command,
['--app_id', 'test-app', '--pipeline_id', 'test-pipeline'],
obj=ctx_obj,
)

assert result.exit_code == 0
mock_validate.assert_called_once()
mock_app_class.assert_called_once_with(
app_id='test-app',
user_id='test-user',
pat='test-pat',
base_url='https://api.clarifai.com',
)
mock_app_instance.list_pipeline_steps.assert_called_once_with(
pipeline_id='test-pipeline', page_no=1, per_page=16
)
mock_display.assert_called_once()

def test_list_command_pipeline_id_without_app_id_error(self):
"""Test that using pipeline_id without app_id raises an error."""
runner = CliRunner()
ctx_obj = Mock()
ctx_obj.current.user_id = 'test-user'
ctx_obj.current.pat = 'test-pat'
ctx_obj.current.api_base = 'https://api.clarifai.com'

# Import here to avoid circular imports in testing
from clarifai.cli.pipeline_step import list as list_command

result = runner.invoke(
list_command,
['--pipeline_id', 'test-pipeline'],
obj=ctx_obj,
)

assert result.exit_code != 0
assert '--pipeline_id must be used together with --app_id' in result.output
Loading
Loading