strukto-ai--mirage/python/tests/commands/builtin/generic_bind/builders/test_common.py

# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========

import pytest

from mirage.commands.builtin.generic_bind.adapter import CommandIO
from mirage.commands.builtin.generic_bind.builders.common import (
    dir_refusing_read, split_readable)
from mirage.types import FileStat, FileType, PathSpec


def _ops(missing: set[str],
         implicit_dirs: set[str] | None = None,
         explicit_dirs: set[str] | None = None) -> CommandIO:
    dirs = implicit_dirs or set()
    typed = explicit_dirs or set()

    async def stat(_accessor, path, _index):
        if path.virtual in missing or path.virtual in dirs:
            raise FileNotFoundError(path.virtual)
        if path.virtual in typed:
            return FileStat(name=path.virtual, type=FileType.DIRECTORY)
        return FileStat(name=path.virtual, size=0)

    async def readdir(_accessor, path, _index):
        target = path.virtual.rstrip("/") or "/"
        entries = [d for d in dirs if (d.rsplit("/", 1)[0] or "/") == target]
        if path.virtual in dirs:
            entries.append(path.virtual.rstrip("/") + "/child.txt")
        return entries

    async def read_stream(_accessor, _path, _index):
        yield b"data"

    async def unused(*_args):
        raise AssertionError("not used")

    return CommandIO(readdir=readdir,
                     read_bytes=unused,
                     read_stream=read_stream,
                     stat=stat,
                     is_mounted=lambda _a: True)


@pytest.mark.asyncio
async def test_split_readable_keeps_order_and_reports_missing():
    paths = [
        PathSpec.from_str_path("/m1.txt"),
        PathSpec.from_str_path("/f.txt"),
        PathSpec.from_str_path("/m2.txt"),
    ]
    good, err = await split_readable(_ops({"/m1.txt", "/m2.txt"}), None, paths,
                                     None, "cat")
    assert [p.virtual for p in good] == ["/f.txt"]
    assert err == (b"cat: /m1.txt: No such file or directory\n"
                   b"cat: /m2.txt: No such file or directory\n")


@pytest.mark.asyncio
async def test_split_readable_reports_implicit_dir_as_eisdir():
    ops = _ops(set(), implicit_dirs={"/sub"})
    good, err = await split_readable(ops, None,
                                     [PathSpec.from_str_path("/sub")], None,
                                     "cat")
    assert good == []
    assert err == b"cat: /sub: Is a directory\n"


@pytest.mark.asyncio
async def test_split_readable_ignores_fabricated_children():
    # Synthetic hierarchies (postgres schema level) answer a readdir of
    # any missing name with fabricated children; only the parent listing
    # decides, so the original ENOENT stands.

    async def stat(_accessor, path, _index):
        raise FileNotFoundError(path.virtual)

    async def readdir(_accessor, path, _index):
        target = path.virtual.rstrip("/") or "/"
        if target == "/":
            return ["/real.txt"]
        return [f"{target}/tables", f"{target}/views"]

    async def unused(*_args):
        raise AssertionError("not used")

    ops = CommandIO(readdir=readdir,
                    read_bytes=unused,
                    read_stream=unused,
                    stat=stat,
                    is_mounted=lambda _a: True)
    good, err = await split_readable(ops, None,
                                     [PathSpec.from_str_path("/nope.txt")],
                                     None, "cat")
    assert good == []
    assert err == b"cat: /nope.txt: No such file or directory\n"


@pytest.mark.asyncio
async def test_split_readable_probe_swallows_driver_errors():
    # A backend whose readdir raises a non-FS driver error for missing
    # names (lancedb: "Table ... was not found") must not leak it through
    # the probe; the original ENOENT stands.

    async def stat(_accessor, path, _index):
        raise FileNotFoundError(path.virtual)

    async def readdir(_accessor, path, _index):
        raise ValueError("Table 'nope.txt' was not found")

    async def unused(*_args):
        raise AssertionError("not used")

    ops = CommandIO(readdir=readdir,
                    read_bytes=unused,
                    read_stream=unused,
                    stat=stat,
                    is_mounted=lambda _a: True)
    good, err = await split_readable(ops, None,
                                     [PathSpec.from_str_path("/nope.txt")],
                                     None, "wc")
    assert good == []
    assert err == b"wc: /nope.txt: No such file or directory\n"


@pytest.mark.asyncio
async def test_split_readable_reports_stat_typed_dir_as_eisdir():
    ops = _ops(set(), explicit_dirs={"/sub"})
    good, err = await split_readable(ops, None,
                                     [PathSpec.from_str_path("/sub")], None,
                                     "head")
    assert good == []
    assert err == b"head: /sub: Is a directory\n"


@pytest.mark.asyncio
async def test_dir_refusing_read_raises_eisdir_for_dirs():
    ops = _ops(set(), implicit_dirs={"/sub"})
    read = dir_refusing_read(ops, None)
    with pytest.raises(IsADirectoryError):
        async for _ in read(None, PathSpec.from_str_path("/sub")):
            raise AssertionError("no data expected")


@pytest.mark.asyncio
async def test_dir_refusing_read_streams_files():
    ops = _ops(set())
    read = dir_refusing_read(ops, None)
    chunks = [c async for c in read(None, PathSpec.from_str_path("/f.txt"))]
    assert chunks == [b"data"]


@pytest.mark.asyncio
async def test_split_readable_all_good_no_stderr():
    paths = [PathSpec.from_str_path("/f.txt")]
    good, err = await split_readable(_ops(set()), None, paths, None, "head")
    assert [p.virtual for p in good] == ["/f.txt"]
    assert err == b""


@pytest.mark.asyncio
async def test_split_readable_propagates_non_fs_errors():

    async def stat(_accessor, _path, _index):
        raise RuntimeError("backend broke")

    async def unused(*_args):
        raise AssertionError("not used")

    ops = CommandIO(readdir=unused,
                    read_bytes=unused,
                    read_stream=unused,
                    stat=stat,
                    is_mounted=lambda _a: True)
    with pytest.raises(RuntimeError):
        await split_readable(ops, None, [PathSpec.from_str_path("/f.txt")],
                             None, "cat")