use pipelinestep

Author: christineyu123

clarifai/cli/pipeline.py

@@ -93,61 +93,6 @@ def pipeline():
     """Create and manage pipelines."""
 
 
-@pipeline.command('local-run')
-@click.argument("step_path", type=click.Path(exists=True), required=False, default=".")
-@click.option(
-    '--mode',
-    type=click.Choice(['container']),
-    default='container',
-    help='Execution mode. Currently only "container" is supported.',
-)
-@click.option(
-    '--keep-image',
-    is_flag=True,
-    default=False,
-    help='Keep the Docker image after the pipeline step finishes.',
-)
-def local_run(step_path, mode, keep_image):
-    """Run a pipeline step locally in a Docker container.
-
-    STEP_PATH: Path to the pipeline step directory (containing config.yaml,
-    requirements.txt, and 1/pipeline_step.py). Defaults to current directory.
-
-    This reuses the same Docker build infrastructure as ``clarifai model serve
-    --mode container`` but executes pipeline_step.py once and exits.
-    """
-    from clarifai.runners.pipeline_steps.pipeline_run_locally import PipelineStepRunLocally
-
-    manager = PipelineStepRunLocally(step_path)
-
-    if not manager.is_docker_installed():
-        raise click.ClickException("Docker is not installed.")
-
-    # Generate Dockerfile if missing
-    manager.builder.create_dockerfile()
-
-    image_tag = manager._docker_hash()
-    step_id = manager.config['pipeline_step']['id'].lower()
-    image_name = f"{step_id}:{image_tag}"
-    container_name = f"{step_id}-local-run"
-
-    if not manager.docker_image_exists(image_name):
-        logger.info("Building Docker image...")
-        manager.build_docker_image(image_name=image_name)
-
-    try:
-        manager.run_pipeline_step_container(
-            image_name=image_name,
-            container_name=container_name,
-        )
-    finally:
-        if manager.container_exists(container_name):
-            manager.stop_docker_container(container_name)
-            manager.remove_docker_container(container_name)
-        if not keep_image:
-            manager.remove_docker_image(image_name)
-
-
 @pipeline.command()
 @click.argument("path", type=click.Path(exists=True), required=False, default=".")
 @click.option(

clarifai/cli/pipeline_step.py

@@ -111,6 +111,61 @@ def init(pipeline_step_path):
     logger.info("4. Implement your pipeline step logic in 1/pipeline_step.py")
 
 
+@pipeline_step.command('local-run')
+@click.argument("pipeline_step_path", type=click.Path(exists=True), required=False, default=".")
+@click.option(
+    '--mode',
+    type=click.Choice(['container']),
+    default='container',
+    help='Execution mode. Currently only "container" is supported.',
+)
+@click.option(
+    '--keep-image',
+    is_flag=True,
+    default=False,
+    help='Keep the Docker image after the pipeline step finishes.',
+)
+def local_run(pipeline_step_path, mode, keep_image):
+    """Run a pipeline step locally in a Docker container.
+
+    PIPELINE_STEP_PATH: Path to the pipeline step directory (containing config.yaml,
+    requirements.txt, and 1/pipeline_step.py). Defaults to current directory.
+
+    This reuses the same Docker build infrastructure as ``clarifai model serve
+    --mode container`` but executes pipeline_step.py once and exits.
+    """
+    from clarifai.runners.pipeline_steps.pipeline_run_locally import PipelineStepRunLocally
+
+    manager = PipelineStepRunLocally(pipeline_step_path)
+
+    if not manager.is_docker_installed():
+        raise click.ClickException("Docker is not installed.")
+
+    # Generate Dockerfile if missing
+    manager.builder.create_dockerfile()
+
+    image_tag = manager._docker_hash()
+    step_id = manager.config['pipeline_step']['id'].lower()
+    image_name = f"{step_id}:{image_tag}"
+    container_name = f"{step_id}-local-run"
+
+    if not manager.docker_image_exists(image_name):
+        logger.info("Building Docker image...")
+        manager.build_docker_image(image_name=image_name)
+
+    try:
+        manager.run_pipeline_step_container(
+            image_name=image_name,
+            container_name=container_name,
+        )
+    finally:
+        if manager.container_exists(container_name):
+            manager.stop_docker_container(container_name)
+            manager.remove_docker_container(container_name)
+        if not keep_image:
+            manager.remove_docker_image(image_name)
+
+
 @pipeline_step.command(['ls'])
 @click.option('--page_no', required=False, help='Page number to list.', default=1)
 @click.option('--per_page', required=False, help='Number of items per page.', default=16)

tests/cli/test_pipeline_step.py

@@ -8,7 +8,7 @@
 import pytest
 from click.testing import CliRunner
 
-from clarifai.cli.pipeline_step import init, list, upload
+from clarifai.cli.pipeline_step import init, list, local_run, upload
 
 
 class TestPipelineStepInitCommand:
@@ -452,3 +452,55 @@ def test_list_command_help(self):
         assert '--pipeline_id' in result.output
         assert '--page_no' in result.output
         assert '--per_page' in result.output
+
+    def test_local_run_command_help(self):
+        """Test that local-run command shows helpful usage information."""
+        runner = CliRunner()
+
+        result = runner.invoke(local_run, ['--help'])
+
+        assert result.exit_code == 0
+        assert 'Run a pipeline step locally in a Docker container' in result.output
+        assert '--mode' in result.output
+        assert '--keep-image' in result.output
+        assert 'PIPELINE_STEP_PATH' in result.output
+
+
+class TestPipelineStepLocalRunCommand:
+    """Test cases for the pipeline step local-run CLI command."""
+
+    def test_local_run_nonexistent_path(self):
+        """Test local-run with nonexistent path fails."""
+        runner = CliRunner()
+
+        result = runner.invoke(local_run, ['nonexistent_path'])
+
+        assert result.exit_code != 0
+        assert 'does not exist' in result.output
+
+    @patch(
+        'clarifai.runners.pipeline_steps.pipeline_run_locally.PipelineStepRunLocally.is_docker_installed',
+        return_value=False,
+    )
+    def test_local_run_no_docker(self, mock_docker):
+        """Test local-run fails gracefully when Docker is not installed."""
+        runner = CliRunner()
+
+        with runner.isolated_filesystem():
+            # Create minimal pipeline step structure
+            os.makedirs('1', exist_ok=True)
+            with open('config.yaml', 'w') as f:
+                f.write(
+                    'pipeline_step:\n  id: test\n  user_id: u\n  app_id: a\n'
+                    'build_info:\n  python_version: "3.12"\n'
+                    'pipeline_step_compute_info:\n  cpu_limit: "500m"\n  cpu_memory: "500Mi"\n  num_accelerators: 0\n'
+                )
+            with open('requirements.txt', 'w') as f:
+                f.write('clarifai\n')
+            with open('1/pipeline_step.py', 'w') as f:
+                f.write('print("hello")\n')
+
+            result = runner.invoke(local_run, ['.'])
+
+            assert result.exit_code != 0
+            assert 'Docker is not installed' in result.output

…s/test_pipeline_run_locally_container.py …t_pipeline_step_run_locally_container.pytests/runners/test_pipeline_run_locally_container.py renamed to tests/runners/test_pipeline_step_run_locally_container.py tests/runners/test_pipeline_run_locally_container.py renamed to tests/runners/test_pipeline_step_run_locally_container.py

@@ -21,8 +21,8 @@ def dummy_pipeline_step_path(tmp_path):
 
 
 @pytest.fixture
-def pipeline_run_locally(dummy_pipeline_step_path):
-    """Instantiate PipelineRunLocally with the dummy pipeline step."""
+def pipeline_step_run_locally(dummy_pipeline_step_path):
+    """Instantiate PipelineStepRunLocally with the dummy pipeline step."""
     return PipelineStepRunLocally(dummy_pipeline_step_path)
 
 
@@ -31,28 +31,28 @@ def pipeline_run_locally(dummy_pipeline_step_path):
     sys.platform not in ["linux", "darwin"],
     reason="Test only runs on Linux and macOS.",
 )
-def test_pipeline_step_docker_build_and_run(pipeline_run_locally):
+def test_pipeline_step_docker_build_and_run(pipeline_step_run_locally):
     """Test building a Docker image and running a pipeline step in a container."""
-    assert pipeline_run_locally.is_docker_installed(), "Docker not installed."
+    assert pipeline_step_run_locally.is_docker_installed(), "Docker not installed."
 
-    pipeline_run_locally.builder.create_dockerfile()
-    image_tag = pipeline_run_locally._docker_hash()
-    step_id = pipeline_run_locally.config['pipeline_step']['id'].lower()
+    pipeline_step_run_locally.builder.create_dockerfile()
+    image_tag = pipeline_step_run_locally._docker_hash()
+    step_id = pipeline_step_run_locally.config['pipeline_step']['id'].lower()
     image_name = f"{step_id}:{image_tag}"
     container_name = "test-pipeline-step-container"
 
-    if not pipeline_run_locally.docker_image_exists(image_name):
-        pipeline_run_locally.build_docker_image(image_name=image_name)
+    if not pipeline_step_run_locally.docker_image_exists(image_name):
+        pipeline_step_run_locally.build_docker_image(image_name=image_name)
 
     try:
-        pipeline_run_locally.run_pipeline_step_container(
+        pipeline_step_run_locally.run_pipeline_step_container(
             image_name=image_name,
             container_name=container_name,
         )
     except subprocess.CalledProcessError:
         pytest.fail("Failed to run pipeline step inside the docker container.")
     finally:
-        if pipeline_run_locally.container_exists(container_name):
-            pipeline_run_locally.stop_docker_container(container_name)
-            pipeline_run_locally.remove_docker_container(container_name)
-        pipeline_run_locally.remove_docker_image(image_name)
+        if pipeline_step_run_locally.container_exists(container_name):
+            pipeline_step_run_locally.stop_docker_container(container_name)
+            pipeline_step_run_locally.remove_docker_container(container_name)
+        pipeline_step_run_locally.remove_docker_image(image_name)