@@ -1,17 +1,31 @@
 import signal
 import subprocess
 import sys
+from pathlib import Path
 from textwrap import dedent
 from unittest.mock import Mock
 
 import pytest
+from pytest import CaptureFixture
 
 import dstack._internal.cli.main as main_module
 from dstack._internal.compat import IS_WINDOWS
 from tests._internal.cli.common import run_dstack_cli
 
 
 class TestMain:
+    def test_prints_help_and_exists_with_0_exit_code(self, capsys: CaptureFixture):
+        exit_code = run_dstack_cli([])
+        assert exit_code == 0
+        assert capsys.readouterr().out.startswith("Usage: dstack [-h] [-v] COMMAND ...")
+
+    def test_prints_errors_to_stderr(self, tmp_path: Path, capsys: CaptureFixture):
+        exit_code = run_dstack_cli(["ps", "--nonsense"], home_dir=tmp_path)
+        assert exit_code == 1
+        captured = capsys.readouterr()
+        assert captured.out == ""
+        assert "Unrecognized arguments: --nonsense" in captured.err
+
     def test_reraises_keyboard_interrupt(self, monkeypatch: pytest.MonkeyPatch, capsys):
         monkeypatch.setattr(sys, "excepthook", sys.excepthook)
         monkeypatch.setattr(main_module, "check_for_updates", Mock(side_effect=KeyboardInterrupt))
 
@@ -4,6 +4,7 @@
 import subprocess
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
+from typing import Union, cast
 from unittest.mock import Mock, patch
 
 import gpuhunt
@@ -19,6 +20,7 @@
     _get_bootstrap_command,
     _get_image_env_and_cwd_command,
     _get_provider_ssh,
+    _get_volume_commands,
 )
 from dstack._internal.core.backends.lium.models import LiumConfig, LiumCreds
 from dstack._internal.core.errors import (
@@ -31,8 +33,10 @@
 from dstack._internal.core.models.common import RegistryAuth
 from dstack._internal.core.models.instances import InstanceAvailability, InstanceRuntime
 from dstack._internal.core.models.resources import ResourcesSpec
-from dstack._internal.core.models.runs import JobProvisioningData, Requirements
+from dstack._internal.core.models.runs import Job, JobProvisioningData, Requirements
 from dstack._internal.core.models.unix import UnixUser
+from dstack._internal.core.models.volumes import LiumVolumeConfiguration, VolumeMountPoint
+from dstack._internal.server.testing.common import get_volume
 from dstack._internal.utils.crypto import generate_rsa_key_pair_bytes
 
 COMPUTE_MODULE = "dstack._internal.core.backends.lium.compute"
@@ -676,3 +680,94 @@ def test_oversized_environment_fails_without_starting_partial_runner(self, tmp_p
 
         assert result.returncode != 0
         assert result.stdout == b""
+
+
+class TestVolumeLifecycle:
+    def test_create_unbounded_volume(self, compute):
+        compute.api_client.create_volume.return_value = {"id": "volume-id"}
+        with patch(f"{COMPUTE_MODULE}.generate_unique_volume_name", return_value="unique-name"):
+            data = compute.create_volume(Mock())
+
+        compute.api_client.create_volume.assert_called_once_with("unique-name")
+        assert data.volume_id == "volume-id"
+        assert data.size_gb is None
+        assert data.price is None
+        assert not data.attachable
+        assert not data.detachable
+
+    def test_recovers_lost_create_response_by_unique_name(self, compute):
+        compute.api_client.create_volume.side_effect = LiumAPIError("timed out")
+        compute.api_client.get_volumes.return_value = [
+            {"id": "other-id", "name": "other"},
+            {"id": "volume-id", "name": "unique-name"},
+        ]
+        with patch(f"{COMPUTE_MODULE}.generate_unique_volume_name", return_value="unique-name"):
+            data = compute.create_volume(Mock())
+
+        assert data.volume_id == "volume-id"
+        assert compute.api_client.create_volume.call_count == 1
+
+    def test_register_existing_volume(self, compute):
+        volume = SimpleNamespace(configuration=LiumVolumeConfiguration(volume_id="existing"))
+        compute.api_client.get_volume.return_value = {"id": "existing"}
+
+        data = compute.register_volume(volume)
+
+        assert data.volume_id == "existing"
+        compute.api_client.create_volume.assert_not_called()
+
+    def test_register_missing_volume(self, compute):
+        compute.api_client.get_volume.return_value = None
+        with pytest.raises(ComputeError, match="not found"):
+            compute.register_volume(
+                SimpleNamespace(configuration=LiumVolumeConfiguration(volume_id="missing"))
+            )
+
+    def test_delete_volume(self, compute):
+        compute.delete_volume(SimpleNamespace(volume_id="volume-id"))
+        compute.api_client.delete_volume.assert_called_once_with("volume-id")
+
+
+class TestVolumeMount:
+    @pytest.mark.parametrize(
+        "user",
+        [None, UnixUser(uid=10001), UnixUser(username="app")],
+    )
+    def test_custom_path(self, user):
+        commands = _volume_commands(user=user, path="/checkpoints")
+
+        assert commands[0] == "grep -q ' /mnt .* - fuse.s3fs ' /proc/self/mountinfo"
+        assert commands[1:] == ["mkdir -p -- /checkpoints", "mount --bind /mnt /checkpoints"]
+
+    def test_native_mount_path_needs_no_bind(self):
+        assert len(_volume_commands(path="/mnt")) == 1
+
+    def test_quotes_mount_path(self):
+        commands = _volume_commands(path="/checkpoint dir; echo unsafe")
+        assert commands[-1] == "mount --bind /mnt '/checkpoint dir; echo unsafe'"
+
+    @pytest.mark.parametrize("path", ["/", "/root", "/root/.dstack-lium", "/proc/a", "/mnt/a"])
+    def test_rejects_paths_that_hide_bootstrap_or_nest_source(self, path):
+        with pytest.raises(ComputeError, match="cannot be mounted"):
+            _volume_commands(path=path)
+
+    def test_two_volumes_rejected(self):
+        with pytest.raises(ComputeError, match="only one"):
+            _get_volume_commands(Mock(), Mock(), [Mock(), Mock()])
+
+    def test_selects_volume_from_alternatives(self):
+        assert _volume_commands(name=["elsewhere", "cache"], path="/data")[-1] == (
+            "mount --bind /mnt /data"
+        )
+
+
+def _volume_commands(user=None, path="/checkpoints", name: Union[str, list[str]] = "cache"):
+    # A real `Job` needs a full `JobSpec`, while `_get_volume_commands()` only reads the user and
+    # the mount points
+    job = cast(
+        Job,
+        SimpleNamespace(
+            job_spec=SimpleNamespace(user=user, volumes=[VolumeMountPoint(name=name, path=path)])
+        ),
+    )
+    return _get_volume_commands(Mock(), job, [get_volume(name="cache")])
@@ -0,0 +1,23 @@
+from typing import Optional
+
+import pytest
+
+from dstack._internal.core.compatibility.gpus import get_list_gpus_excludes
+from dstack._internal.server.schemas.gpus import ListGpusRequest
+from dstack._internal.server.testing.common import get_run_spec
+
+
+class TestGetListGpusExcludes:
+    @pytest.mark.parametrize("reverse", [None, False, True])
+    def test_omits_disabled_reverse(self, reverse: Optional[bool]):
+        kwargs = {} if reverse is None else {"reverse": reverse}
+        request = ListGpusRequest.model_validate(
+            {"run_spec": get_run_spec(repo_id="test-repo"), **kwargs}
+        )
+
+        body = request.model_dump(exclude=get_list_gpus_excludes(request))
+
+        if reverse:
+            assert body["reverse"] is True
+        else:
+            assert "reverse" not in body
@@ -0,0 +1,23 @@
+from typing import Optional
+
+import pytest
+
+from dstack._internal.core.compatibility.runs import get_get_plan_excludes
+from dstack._internal.server.schemas.runs import GetRunPlanRequest
+from dstack._internal.server.testing.common import get_run_spec
+
+
+class TestGetGetPlanExcludes:
+    @pytest.mark.parametrize("reverse_offers", [None, False, True])
+    def test_omits_disabled_reverse_offers(self, reverse_offers: Optional[bool]):
+        kwargs = {} if reverse_offers is None else {"reverse_offers": reverse_offers}
+        request = GetRunPlanRequest.model_validate(
+            {"run_spec": get_run_spec(repo_id="test-repo"), **kwargs}
+        )
+
+        body = request.model_dump(exclude=get_get_plan_excludes(request))
+
+        if reverse_offers:
+            assert body["reverse_offers"] is True
+        else:
+            assert "reverse_offers" not in body