diff --git a/mkdocs/docs/reference/cli/dstack/apply.md b/mkdocs/docs/reference/cli/dstack/apply.md index 4cc6215a0..23a399084 100644 --- a/mkdocs/docs/reference/cli/dstack/apply.md +++ b/mkdocs/docs/reference/cli/dstack/apply.md @@ -17,6 +17,34 @@ $ dstack apply --help +## Preventing implicit run recreation + +Use `--no-recreate` when a script should ensure that an exact named run is active without +silently changing, stopping, or replacing an existing run: + +```shell +dstack apply -f dev.dstack.yml --no-recreate -y -d +``` + +The option supports named dev environment, task, and service configurations. It rejects unnamed +runs and non-run configurations, and it is mutually exclusive with `--force`. + +If the named run is absent or finished, the command follows the normal apply flow. If the run is +active, the command succeeds as a no-op only when all of the following are true: + +- the run is owned by the user requesting the plan; +- the plan requires an in-place update action but the effective run specification has no changes; +- the run is `submitted`, `provisioning`, or `running`. + +Any configuration change is rejected, including one that could normally be updated in place. +Recreation actions and runs in `pending` or `terminating` state are also rejected. These failures +happen before the CLI sends a stop or apply request. Stop the run explicitly before applying a +configuration that needs to replace it. + +For absent or finished runs, the server still validates that the resource observed by the plan has +not changed before applying it. A concurrent same-name run therefore causes the apply to fail +instead of being overwritten. + ## User SSH key By default, `dstack` uses its own SSH key to attach to runs (`~/.dstack/ssh/id_rsa`). diff --git a/src/dstack/_internal/cli/commands/apply.py b/src/dstack/_internal/cli/commands/apply.py index c10f54f33..6602b0315 100644 --- a/src/dstack/_internal/cli/commands/apply.py +++ b/src/dstack/_internal/cli/commands/apply.py @@ -11,9 +11,25 @@ ) from dstack._internal.cli.utils.common import console from dstack._internal.core.errors import CLIError -from dstack._internal.core.models.configurations import ApplyConfigurationType +from dstack._internal.core.models.configurations import ( + AnyApplyConfiguration, + ApplyConfigurationType, +) NOTSET = object() +_RUN_CONFIGURATION_TYPES = { + ApplyConfigurationType.DEV_ENVIRONMENT, + ApplyConfigurationType.TASK, + ApplyConfigurationType.SERVICE, +} + + +def validate_no_recreate_configuration(configuration: AnyApplyConfiguration) -> None: + configuration_type = ApplyConfigurationType(configuration.type) + if configuration_type not in _RUN_CONFIGURATION_TYPES: + raise CLIError("--no-recreate is only supported for run configurations") + if configuration.name is None: + raise CLIError("--no-recreate requires a named run") class ApplyCommand(APIBaseCommand): @@ -51,11 +67,17 @@ def _register(self): help="Do not ask for confirmation", action="store_true", ) - self._parser.add_argument( + apply_safety = self._parser.add_mutually_exclusive_group() + apply_safety.add_argument( "--force", help="Force apply when no changes detected", action="store_true", ) + apply_safety.add_argument( + "--no-recreate", + help="Fail instead of changing an active run; unchanged active runs are no-op", + action="store_true", + ) self._parser.add_argument( "-d", "--detach", @@ -89,6 +111,8 @@ def _command(self, args: argparse.Namespace): if not args.yes and args.configuration_file == APPLY_STDIN_NAME: raise CLIError("Cannot read configuration from stdin if -y/--yes is not specified") configuration_path, configuration = load_apply_configuration(args.configuration_file) + if args.no_recreate: + validate_no_recreate_configuration(configuration) configurator_class = get_apply_configurator_class(configuration.type) configurator = configurator_class(api_client=self.api) configurator_parser = configurator.get_parser() diff --git a/src/dstack/_internal/cli/services/configurators/run.py b/src/dstack/_internal/cli/services/configurators/run.py index 0733e4f71..c4c1f6fac 100644 --- a/src/dstack/_internal/cli/services/configurators/run.py +++ b/src/dstack/_internal/cli/services/configurators/run.py @@ -6,6 +6,8 @@ import subprocess import sys import time +from dataclasses import dataclass +from enum import Enum from pathlib import Path from typing import Dict, List, Optional, TypeVar @@ -65,6 +67,9 @@ RunSpec, RunStatus, ) +from dstack._internal.core.models.runs import ( + Run as RunModel, +) from dstack._internal.core.services.diff import diff_models from dstack._internal.core.services.repos import get_repo_creds_and_default_branch from dstack._internal.core.services.ssh.ports import PortUsedError @@ -77,12 +82,70 @@ from dstack.api._public.runs import Run _BIND_ADDRESS_ARG = "bind_address" +_NO_RECREATE_NOOP_STATUSES = { + RunStatus.SUBMITTED, + RunStatus.PROVISIONING, + RunStatus.RUNNING, +} logger = get_logger(__name__) RunConfigurationT = TypeVar("RunConfigurationT", bound=AnyRunConfiguration) +class ApplyPlanResult(str, Enum): + """Outcome of the non-streaming part of applying a run plan.""" + + CANCELLED = "cancelled" + NOOP = "noop" + SUBMITTED = "submitted" + + +@dataclass(frozen=True) +class RunApplyFence: + run_id: str + deployment_num: int + user: str + + +@dataclass(frozen=True) +class ApplyPlanOutcome: + result: ApplyPlanResult + fence: Optional[RunApplyFence] = None + + +def get_run_apply_fence(run: RunModel) -> RunApplyFence: + return RunApplyFence( + run_id=str(run.id), + deployment_num=run.deployment_num, + user=run.user, + ) + + +def validate_no_recreate_plan(run_plan: RunPlan) -> Optional[RunApplyFence]: + """Validate an active plan using the fail-closed no-recreate contract.""" + current = run_plan.current_resource + if current is None or current.status.is_finished(): + return None + run_name = run_plan.run_spec.run_name + if current.user != run_plan.user: + raise CLIError( + f"Refusing to use active run {run_name} with --no-recreate;" + " the run is owned by another user" + ) + diff = render_run_spec_diff(run_plan.get_effective_run_spec(), current.run_spec) + if ( + run_plan.action == ApplyAction.UPDATE + and diff is None + and current.status in _NO_RECREATE_NOOP_STATUSES + ): + return get_run_apply_fence(current) + raise CLIError( + f"Refusing to change active run {run_name} with --no-recreate;" + " stop it explicitly before applying this configuration" + ) + + class BaseRunConfigurator( ApplyEnvVarsConfiguratorMixin, BaseApplyConfigurator[RunConfigurationT], @@ -99,12 +162,15 @@ def apply_configuration( configuration_path=configuration_path, configurator_args=configurator_args, ) - return self.apply_plan( + self.apply_plan( run_plan=run_plan, repo=repo, command_args=command_args, configurator_args=configurator_args, ) + # Preserve the historical apply_configuration() return contract. Callers that need a + # structured outcome can invoke apply_plan() directly. + return None def get_plan( self, @@ -145,9 +211,16 @@ def apply_plan( command_args: argparse.Namespace, configurator_args: argparse.Namespace, plan_properties: Optional[Dict[str, str]] = None, - ): - """Apply a run plan using the standard CLI behavior.""" + ) -> Optional[ApplyPlanOutcome]: + """Apply a run plan using the standard CLI behavior. + + A structured outcome is returned for cancellation, no-op, and detached submission. + Foreground apply retains its existing attach and interrupt behavior and may return ``None``. + """ run_name = run_plan.run_spec.run_name + no_recreate = getattr(command_args, "no_recreate", False) + if no_recreate and run_name is None: + raise CLIError("--no-recreate requires a named run") no_fleets = False if len(run_plan.job_plans[0].offers) == 0: @@ -171,6 +244,14 @@ def apply_plan( run_plan.get_effective_run_spec(), run_plan.current_resource.run_spec, ) + if no_recreate and not run_plan.current_resource.status.is_finished(): + fence = validate_no_recreate_plan(run_plan) + assert fence is not None + console.print(f"Run [code]{run_name}[/] is already up to date.") + return ApplyPlanOutcome( + result=ApplyPlanResult.NOOP, + fence=fence, + ) if run_plan.action == ApplyAction.UPDATE and diff is not None: console.print( f"Active run [code]{run_name}[/] already exists." @@ -184,7 +265,10 @@ def apply_plan( ) if command_args.yes and not command_args.force: console.print("Use --force to apply anyway.") - return + return ApplyPlanOutcome( + result=ApplyPlanResult.NOOP, + fence=get_run_apply_fence(run_plan.current_resource), + ) confirm_message = "Stop and override the run?" elif not run_plan.current_resource.status.is_finished(): stop_run_name = run_plan.current_resource.run_spec.run_name @@ -198,7 +282,7 @@ def apply_plan( if not command_args.yes and not confirm_ask(confirm_message): console.print("\nExiting...") - return + return ApplyPlanOutcome(result=ApplyPlanResult.CANCELLED) if stop_run_name is not None: with console.status("Stopping run..."): @@ -229,7 +313,10 @@ def apply_plan( if run_plan.action == ApplyAction.UPDATE: detach_message = f"Run [code]{run.name}[/] updated, detaching..." console.print(detach_message) - return + return ApplyPlanOutcome( + result=ApplyPlanResult.SUBMITTED, + fence=get_run_apply_fence(run._run), + ) abort_at_exit = False try: diff --git a/src/tests/_internal/cli/commands/test_apply.py b/src/tests/_internal/cli/commands/test_apply.py new file mode 100644 index 000000000..dadc5a65c --- /dev/null +++ b/src/tests/_internal/cli/commands/test_apply.py @@ -0,0 +1,82 @@ +from types import SimpleNamespace +from typing import cast + +import pytest + +from dstack._internal.cli.commands.apply import validate_no_recreate_configuration +from dstack._internal.core.errors import CLIError +from dstack._internal.core.models.configurations import ( + AnyApplyConfiguration, + ApplyConfigurationType, +) +from tests._internal.cli.common import run_dstack_cli + + +def _configuration( + configuration_type: ApplyConfigurationType, + *, + name: str | None, +) -> AnyApplyConfiguration: + return cast( + AnyApplyConfiguration, + SimpleNamespace(type=configuration_type.value, name=name), + ) + + +def test_help_documents_no_recreate(capsys: pytest.CaptureFixture[str]) -> None: + assert run_dstack_cli(["apply", "--help"]) == 0 + + normalized_output = " ".join(capsys.readouterr().out.split()) + assert "[--force | --no-recreate]" in normalized_output + assert ( + "Fail instead of changing an active run; unchanged active runs are no-op" + in normalized_output + ) + + +def test_no_recreate_is_mutually_exclusive_with_force( + capsys: pytest.CaptureFixture[str], +) -> None: + assert run_dstack_cli(["apply", "--force", "--no-recreate"]) == 2 + + assert "not allowed with argument --force" in capsys.readouterr().err + + +@pytest.mark.parametrize( + "configuration_type", + [ + ApplyConfigurationType.FLEET, + ApplyConfigurationType.GATEWAY, + ApplyConfigurationType.VOLUME, + ], +) +def test_no_recreate_rejects_non_run_configurations( + configuration_type: ApplyConfigurationType, +) -> None: + configuration = _configuration(configuration_type, name="resource") + + with pytest.raises(CLIError, match="only supported for run configurations"): + validate_no_recreate_configuration(configuration) + + +@pytest.mark.parametrize( + "configuration_type", + [ + ApplyConfigurationType.DEV_ENVIRONMENT, + ApplyConfigurationType.TASK, + ApplyConfigurationType.SERVICE, + ], +) +def test_no_recreate_accepts_named_run_configurations( + configuration_type: ApplyConfigurationType, +) -> None: + configuration = _configuration(configuration_type, name="dev-run") + + validate_no_recreate_configuration(configuration) + + +def test_no_recreate_rejects_unnamed_run_configuration() -> None: + configuration = _configuration(ApplyConfigurationType.DEV_ENVIRONMENT, name=None) + + with pytest.raises(CLIError, match="requires a named run"): + validate_no_recreate_configuration(configuration) diff --git a/src/tests/_internal/cli/services/configurators/test_run.py b/src/tests/_internal/cli/services/configurators/test_run.py index 1e86759a5..a4a6d3a42 100644 --- a/src/tests/_internal/cli/services/configurators/test_run.py +++ b/src/tests/_internal/cli/services/configurators/test_run.py @@ -113,15 +113,16 @@ class TestApplyConfiguration: def test_composes_get_plan_and_apply_plan(self, monkeypatch): run_plan, repo = Mock(), Mock() get_plan = Mock(return_value=(run_plan, repo)) - apply_plan = Mock() + apply_plan = Mock(return_value=object()) monkeypatch.setattr(ServiceConfigurator, "get_plan", get_plan) monkeypatch.setattr(ServiceConfigurator, "apply_plan", apply_plan) conf, command_args, configurator_args = Mock(), Mock(), Mock() - ServiceConfigurator(api_client=Mock()).apply_configuration( + result = ServiceConfigurator(api_client=Mock()).apply_configuration( conf, "svc.dstack.yml", command_args, configurator_args ) + assert result is None get_plan.assert_called_once_with( conf=conf, configuration_path="svc.dstack.yml", diff --git a/src/tests/_internal/cli/services/configurators/test_run_no_recreate.py b/src/tests/_internal/cli/services/configurators/test_run_no_recreate.py new file mode 100644 index 000000000..20bc2768c --- /dev/null +++ b/src/tests/_internal/cli/services/configurators/test_run_no_recreate.py @@ -0,0 +1,324 @@ +from argparse import Namespace +from types import SimpleNamespace +from unittest.mock import Mock +from uuid import uuid4 + +import pytest + +from dstack._internal.cli.services.configurators import run as run_configurator +from dstack._internal.cli.services.configurators.run import ( + ApplyPlanOutcome, + ApplyPlanResult, + BaseRunConfigurator, + RunApplyFence, +) +from dstack._internal.core.errors import CLIError +from dstack._internal.core.models.common import ApplyAction +from dstack._internal.core.models.runs import RunStatus + + +def _plan(*, action: ApplyAction, current_resource, run_name: str | None = "dev-run"): + run_spec = SimpleNamespace(run_name=run_name) + return SimpleNamespace( + user="owner", + run_spec=run_spec, + effective_run_spec=run_spec, + get_effective_run_spec=lambda: run_spec, + job_plans=[SimpleNamespace(offers=[object()])], + current_resource=current_resource, + action=action, + ) + + +def _current_resource(*, status: RunStatus = RunStatus.RUNNING, user: str = "owner"): + return SimpleNamespace( + id=uuid4(), + deployment_num=4, + status=status, + user=user, + run_spec=SimpleNamespace(run_name="dev-run"), + ) + + +def _applied_run(*, deployment_num: int = 5, user: str = "owner"): + run = Mock() + run.name = "dev-run" + run._run = SimpleNamespace( + id=uuid4(), + deployment_num=deployment_num, + user=user, + ) + return run + + +def _args(**overrides): + values = { + "yes": True, + "force": False, + "no_recreate": True, + "detach": True, + "verbose": False, + } + values.update(overrides) + return Namespace(**values) + + +def _configurator_args(): + return Namespace(max_offers=3) + + +@pytest.fixture(autouse=True) +def quiet_plan(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(run_configurator, "print_run_plan", Mock()) + + +class TestApplyPlanNoRecreate: + @pytest.mark.parametrize( + "status", + [RunStatus.SUBMITTED, RunStatus.PROVISIONING, RunStatus.RUNNING], + ) + def test_treats_unchanged_safe_active_run_as_noop( + self, + status: RunStatus, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + ) -> None: + current = _current_resource(status=status) + plan = _plan(action=ApplyAction.UPDATE, current_resource=current) + api = Mock() + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: None) + + outcome = BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(detach=False), + configurator_args=_configurator_args(), + ) + + assert outcome == ApplyPlanOutcome( + result=ApplyPlanResult.NOOP, + fence=RunApplyFence( + run_id=str(current.id), + deployment_num=current.deployment_num, + user=current.user, + ), + ) + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_not_called() + assert "already up to date" in capsys.readouterr().out + + @pytest.mark.parametrize("status", [RunStatus.PENDING, RunStatus.TERMINATING]) + def test_refuses_unsafe_active_status_even_when_unchanged( + self, + status: RunStatus, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + current = _current_resource(status=status) + plan = _plan(action=ApplyAction.UPDATE, current_resource=current) + api = Mock() + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: None) + + with pytest.raises(CLIError, match="Refusing to change active run"): + BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(), + configurator_args=_configurator_args(), + ) + + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_not_called() + + def test_refuses_active_run_owned_by_another_user( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + current = _current_resource(user="another-user") + plan = _plan(action=ApplyAction.UPDATE, current_resource=current) + api = Mock() + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: None) + + with pytest.raises(CLIError, match="owned by another user"): + BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(), + configurator_args=_configurator_args(), + ) + + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_not_called() + + def test_refuses_in_place_update(self, monkeypatch: pytest.MonkeyPatch) -> None: + current = _current_resource() + plan = _plan(action=ApplyAction.UPDATE, current_resource=current) + api = Mock() + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: "safe diff") + + with pytest.raises(CLIError, match="Refusing to change active run"): + BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(), + configurator_args=_configurator_args(), + ) + + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_not_called() + + @pytest.mark.parametrize("diff", [None, "unsafe diff"]) + def test_refuses_recreation_action( + self, + diff: str | None, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + current = _current_resource() + plan = _plan(action=ApplyAction.CREATE, current_resource=current) + api = Mock() + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: diff) + + with pytest.raises(CLIError, match="Refusing to change active run"): + BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(), + configurator_args=_configurator_args(), + ) + + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_not_called() + + def test_submits_when_run_is_absent(self) -> None: + plan = _plan(action=ApplyAction.CREATE, current_resource=None) + api = Mock() + applied = _applied_run() + api.runs.apply_plan.return_value = applied + + outcome = BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(), + configurator_args=_configurator_args(), + ) + + assert outcome == ApplyPlanOutcome( + result=ApplyPlanResult.SUBMITTED, + fence=RunApplyFence( + run_id=str(applied._run.id), + deployment_num=applied._run.deployment_num, + user=applied._run.user, + ), + ) + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_called_once() + + def test_submits_when_previous_run_is_terminal(self, monkeypatch: pytest.MonkeyPatch) -> None: + current = _current_resource(status=RunStatus.TERMINATED) + plan = _plan(action=ApplyAction.CREATE, current_resource=current) + api = Mock() + applied = _applied_run(deployment_num=current.deployment_num + 1) + api.runs.apply_plan.return_value = applied + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: "new spec") + + outcome = BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(), + configurator_args=_configurator_args(), + ) + + assert outcome == ApplyPlanOutcome( + result=ApplyPlanResult.SUBMITTED, + fence=RunApplyFence( + run_id=str(applied._run.id), + deployment_num=applied._run.deployment_num, + user=applied._run.user, + ), + ) + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_called_once() + + def test_rejects_unnamed_run(self) -> None: + plan = _plan(action=ApplyAction.CREATE, current_resource=None, run_name=None) + api = Mock() + + with pytest.raises(CLIError, match="requires a named run"): + BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(), + configurator_args=_configurator_args(), + ) + + api.runs.apply_plan.assert_not_called() + + +class TestApplyPlanOutcome: + def test_reports_cancelled_without_applying(self, monkeypatch: pytest.MonkeyPatch) -> None: + plan = _plan(action=ApplyAction.CREATE, current_resource=None) + api = Mock() + monkeypatch.setattr(run_configurator, "confirm_ask", lambda *_: False) + + outcome = BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(yes=False), + configurator_args=_configurator_args(), + ) + + assert outcome == ApplyPlanOutcome(result=ApplyPlanResult.CANCELLED) + api.runs.apply_plan.assert_not_called() + + def test_reports_normal_unchanged_apply_as_noop(self, monkeypatch: pytest.MonkeyPatch) -> None: + current = _current_resource() + plan = _plan(action=ApplyAction.UPDATE, current_resource=current) + api = Mock() + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: None) + + outcome = BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(no_recreate=False), + configurator_args=_configurator_args(), + ) + + assert outcome == ApplyPlanOutcome( + result=ApplyPlanResult.NOOP, + fence=RunApplyFence( + run_id=str(current.id), + deployment_num=current.deployment_num, + user=current.user, + ), + ) + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_not_called() + + def test_submitted_fence_comes_from_applied_run(self, monkeypatch: pytest.MonkeyPatch) -> None: + current = _current_resource() + plan = _plan(action=ApplyAction.UPDATE, current_resource=current) + api = Mock() + applied = _applied_run(deployment_num=current.deployment_num + 1) + api.runs.apply_plan.return_value = applied + monkeypatch.setattr(run_configurator, "render_run_spec_diff", lambda *_: "safe diff") + + outcome = BaseRunConfigurator(api).apply_plan( + run_plan=plan, + repo=Mock(), + command_args=_args(no_recreate=False), + configurator_args=_configurator_args(), + ) + + assert outcome == ApplyPlanOutcome( + result=ApplyPlanResult.SUBMITTED, + fence=RunApplyFence( + run_id=str(applied._run.id), + deployment_num=applied._run.deployment_num, + user=applied._run.user, + ), + ) + assert outcome.fence != RunApplyFence( + run_id=str(current.id), + deployment_num=current.deployment_num, + user=current.user, + ) + api.client.runs.stop.assert_not_called() + api.runs.apply_plan.assert_called_once() diff --git a/src/tests/_internal/server/routers/test_runs.py b/src/tests/_internal/server/routers/test_runs.py index 08207c944..eb7976268 100644 --- a/src/tests/_internal/server/routers/test_runs.py +++ b/src/tests/_internal/server/routers/test_runs.py @@ -3294,6 +3294,59 @@ async def test_submits_new_run_if_no_current_resource( job = res.scalar() assert job is not None + @pytest.mark.asyncio + async def test_rejects_apply_when_active_run_appears_after_absent_plan( + self, + session: AsyncSession, + client: AsyncClient, + ) -> None: + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, + project=project, + user=user, + project_role=ProjectRole.USER, + ) + repo = await create_repo(session=session, project_id=project.id) + run_spec = get_run_spec(run_name="test-run", repo_id=repo.name) + validate_run_spec_and_set_defaults(user, run_spec) + active_run = await create_run( + session=session, + project=project, + repo=repo, + user=user, + run_name=run_spec.run_name, + run_spec=run_spec, + status=RunStatus.RUNNING, + ) + original_id = active_run.id + original_deployment_num = active_run.deployment_num + + response = await client.post( + f"/api/project/{project.name}/runs/apply", + headers=get_auth_headers(user.token), + json=json.loads( + ApplyRunPlanRequest( + plan=ApplyRunPlanInput( + run_spec=run_spec, + # The plan was computed while the named run was absent. + current_resource=None, + ), + force=False, + ).model_dump_json() + ), + ) + + assert response.status_code == 400 + assert "Resource has been changed" in response.text + runs = (await session.execute(select(RunModel))).scalars().all() + assert len(runs) == 1 + assert runs[0].id == original_id + assert runs[0].deployment_num == original_deployment_num + assert runs[0].status == RunStatus.RUNNING + assert not runs[0].deleted + @pytest.mark.asyncio @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) async def test_submits_new_run_docker_true(