bugfix: rerun stage when command is changed (#3922)

Previously, the stage would not rerun if the deps/outs are cached
even if the cmd has changed.

Versions affected: >0.93.0

Fixes #3918

dvc/stage/run.py

@@ -82,6 +82,7 @@ def _is_cached(stage):
     cached = (
         not stage.is_callback
         and not stage.always_changed
+        and not stage.changed_stage()
         and stage.already_cached()
     )
     if cached:

tests/func/test_repro.py

@@ -1770,3 +1770,25 @@ def test_ssh_dir_out(tmp_dir, dvc, ssh_server):
 
     repo.reproduce("dir-out.dvc")
     repo.reproduce("dir-out.dvc", force=True)
+
+
+def test_repro_when_cmd_changes(tmp_dir, dvc, run_copy, mocker):
+    from dvc.dvcfile import SingleStageFile
+
+    tmp_dir.gen("foo", "foo")
+    stage = run_copy("foo", "bar", single_stage=True)
+    assert not dvc.reproduce(stage.addressing)
+
+    from dvc.stage.run import cmd_run
+
+    m = mocker.patch("dvc.stage.run.cmd_run", wraps=cmd_run)
+
+    data = SingleStageFile(dvc, stage.path)._load()[0]
+    data["cmd"] = "  ".join(stage.cmd.split())  # change cmd spacing by two
+    dump_stage_file(stage.path, data)
+
+    assert dvc.status([stage.addressing]) == {
+        stage.addressing: ["changed checksum"]
+    }
+    assert dvc.reproduce(stage.addressing)[0] == stage
+    m.assert_called_once_with(stage)

tests/func/test_repro_multistage.py

@@ -307,7 +307,7 @@ def test_downstream(tmp_dir, dvc):
     )
 
 
-def test_repro_when_cmd_changes(tmp_dir, dvc, run_copy):
+def test_repro_when_cmd_changes(tmp_dir, dvc, run_copy, mocker):
     from dvc.dvcfile import PipelineFile
 
     tmp_dir.gen("foo", "foo")
@@ -315,11 +315,15 @@ def test_repro_when_cmd_changes(tmp_dir, dvc, run_copy):
     target = "copy-file"
     assert not dvc.reproduce(target)
 
+    from dvc.stage.run import cmd_run
+
+    m = mocker.patch("dvc.stage.run.cmd_run", wraps=cmd_run)
     stage.cmd = "  ".join(stage.cmd.split())  # change cmd spacing by two
     PipelineFile(dvc, PIPELINE_FILE)._dump_pipeline_file(stage)
 
     assert dvc.status([target]) == {target: ["changed command"]}
     assert dvc.reproduce(target)[0] == stage
+    m.assert_called_once_with(stage)
 
 
 def test_repro_when_new_deps_is_added_in_dvcfile(tmp_dir, dvc, run_copy):