"""Backup and restore, end to end on real files and real archives.

Docker is faked (containers, volumes, the helper container); everything that
touches the stack directory and the archive runs for real. The cases are the
ways a backup used to fail to be one:

* data folders inside the stack directory were archived while the stack kept
  writing to them (it was only stopped when named volumes were selected);
* the stack came back with ``up -d`` — starting services that had been
  stopped on purpose and applying undeployed compose edits;
* a restore with "overwrite" deleted the whole stack directory, including the
  folders the backup had left out (too large, deselected, NFS);
* a truncated archive was only noticed after the old stack was gone;
* a restore under a new name overwrote the original stack's volumes;
* retention for stack "app" deleted the backups of stack "app-db";
* a backup with a failed volume was reported as "ok".
"""
from __future__ import annotations

import asyncio
import gzip
import io
import json
import os
import tarfile
import uuid

import pytest

from services import backup_service, compose_service, stack_assets_service as assets
from services.compose_service import ContainerInfo


# --------------------------------------------------------------------------- #
# Fakes
# --------------------------------------------------------------------------- #


class FakeDocker:
    """Records what the backup asked Docker/compose to do."""

    def __init__(self):
        self.containers: list[ContainerInfo] = []
        self.compose_calls: list[list[str]] = []
        self.compose_fail: set[str] = set()
        self.volumes: dict[str, dict] = {}  # name -> {"files": {...}, "labels", ...}
        self.running_users: dict[str, list[str]] = {}
        self.imported: list[dict] = []
        self.export_fail: set[str] = set()

    def container(self, service: str, state: str = "running") -> ContainerInfo:
        c = ContainerInfo(
            id=f"id-{service}", name=f"x-{service}-1", service=service,
            image="img", state=state, status=state,
        )
        self.containers.append(c)
        return c

    # compose_service
    def containers_for_stack(self, stack_id):
        return list(self.containers)

    async def run_compose(self, stack_id, args, override=None, timeout=600.0):
        self.compose_calls.append(list(args))
        failed = args[0] in self.compose_fail
        if not failed and args[0] == "stop":
            for c in self.containers:
                if c.service in args[1:]:
                    c.state = "exited"
        if not failed and args[0] in ("start", "up"):
            for c in self.containers:
                if args[0] == "up" or c.service in args[1:]:
                    c.state = "running"
        return {
            "returncode": 1 if failed else 0,
            "stdout": "",
            "stderr": f"{args[0]} failed" if failed else "",
            "command": " ".join(args),
        }

    # stack_assets_service
    def export_volume(self, name, dest_file):
        if name in self.export_fail:
            raise RuntimeError("helper container could not mount the volume")
        with tarfile.open(dest_file, "w") as tar:
            for rel, content in self.volumes[name]["files"].items():
                info = tarfile.TarInfo(rel)
                info.size = len(content)
                tar.addfile(info, io.BytesIO(content))
        return os.path.getsize(dest_file)

    def import_volume(self, name, labels, src_file, wipe=True, driver=None, driver_opts=None):
        with tarfile.open(src_file) as tar:
            files = {m.name: tar.extractfile(m).read() for m in tar.getmembers() if m.isreg()}
        self.imported.append(
            {"name": name, "labels": labels, "wipe": wipe, "driver": driver,
             "driver_opts": driver_opts, "files": files}
        )
        self.volumes[name] = {"files": files, "labels": labels}

    def volume_attrs(self, name):
        vol = self.volumes.get(name)
        return None if vol is None else {"Labels": vol.get("labels") or {}}


@pytest.fixture
def docker(monkeypatch, db):
    fake = FakeDocker()
    monkeypatch.setattr(compose_service, "containers_for_stack", fake.containers_for_stack)
    monkeypatch.setattr(compose_service, "run_compose", fake.run_compose)
    monkeypatch.setattr(assets, "export_volume", fake.export_volume)
    monkeypatch.setattr(assets, "import_volume", fake.import_volume)
    monkeypatch.setattr(assets, "volume_attrs", fake.volume_attrs)
    monkeypatch.setattr(assets, "running_users", lambda name: fake.running_users.get(name, []))

    def no_docker(ids, action):
        raise RuntimeError("docker unavailable")

    monkeypatch.setattr(backup_service, "_docker_each", no_docker)
    return fake


def _bind(source, directory, size=1024, include=True, reason=None, via="compose"):
    inside = source == directory or source.startswith(directory + "/")
    return {
        "source": source, "mounts": [{"service": "app", "target": "/data"}],
        "kind": "dir", "size": size, "inside_stack_dir": inside,
        "covered_by_compose": via == "compose", "via": via, "system": False,
        "remote": None, "include_default": include, "reason": reason,
    }


def _volume(name, stack_id, include=True, reason=None, external=False, remote=False):
    labels = {} if external else {
        assets.COMPOSE_PROJECT_LABEL: stack_id,
        assets.COMPOSE_VOLUME_LABEL: name.split("_", 1)[-1],
    }
    return {
        "name": name, "short": name.split("_", 1)[-1], "labels": labels,
        "driver": "local", "options": {"type": "none", "o": "bind", "device": "/srv/x"} if remote else {},
        "remote": remote, "remote_type": "nfs" if remote else None,
        "external": external, "bind_device": None,
        "include_default": include, "reason": reason,
    }


@pytest.fixture
def stack(monkeypatch, docker):
    """A stack laid out the way the user keeps them: compose, .env and data
    folders side by side in /opt/stacks/<stack>."""
    stack_id = f"s{uuid.uuid4().hex[:8]}"
    directory = compose_service.stack_dir(stack_id)
    os.makedirs(os.path.join(directory, "config"))
    os.makedirs(os.path.join(directory, "media", "movies"))
    _write(directory, "compose.yaml", "services:\n  app:\n    image: img\n")
    _write(directory, ".env", "PASSWORD=hunter2\n")
    _write(directory, "config/app.db", "state-at-backup")
    _write(directory, "media/movies/big.mkv", "x" * 100)

    inv = {
        "stack_id": stack_id, "stack_dir": directory, "stack_dir_visible": True,
        "path_mismatch": None,
        "binds": [
            _bind(os.path.join(directory, "config"), directory),
            # Too big for a backup: listed, not selected, left out of compose/.
            _bind(os.path.join(directory, "media"), directory, size=500 * 1024**3,
                  include=False, reason="larger than 2 GiB"),
        ],
        "volumes": [],
    }
    monkeypatch.setattr(assets, "inventory", lambda sid, max_bind_bytes=0: json.loads(json.dumps(inv)))
    return stack_id, directory, inv


def _write(directory, rel, content):
    path = os.path.join(directory, rel)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as fh:
        fh.write(content)


def _read(directory, rel):
    with open(os.path.join(directory, rel)) as fh:
        return fh.read()


def _names(archive):
    with tarfile.open(archive, "r:gz") as tar:
        return set(tar.getnames())


def run(coro):
    return asyncio.run(coro)


# --------------------------------------------------------------------------- #
# Backup
# --------------------------------------------------------------------------- #


def test_data_folder_in_stack_dir_stops_the_stack_and_starts_exactly_what_ran(stack, docker):
    stack_id, directory, _ = stack
    docker.container("app", "running")
    docker.container("worker", "exited")  # stopped on purpose

    path, report = run(backup_service.create_backup_ex(stack_id, "S"))
    try:
        assert docker.compose_calls == [["stop", "app"], ["start", "app"]]
        assert report["stopped"] == ["app"]
        assert report["restart_error"] is None and report["errors"] == []
        assert [c.state for c in docker.containers] == ["running", "exited"]
    finally:
        os.unlink(path)


def test_stopped_stack_is_not_started_by_a_backup(stack, docker):
    stack_id, _, _ = stack
    docker.container("app", "exited")

    path, _ = run(backup_service.create_backup_ex(stack_id, "S"))
    os.unlink(path)
    assert docker.compose_calls == []


def test_large_folder_stays_out_but_its_mount_point_stays_in(stack):
    stack_id, directory, _ = stack

    path, report = run(backup_service.create_backup_ex(stack_id, "S"))
    try:
        names = _names(path)
        manifest = backup_service.read_manifest(path)
    finally:
        os.unlink(path)
    assert {"compose/compose.yaml", "compose/.env", "compose/config/app.db"} <= names
    assert "compose/media" in names
    assert not any(n.startswith("compose/media/") for n in names)
    assert manifest["format_version"] == 3
    assert manifest["excluded"] == ["media"]
    assert manifest["compose_binds"] == ["config"]
    assert [b["source"] for b in report["binds"]] == [os.path.join(directory, "config")]


def test_stack_that_cannot_be_stopped_is_not_backed_up(stack, docker):
    stack_id, _, _ = stack
    docker.container("app", "running")
    docker.compose_fail.add("stop")

    with pytest.raises(backup_service.BackupError, match="consistent snapshot"):
        run(backup_service.create_backup_ex(stack_id, "S"))
    # Whatever did stop is started again.
    assert docker.compose_calls[-1] == ["start", "app"]


def test_restart_failure_is_reported_not_swallowed(stack, docker):
    stack_id, _, _ = stack
    docker.container("app", "running")
    docker.compose_fail.add("start")

    path, report = run(backup_service.create_backup_ex(stack_id, "S"))
    os.unlink(path)
    assert report["restart_error"]


def test_failed_volume_is_an_error_and_listed_as_skipped(stack, docker):
    stack_id, _, inv = stack
    inv["volumes"].append(_volume(f"{stack_id}_db", stack_id))
    docker.export_fail.add(f"{stack_id}_db")

    path, report = run(backup_service.create_backup_ex(stack_id, "S"))
    try:
        manifest = backup_service.read_manifest(path)
    finally:
        os.unlink(path)
    assert report["errors"] and f"{stack_id}_db" in report["errors"][0]
    assert manifest["volumes"] == []
    assert any(s.get("name") == f"{stack_id}_db" and "backup failed" in s["reason"]
               for s in manifest["skipped"])


def test_file_that_shrinks_while_read_keeps_the_archive_valid(tmp_path, monkeypatch):
    root = tmp_path / "stack"
    root.mkdir()
    log = root / "log.txt"
    log.write_bytes(b"a" * 5000)

    class RotatedMidRead:
        """The file is truncated after its size was taken, before it is read."""

        def __init__(self, fh):
            self._fh = fh
            self.name = fh.name

        def fileno(self):
            return self._fh.fileno()

        def read(self, n=-1):
            if log.stat().st_size > 100:
                os.truncate(log, 100)
            return self._fh.read(n)

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            self._fh.close()

    real_open = open

    def rotating_open(path, mode="r", *args, **kwargs):
        fh = real_open(path, mode, *args, **kwargs)
        return RotatedMidRead(fh) if str(path) == str(log) else fh

    monkeypatch.setattr("builtins.open", rotating_open)
    archive = tmp_path / "a.tar.gz"
    warnings: list[str] = []
    with tarfile.open(archive, "w:gz") as tar:
        backup_service._add_tree(tar, str(root), "compose", set(), warnings)
    monkeypatch.undo()

    backup_service.verify_archive(str(archive))
    assert warnings == ["log.txt: changed while it was read"]
    with tarfile.open(archive, "r:gz") as tar:
        data = tar.extractfile("compose/log.txt").read()
    assert data == b"a" * 100 + b"\0" * 4900


def test_backup_of_a_busy_stack_is_refused(stack):
    from sqlmodel import Session

    from database import engine
    from services import stack_lock_service

    stack_id, _, _ = stack
    with Session(engine) as session:
        stack_lock_service.acquire(session, stack_id, "update", "someone")
        try:
            with pytest.raises(backup_service.StackBusyError):
                run(backup_service.create_backup_ex(stack_id, "S"))
        finally:
            stack_lock_service.release(session, stack_id)


# --------------------------------------------------------------------------- #
# Restore
# --------------------------------------------------------------------------- #


def _backup(stack_id):
    path, _ = run(backup_service.create_backup_ex(stack_id, "S"))
    return path


def test_overwrite_restores_the_snapshot_and_keeps_what_was_left_out(stack, docker):
    stack_id, directory, _ = stack
    archive = _backup(stack_id)
    try:
        # Life goes on after the backup...
        _write(directory, "config/app.db", "state-after-backup")
        _write(directory, "config/wal-0002", "newer than the snapshot")
        _write(directory, "media/movies/new.mkv", "y")
        _write(directory, "compose.yaml", "services: {}\n")

        result = run(backup_service.restore(archive, overwrite=True))
    finally:
        os.unlink(archive)

    assert _read(directory, "config/app.db") == "state-at-backup"
    assert not os.path.exists(os.path.join(directory, "config/wal-0002"))
    assert _read(directory, "compose.yaml").startswith("services:\n  app:")
    assert _read(directory, ".env") == "PASSWORD=hunter2\n"
    # Not in the backup, so not the restore's to delete.
    assert _read(directory, "media/movies/big.mkv") == "x" * 100
    assert _read(directory, "media/movies/new.mkv") == "y"
    assert result["stack_id"] == stack_id
    # "media" was left out when the backup was made: listed, but not a problem.
    assert result["problems"] == []
    # Nothing staged is left behind or picked up as a stack.
    root = os.path.dirname(directory)
    assert not [n for n in os.listdir(root) if n.startswith(".restore-")]


def test_restore_stops_a_running_stack_and_brings_it_back_up(stack, docker):
    stack_id, _, _ = stack
    archive = _backup(stack_id)
    docker.container("app", "running")
    docker.compose_calls.clear()
    try:
        result = run(backup_service.restore(archive, overwrite=True))
    finally:
        os.unlink(archive)
    assert docker.compose_calls == [["stop", "app"], ["up", "-d", "--remove-orphans"]]
    assert result["restarted"] is True


def test_without_restore_binds_the_data_folders_are_kept(stack):
    stack_id, directory, _ = stack
    archive = _backup(stack_id)
    try:
        _write(directory, "config/app.db", "state-after-backup")
        _write(directory, "compose.yaml", "services: {}\n")
        run(backup_service.restore(archive, overwrite=True, restore_binds=False))
    finally:
        os.unlink(archive)
    assert _read(directory, "config/app.db") == "state-after-backup"
    assert _read(directory, "compose.yaml").startswith("services:\n  app:")


def test_restore_to_a_new_name_creates_the_stack(stack):
    stack_id, directory, _ = stack
    archive = _backup(stack_id)
    new_id = stack_id + "-copy"
    try:
        result = run(backup_service.restore(archive, target_id=new_id))
    finally:
        os.unlink(archive)
    new_dir = compose_service.stack_dir(new_id)
    assert result["stack_id"] == new_id
    assert _read(new_dir, "config/app.db") == "state-at-backup"
    assert os.path.isdir(os.path.join(new_dir, "media"))  # the mount point
    assert os.listdir(os.path.join(new_dir, "media")) == []


def test_existing_stack_without_overwrite_is_refused(stack):
    stack_id, _, _ = stack
    archive = _backup(stack_id)
    try:
        with pytest.raises(backup_service.BackupError, match="already exists"):
            run(backup_service.restore(archive))
    finally:
        os.unlink(archive)


def test_truncated_archive_is_refused_before_anything_is_touched(stack):
    stack_id, directory, _ = stack
    archive = _backup(stack_id)
    with open(archive, "rb") as fh:
        data = fh.read()
    with open(archive, "wb") as fh:
        fh.write(data[: len(data) // 2])
    _write(directory, "config/app.db", "current")
    try:
        with pytest.raises(backup_service.BackupError, match="damaged or incomplete"):
            run(backup_service.restore(archive, overwrite=True))
    finally:
        os.unlink(archive)
    assert _read(directory, "config/app.db") == "current"


def test_archive_cannot_write_through_its_own_symlink(tmp_path):
    outside = tmp_path / "outside.txt"
    outside.write_text("untouched")
    archive = tmp_path / "evil.tar.gz"
    with tarfile.open(archive, "w:gz") as tar:
        link = tarfile.TarInfo("compose/x")
        link.type = tarfile.SYMTYPE
        link.linkname = str(outside)
        tar.addfile(link)
        payload = b"pwned"
        info = tarfile.TarInfo("compose/x")
        info.size = len(payload)
        tar.addfile(info, io.BytesIO(payload))
    dest = tmp_path / "dest"
    with tarfile.open(archive, "r:gz") as tar:
        backup_service._extract_tree(tar, "compose/", str(dest))
    assert outside.read_text() == "untouched"
    assert (dest / "x").read_bytes() == b"pwned"


def test_volumes_restore_with_their_driver_and_options(stack, docker):
    stack_id, _, inv = stack
    vol = _volume(f"{stack_id}_db", stack_id)
    vol["options"] = {"type": "none", "o": "bind", "device": "/srv/db"}
    inv["volumes"].append(vol)
    docker.volumes[f"{stack_id}_db"] = {"files": {"PG_VERSION": b"16"}, "labels": vol["labels"]}
    archive = _backup(stack_id)
    del docker.volumes[f"{stack_id}_db"]
    try:
        result = run(backup_service.restore(archive, overwrite=True))
    finally:
        os.unlink(archive)
    assert result["volumes_restored"] == 1
    imported = docker.imported[0]
    assert imported["driver_opts"] == {"type": "none", "o": "bind", "device": "/srv/db"}
    assert imported["files"] == {"PG_VERSION": b"16"}


def test_renamed_restore_never_touches_the_originals_volumes(stack, docker):
    stack_id, _, inv = stack
    inv["volumes"] += [
        _volume(f"{stack_id}_db", stack_id),
        # explicit `name:` in compose, so not prefixed with the stack id
        {**_volume("shared_media", stack_id), "short": "media"},
    ]
    docker.volumes[f"{stack_id}_db"] = {"files": {"a": b"1"}, "labels": inv["volumes"][0]["labels"]}
    docker.volumes["shared_media"] = {"files": {"b": b"2"}, "labels": inv["volumes"][1]["labels"]}
    archive = _backup(stack_id)
    new_id = stack_id + "-test"
    try:
        result = run(backup_service.restore(archive, target_id=new_id))
    finally:
        os.unlink(archive)
    assert [i["name"] for i in docker.imported] == [f"{new_id}_db"]
    assert docker.imported[0]["labels"][assets.COMPOSE_PROJECT_LABEL] == new_id
    reasons = {s.get("name"): s["reason"] for s in result["skipped"] if s["kind"] == "volume"}
    assert "belongs to stack" in reasons["shared_media"]
    assert any(p.startswith("shared_media: belongs to stack") for p in result["problems"])
    assert docker.volumes["shared_media"]["files"] == {"b": b"2"}


def test_volume_in_use_by_another_stack_is_not_overwritten(stack, docker):
    stack_id, _, inv = stack
    inv["volumes"].append(_volume("common", stack_id, external=True))
    docker.volumes["common"] = {"files": {"a": b"1"}, "labels": {}}
    archive = _backup(stack_id)
    docker.running_users["common"] = ["other-app-1"]
    try:
        result = run(backup_service.restore(archive, overwrite=True))
    finally:
        os.unlink(archive)
    assert docker.imported == []
    assert any("other-app-1" in s["reason"] for s in result["skipped"] if s.get("name") == "common")


def test_external_volume_keeps_its_labels(stack, docker):
    stack_id, _, inv = stack
    inv["volumes"].append(_volume("common", stack_id, external=True))
    docker.volumes["common"] = {"files": {"a": b"1"}, "labels": {}}
    archive = _backup(stack_id)
    del docker.volumes["common"]
    try:
        run(backup_service.restore(archive, overwrite=True))
    finally:
        os.unlink(archive)
    assert docker.imported[0]["labels"] == {}


def test_version_2_backup_keeps_its_skipped_folders(stack, tmp_path):
    """Backups made before this change have no "excluded" list."""
    stack_id, directory, _ = stack
    archive = tmp_path / "v2.tar.gz"
    manifest = {
        "format_version": 2, "stack_id": stack_id, "name": "S", "stack_dir": directory,
        "volumes": [], "binds": [],
        "skipped": [{"kind": "bind", "source": os.path.join(directory, "media"),
                     "reason": "larger than 2 GiB"}],
    }
    with tarfile.open(archive, "w:gz") as tar:
        data = json.dumps(manifest).encode()
        info = tarfile.TarInfo("manifest.json")
        info.size = len(data)
        tar.addfile(info, io.BytesIO(data))
        body = b"services: {}\n"
        info = tarfile.TarInfo("compose/compose.yaml")
        info.size = len(body)
        tar.addfile(info, io.BytesIO(body))

    run(backup_service.restore(str(archive), overwrite=True))
    assert _read(directory, "compose.yaml") == "services: {}\n"
    assert _read(directory, "media/movies/big.mkv") == "x" * 100
    assert not os.path.exists(os.path.join(directory, "config"))


def test_verify_accepts_a_good_archive_and_rejects_garbage(stack, tmp_path):
    stack_id, _, _ = stack
    archive = _backup(stack_id)
    try:
        backup_service.verify_archive(archive)
    finally:
        os.unlink(archive)
    junk = tmp_path / "junk.tar.gz"
    junk.write_bytes(gzip.compress(b"not a tar"))
    with pytest.raises(backup_service.BackupError):
        backup_service.verify_archive(str(junk))


# --------------------------------------------------------------------------- #
# Retention (schedules: test_backup_schedules.py)
# --------------------------------------------------------------------------- #


def test_retention_pattern_is_exact():
    pattern = backup_service.backup_name_pattern("app")
    assert pattern.match("backup-app-full-20260918-030000.tar.gz")
    assert pattern.match("backup-app-config-20260918-030000.tar.gz")
    assert not pattern.match("backup-app-db-full-20260918-030000.tar.gz")
    assert not pattern.match("backup-app-full-20260918-030000.tar.gz.part")


def test_retention_leaves_other_stacks_alone(monkeypatch):
    from services import schedule_service

    names = [
        "backup-app-full-20260910-030000.tar.gz",
        "backup-app-full-20260911-030000.tar.gz",
        "backup-app-full-20260912-030000.tar.gz",
        "backup-app-db-full-20260901-030000.tar.gz",
    ]
    deleted: list[str] = []
    monkeypatch.setattr(
        schedule_service.dest_service, "list_backups",
        lambda dest: [{"name": n, "modified": i} for i, n in enumerate(names)],
    )
    monkeypatch.setattr(schedule_service.dest_service, "delete", lambda dest, n: deleted.append(n))

    assert schedule_service._prune(object(), "app", keep=2) == 1
    assert deleted == ["backup-app-full-20260910-030000.tar.gz"]


# --------------------------------------------------------------------------- #
# Routes
# --------------------------------------------------------------------------- #


def test_download_removes_its_temp_file(stack, as_admin, db):
    from sqlmodel import Session

    from models.stack import Stack

    stack_id, _, _ = stack
    with Session(db) as session:
        session.add(Stack(id=stack_id, name=stack_id))
        session.commit()
    created: list[str] = []
    real = backup_service.create_backup_ex

    async def spy(*args, **kwargs):
        path, report = await real(*args, **kwargs)
        created.append(path)
        return path, report

    import routers.backups as routes

    original = routes.backup_service.create_backup_ex
    routes.backup_service.create_backup_ex = spy
    try:
        r = as_admin.get(f"/api/stacks/{stack_id}/backup")
    finally:
        routes.backup_service.create_backup_ex = original
    assert r.status_code == 200
    header = json.loads(r.headers["X-Stackpilot-Backup"])
    assert header["errors"] == [] and header["restart_error"] is None
    assert created and not os.path.exists(created[0])


# --------------------------------------------------------------------------- #
# Inventory
# --------------------------------------------------------------------------- #


class _HelperClient:
    def __init__(self, output: str):
        self.output = output
        self.images = self
        self.containers = self

    def get(self, image):
        return object()

    def run(self, image, command, **kwargs):
        return self.output.encode()


def test_inspect_paths_flags_remote_and_unmeasured(monkeypatch):
    out = "\n".join([
        "0 dir 1024 ef53",       # local ext4, 1 MiB
        "1 dir - 6969",          # NFS: not walked
        "2 dir - ef53",          # du ran out of time
        "3 file 12 ff534d42",    # a file on CIFS
    ])
    monkeypatch.setattr(assets, "get_long_client", lambda: _HelperClient(out))
    res = assets.inspect_paths(["/a", "/b", "/c", "/d"])
    assert res["/a"] == {"kind": "dir", "size": 1024 * 1024, "fs": "ef53", "remote": None, "measured": True}
    assert res["/b"]["remote"] == "nfs" and res["/b"]["measured"] is True
    assert res["/c"]["measured"] is False
    assert res["/d"]["remote"] == "cifs"


def test_inventory_keeps_remote_and_unmeasured_folders_out(monkeypatch):
    stack_id = "inv"
    directory = compose_service.stack_dir(stack_id)
    specs = [
        {"source": "/mnt/nas/media", "service": "app", "target": "/media"},
        {"source": f"{directory}/cache", "service": "app", "target": "/cache"},
        {"source": f"{directory}/config", "service": "app", "target": "/config"},
    ]
    monkeypatch.setattr(assets, "_bind_specs_from_containers", lambda sid: specs)
    monkeypatch.setattr(assets, "_bind_specs_from_compose", lambda sid: [])
    monkeypatch.setattr(assets, "compose_volumes", lambda sid: [
        {**_volume(f"{stack_id}_db", stack_id), "bind_device": "/mnt/nas/db"},
        _volume("a" * 64, stack_id),
    ])
    monkeypatch.setattr(assets, "remote_mount_type", lambda path: None)
    monkeypatch.setattr(assets, "inspect_paths", lambda paths: {
        "/mnt/nas/media": {"kind": "dir", "size": None, "fs": "6969", "remote": "nfs", "measured": True},
        f"{directory}/cache": {"kind": "dir", "size": None, "fs": "ef53", "remote": None, "measured": False},
        f"{directory}/config": {"kind": "dir", "size": 4096, "fs": "ef53", "remote": None, "measured": True},
        "/mnt/nas/db": {"kind": "dir", "size": None, "fs": "6969", "remote": "nfs", "measured": True},
    })

    inv = assets.inventory(stack_id)

    binds = {b["source"]: b for b in inv["binds"]}
    assert binds["/mnt/nas/media"]["include_default"] is False
    assert binds["/mnt/nas/media"]["reason"] == "remote storage (nfs)"
    assert binds[f"{directory}/cache"]["reason"] == "too large to measure"
    assert binds[f"{directory}/config"]["include_default"] is True
    vols = {v["name"]: v for v in inv["volumes"]}
    assert vols[f"{stack_id}_db"]["include_default"] is False
    assert vols[f"{stack_id}_db"]["reason"] == "remote storage (nfs)"
    assert "anonymous" in vols["a" * 64]["reason"]


def test_external_volumes_are_found(monkeypatch):
    stack_id = f"s{uuid.uuid4().hex[:8]}"
    directory = compose_service.stack_dir(stack_id)
    os.makedirs(directory)
    _write(directory, "compose.yaml",
           "services:\n  app:\n    image: img\n    volumes: [shared:/data]\n"
           "volumes:\n  shared:\n    external: true\n    name: ${SHARED:-common}\n")

    class Vol:
        def __init__(self, name, attrs):
            self.name, self.attrs = name, attrs

    class Client:
        def __init__(self):
            self.volumes = self
            self.containers = self

        def list(self, **kwargs):
            if "all" in kwargs:  # containers
                return []
            return [Vol(f"{stack_id}_db", {"Labels": {assets.COMPOSE_PROJECT_LABEL: stack_id}})]

        def get(self, name):
            if name != "common":
                raise assets.DockerError("not_found")
            return Vol("common", {"Labels": {}, "Driver": "local"})

    client = Client()
    monkeypatch.setattr(assets, "get_client", lambda: client)

    vols = {v["name"]: v for v in assets.compose_volumes(stack_id)}
    assert set(vols) == {f"{stack_id}_db", "common"}
    assert vols["common"]["external"] is True
    assert vols[f"{stack_id}_db"]["external"] is False