Explicit passing of arguments to drop (#3386)

dlt/_workspace/cli/_pipeline_command.py

@@ -1,6 +1,7 @@
 import os
 import yaml
 from typing import Any, Sequence, Tuple
+from inspect import signature
 import dlt
 
 from dlt.common.json import json
@@ -386,7 +387,14 @@ def pipeline_command(
         fmt.echo(schema_str)
 
     if operation == "drop":
-        drop = pipeline_drop(p, **command_kwargs)
+        drop = pipeline_drop(
+            p,
+            resources=command_kwargs.get("resources", ()),
+            schema_name=command_kwargs.get("schema_name"),
+            state_paths=command_kwargs.get("state_paths", ()),
+            drop_all=command_kwargs.get("drop_all", False),
+            state_only=command_kwargs.get("state_only", False),
+        )
         if drop.is_empty:
             fmt.echo(
                 "Could not select any resources to drop and no resource/source state to reset. Use"

tests/workspace/cli/test_pipeline_command.py

@@ -287,3 +287,36 @@ def test_drop_from_wrong_dir(repo_dir: str) -> None:
         )
         _out = buf.getvalue()
         assert "WARNING: You should run this from the same directory as the pipeline script" in _out
+
+
+def test_pipeline_command_drop_with_global_args(repo_dir: str) -> None:
+    """Test that global CLI arguments don't cause errors in pipeline drop command."""
+    _init_command.init_command("chess", "duckdb", repo_dir)
+
+    os.environ.pop("DESTINATION__DUCKDB__CREDENTIALS", None)
+    venv = Venv.restore_current()
+    try:
+        print(venv.run_script("chess_pipeline.py"))
+    except CalledProcessError as cpe:
+        print(cpe.stdout)
+        print(cpe.stderr)
+        raise
+
+    # Test drop command with global arguments that should be ignored
+    with io.StringIO() as buf, contextlib.redirect_stdout(buf):
+        with echo.always_choose(False, True):
+            _pipeline_command.pipeline_command(
+                "drop",
+                "chess_pipeline",
+                None,
+                0,
+                resources=["players_games"],
+                no_pwd=False,  # Global arg that should be ignored
+                debug=False,  # Another global arg
+            )
+        _out = buf.getvalue()
+        assert "Selected resource(s): ['players_games']" in _out
+
+    # Verify the command actually executed
+    pipeline = dlt.attach(pipeline_name="chess_pipeline")
+    assert "players_games" not in pipeline.default_schema.tables