# ========= 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 asyncio
from functools import partial
from unittest.mock import AsyncMock, MagicMock

from mirage.io import IOResult
from mirage.io.stream import materialize
from mirage.policy import Policies
from mirage.shell import parse
from mirage.shell.barrier import BarrierPolicy, apply_barrier
from mirage.shell.job_table import JobTable
from mirage.types import MountMode, PathSpec
from mirage.workspace.cli.registry import CLIRegistry
from mirage.workspace.mount.namespace import Namespace
from mirage.workspace.node.execute_node import execute_node as _execute_node
from mirage.workspace.session import Session
from mirage.workspace.session.session import vars_from_env


def execute_node(dispatch, registry, *args, **kwargs):
    return _execute_node(dispatch, registry, Namespace(registry), *args,
                         **kwargs)


def _session(cwd="/", env=None):
    return Session(session_id="test", cwd=cwd, vars=vars_from_env(env or {}))


def _mock_dispatch():
    d = AsyncMock()
    d.return_value = (b"", IOResult())
    return d


async def _answer_script(script: bytes, op: str, *args, **kwargs):
    if op == "stat":
        return None, IOResult()
    if op == "readdir":
        return [], IOResult()
    return script, IOResult()


def _script_dispatch(script: bytes):
    """A dispatch serving one script file and nothing else.

    The loader stats a path before it reads it (a directory is EISDIR
    before any read), so a double answering every op with the script
    bytes would hand ``stat`` a bytes object. This one answers stat with
    a miss and readdir with an empty listing, which is what a keyed
    store says about a plain file's path, and serves the bytes on read.

    Args:
        script (bytes): the script text a read returns.
    """
    return AsyncMock(side_effect=partial(_answer_script, script))


async def _no_match_resolve_glob(scopes, prefix=""):
    # A backend holding nothing the patterns match: a resolved spec passes
    # through, a pattern spec resolves to no matches and the expander
    # reinstates the literal word (bash with nullglob off).
    return [s for s in scopes if not s.pattern]


def _mock_registry():
    mount = MagicMock()
    mount.prefix = "/data/"
    mount.mode = MountMode.EXEC
    mount.execute_cmd = AsyncMock(return_value=(b"ok\n", IOResult()))
    mount.resource = MagicMock()
    mount.resource.resolve_glob = _no_match_resolve_glob
    mount.spec_for = MagicMock(return_value=None)

    reg = MagicMock()
    reg.mount_for = MagicMock(return_value=mount)
    reg.try_mount_for = MagicMock(return_value=mount)
    reg.resolve_mount = AsyncMock(return_value=mount)
    # A bare MagicMock policies answers wants() truthy and then fails the
    # await inside the gates; an empty Policies is what production carries.
    reg.policies = Policies()
    # A bare MagicMock answers clis.get() with a truthy mock, which
    # would spuriously dispatch every command as an installed CLI.
    reg.clis = CLIRegistry()
    return reg, mount


async def _sort_execute_cmd(name,
                            paths,
                            texts,
                            flag_kwargs,
                            *,
                            stdin=None,
                            **kwargs):
    """Mock execute_cmd that sorts stdin for sort command."""
    if name == "sort" and stdin:
        data = stdin if isinstance(stdin, bytes) else b""
        lines = data.decode().strip().split("\n")
        lines.sort()
        return "\n".join(lines).encode() + b"\n", IOResult()
    return b"ok\n", IOResult()


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


async def _aexec(cmd, session=None, dispatch=None, registry=None, env=None):
    session = session or _session(env=env or {})
    dispatch = dispatch or _mock_dispatch()
    reg, mount = registry or _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse(cmd)

    stdout, io, exec_node = await execute_node(dispatch, reg, job_table,
                                               execute_fn, "agent-1", node,
                                               session)
    stdout = await apply_barrier(stdout, io, BarrierPolicy.VALUE)
    return stdout, io, exec_node, session, mount, dispatch


def _exec(cmd, session=None, dispatch=None, registry=None, env=None):
    return _run(
        _aexec(cmd,
               session=session,
               dispatch=dispatch,
               registry=registry,
               env=env))


# ── simple commands ─────────────────────────────


def test_true():
    stdout, io, _, _, _, _ = _exec("true")
    assert io.exit_code == 0


def test_false():
    stdout, io, _, _, _, _ = _exec("false")
    assert io.exit_code == 1


def test_command_dispatches_to_mount():
    stdout, io, _, _, mount, _ = _exec("cat /data/file.txt")
    mount.execute_cmd.assert_called_once()
    args = mount.execute_cmd.call_args
    assert args[0][0] == "cat"
    scopes = args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].resolved is True
    assert io.exit_code == 0


def test_trailing_comment_is_ignored_at_program_level():
    # Regression: tree-sitter-bash emits a `comment` sibling under `program`
    # for `cmd # text`. Previously this raised
    # "unsupported tree-sitter node type: comment".
    stdout, io, _, _, mount, _ = _exec(
        "cat /data/file.txt        # -l, -a clustered")
    mount.execute_cmd.assert_called_once()
    assert io.exit_code == 0


def test_standalone_comment_is_a_noop():
    # A line that's *only* a comment must execute cleanly with exit 0.
    _, io, _, _, mount, _ = _exec("# just a comment")
    mount.execute_cmd.assert_not_called()
    assert io.exit_code == 0


def test_comment_inside_compound_statement_is_skipped():
    # Comments can also appear inside `{ ... }` blocks. The compound_statement
    # named_children iterator must skip them rather than dispatching them.
    _, io, _, _, mount, _ = _exec(
        "{ cat /data/a.txt; # mid-block comment\ncat /data/b.txt; }")
    assert mount.execute_cmd.call_count == 2
    assert io.exit_code == 0


# ── export / unset / local ──────────────────────


def test_export_sets_env():
    _, io, _, session, _, _ = _exec("export FOO=bar")
    assert io.exit_code == 0
    assert session.env["FOO"] == "bar"


def test_export_multiple():
    _, _, _, session, _, _ = _exec("export A=1 B=2")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_unset_removes_env():
    _, _, _, session, _, _ = _exec("unset FOO", env={"FOO": "bar"})
    assert "FOO" not in session.env


def test_local_outside_a_function_is_refused():
    # GNU: `bash: line 1: local: can only be used in a function`, exit 1
    # and nothing stored. Storing it globally and exiting 0 is the
    # silent-accept this tier removes; `declare` is the spelling that is
    # legal here.
    _, io, _, session, _, _ = _exec("local X=hello")
    assert io.exit_code == 1
    assert "X" not in session.env


def test_declare_sets_env_at_top_level():
    _, _, _, session, _, _ = _exec("declare X=hello")
    assert session.env["X"] == "hello"


# ── cd ──────────────────────────────────────────


def test_cd_updates_cwd():
    dispatch = _mock_dispatch()
    stat = MagicMock()
    stat.type = "directory"
    dispatch.return_value = (stat, IOResult())

    _, io, _, session, _, _ = _exec("cd /data", dispatch=dispatch)
    assert io.exit_code == 0
    assert session.cwd == "/data"


# ── pipeline ────────────────────────────────────


def test_pipeline():
    _, io, exec_node, _, _, _ = _exec("echo a | grep b")
    assert exec_node.op == "|"
    assert len(exec_node.children) == 2
    assert io.exit_code == 0


# ── list (&&, ||) ───────────────────────────────


def test_and_success():
    stdout, io, exec_node, _, _, _ = _exec("true && true")
    assert io.exit_code == 0
    assert exec_node.op == "&&"


def test_and_short_circuit():
    _, io, exec_node, _, _, _ = _exec("false && true")
    assert io.exit_code == 1
    assert exec_node.op == "&&"


def test_or_success():
    _, io, exec_node, _, _, _ = _exec("true || false")
    assert io.exit_code == 0


def test_or_fallback():
    _, io, exec_node, _, _, _ = _exec("false || true")
    assert io.exit_code == 0
    assert exec_node.op == "||"


# ── redirect ────────────────────────────────────


def test_redirect_stdout():
    _, io, _, _, _, dispatch = _exec("echo hello > /out.txt")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/out.txt"
    assert io.exit_code == 0


# ── semicolons ──────────────────────────────────


def test_semicolons():
    _, _, _, session, _, _ = _exec("export A=1; export B=2")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


# ── if statement ────────────────────────────────


def test_if_true_branch():
    _, _, _, session, _, _ = _exec(
        "if true; then export R=yes; else export R=no; fi")
    assert session.env["R"] == "yes"


def test_if_false_branch():
    _, _, _, session, _, _ = _exec(
        "if false; then export R=yes; else export R=no; fi")
    assert session.env["R"] == "no"


def test_if_no_else():
    _, io, _, session, _, _ = _exec("if true; then export R=ok; fi")
    assert session.env["R"] == "ok"


# ── for loop ────────────────────────────────────


def test_for_loop():
    _, io, _, session, _, _ = _exec("for x in a b c; do export LAST=$x; done")
    assert session.env["LAST"] == "c"
    assert io.exit_code == 0


# ── while loop ──────────────────────────────────


def test_while_false_no_execute():
    _, io, _, session, _, _ = _exec("while false; do export RAN=yes; done")
    assert "RAN" not in session.env
    assert io.exit_code == 0


# ── case ────────────────────────────────────────


def test_case_match():
    _, _, _, session, _, _ = _exec(
        "case hello in hello) export M=yes;; world) export M=no;; esac")
    assert session.env["M"] == "yes"


def test_case_no_match():
    _, io, _, session, _, _ = _exec("case xyz in hello) export M=yes;; esac")
    assert "M" not in session.env
    assert io.exit_code == 0


# ── subshell ────────────────────────────────────


def test_subshell_isolates_env():
    session = _session(env={"X": "before"})
    _exec("(export X=inside)", session=session)
    assert session.env["X"] == "before"


def test_subshell_updates_status_between_commands():
    """$? inside ( ... ) must reflect the previous command in the body."""
    stdout, io, _, _, _, _ = _exec("(false; echo subshell=$?)")
    assert io.exit_code == 0
    assert stdout == b"subshell=1\n"


def test_subshell_status_after_true_then_false():
    stdout, io, _, _, _, _ = _exec("(true; false; echo s=$?)")
    assert io.exit_code == 0
    assert stdout == b"s=1\n"


def test_subshell_isolates_cwd():
    dispatch = _mock_dispatch()
    stat = MagicMock()
    stat.type = "directory"
    dispatch.return_value = (stat, IOResult())

    session = _session(cwd="/start")
    _exec("(cd /other)", session=session, dispatch=dispatch)
    assert session.cwd == "/start"


# ── function definition ────────────────────────


def test_function_definition():
    _, _, _, session, _, _ = _exec("myfunc() { echo hello; }")
    assert "myfunc" in session.functions


# ── negated command ─────────────────────────────


def test_negated_true():
    _, io, _, _, _, _ = _exec("! true")
    assert io.exit_code == 1


def test_negated_false():
    _, io, _, _, _, _ = _exec("! false")
    assert io.exit_code == 0


# ── set ─────────────────────────────────────────


def test_set_outputs_env():
    stdout, io, _, _, _, _ = _exec("set", env={"A": "1", "B": "2"})
    assert io.exit_code == 0
    assert stdout is not None
    text = stdout.decode() if isinstance(stdout, bytes) else ""
    assert "A=1" in text
    assert "B=2" in text


# ── printenv ────────────────────────────────────


def test_printenv_single():
    stdout, io, _, _, _, _ = _exec("printenv FOO", env={"FOO": "hello"})
    assert io.exit_code == 0
    assert stdout is not None
    assert b"hello" in stdout


def test_printenv_all():
    stdout, io, _, _, _, _ = _exec("printenv", env={"X": "1"})
    assert io.exit_code == 0
    assert stdout is not None
    assert b"X=1" in stdout


# ── whoami ──────────────────────────────────────


def test_whoami_errors_without_identity():
    # No agent ever claimed this workspace: GNU errors for a uid with
    # no passwd entry, and so do we.
    stdout, io, _, _, _, _ = _exec("whoami", env={})
    assert io.exit_code == 1
    assert io.stderr == b"whoami: cannot find name for user ID\n"


def test_whoami_ignores_env_user():
    # GNU whoami reports the effective user, never $USER.
    stdout, io, _, _, _, _ = _exec("whoami", env={"USER": "alice"})
    assert io.exit_code == 1
    assert stdout != b"alice\n"


# ── unsupported node errors gracefully ──────────


def test_unsupported_node_raises():
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()

    fake_node = MagicMock()
    fake_node.type = "some_unknown_type_xyz"
    fake_node.text = b"some_unknown_type_xyz"

    stdout, io, _ = _run(
        execute_node(dispatch, reg, job_table, execute_fn, "agent-1",
                     fake_node, session))
    assert stdout is None
    assert io.exit_code == 2
    assert (_run(materialize(io.stderr)) ==
            b"mirage: unsupported shell construct: some_unknown_type_xyz\n")


# ── read ────────────────────────────────────────


def test_read_from_bytes():
    session = _session()
    dispatch = _mock_dispatch()
    reg, mount = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("read VAR")

    _, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "agent-1",
                     node,
                     session,
                     stdin=b"hello world\n"))
    assert io.exit_code == 0
    assert session.env["VAR"] == "hello world"


# ── shift ───────────────────────────────────────


def test_shift():
    _, io, _, _, _, _ = _exec("shift")
    assert io.exit_code == 0


# ── trap ────────────────────────────────────────


def test_trap():
    _, io, _, _, _, _ = _exec("trap")
    assert io.exit_code == 0


# ── return ──────────────────────────────────────


def test_return_raises_inside_function_frame():
    from mirage.shell.call_stack import CallStack
    from mirage.workspace.executor.control import ReturnSignal

    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("return 42")
    cs = CallStack()
    cs.push([], function_name="f")

    try:
        _run(
            execute_node(dispatch, reg, job_table, execute_fn, "agent-1", node,
                         session, None, cs))
        assert False, "should have raised ReturnSignal"
    except ReturnSignal as e:
        assert e.exit_code == 42


def test_return_top_level_fails_and_continues():
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("return 42")

    _, io, _ = _run(
        execute_node(dispatch, reg, job_table, execute_fn, "agent-1", node,
                     session))
    assert io.exit_code == 2
    assert b"can only `return'" in io.stderr


# ── break / continue ───────────────────────────


def test_break_outside_loop_absorbed():
    """A stray top-level break is absorbed and execution continues,
    like bash (which clamps break to the actual loop depth)."""
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("break")

    _, io, _ = _run(
        execute_node(dispatch, reg, job_table, execute_fn, "agent-1", node,
                     session))
    assert io.exit_code == 0


def test_continue_outside_loop_absorbed():
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("continue")

    _, io, _ = _run(
        execute_node(dispatch, reg, job_table, execute_fn, "agent-1", node,
                     session))
    assert io.exit_code == 0


# ── eval ────────────────────────────────────────


def test_eval():
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("eval echo hello")

    _, io, _ = _run(
        execute_node(dispatch, reg, job_table, execute_fn, "agent-1", node,
                     session))
    execute_fn.assert_called_once()
    assert "echo hello" in execute_fn.call_args[0][0]


# ── source ──────────────────────────────────────


def test_source():
    dispatch = _script_dispatch(b"export X=1")
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("source /script.sh")

    _, io, _ = _run(
        execute_node(dispatch, reg, job_table, execute_fn, "agent-1", node,
                     session))
    execute_fn.assert_called_once()


# ═══════════════════════════════════════════════
# Complex / combined tests
# ═══════════════════════════════════════════════

# ── nested control flow ─────────────────────────


def test_if_inside_for():
    _, _, _, session, _, _ = _exec("for x in a b c; do "
                                   "if true; then export COUNT=yes; fi; "
                                   "done")
    assert session.env["COUNT"] == "yes"


def test_for_with_break_via_if():
    _, _, _, session, _, _ = _exec("for x in a b c; do "
                                   "export LAST=ran; "
                                   "if true; then break; fi; "
                                   "done")
    assert session.env["LAST"] == "ran"


def test_nested_if():
    _, _, _, session, _, _ = _exec("if true; then "
                                   "if true; then export DEEP=yes; fi; "
                                   "fi")
    assert session.env["DEEP"] == "yes"


def test_nested_if_outer_false():
    _, _, _, session, _, _ = _exec("if false; then "
                                   "if true; then export DEEP=yes; fi; "
                                   "fi")
    assert "DEEP" not in session.env


def test_elif_chain():
    _, _, _, session, _, _ = _exec("if false; then export R=a; "
                                   "elif false; then export R=b; "
                                   "elif true; then export R=c; "
                                   "else export R=d; fi")
    assert session.env["R"] == "c"


def test_for_inside_if():
    _, _, _, session, _, _ = _exec("if true; then "
                                   "for x in 1 2 3; do export N=loop; done; "
                                   "fi")
    assert session.env["N"] == "loop"


# ── pipelines + control flow ───────────────────


def test_pipeline_three_stages():
    _, _, exec_node, _, _, _ = _exec("echo a | grep b | sort")
    assert exec_node.op == "|"
    assert len(exec_node.children) == 3


def test_pipeline_in_if_condition():
    _, _, _, session, _, _ = _exec("if true | true; then export P=yes; fi")
    assert session.env["P"] == "yes"


# ── chained list operators ──────────────────────


def test_chained_and():
    _, _, _, session, _, _ = _exec("export A=1 && export B=2 && export C=3")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"
    assert session.env["C"] == "3"


def test_and_then_or():
    _, _, _, session, _, _ = _exec("true && export A=yes || export A=no")
    assert session.env["A"] == "yes"


def test_or_then_and():
    _, _, _, session, _, _ = _exec(
        "false || export A=fallback && export B=after")
    assert session.env["A"] == "fallback"
    assert session.env["B"] == "after"


def test_and_short_circuit_skips_rest():
    _, _, _, session, _, _ = _exec("false && export SKIP=yes")
    assert "SKIP" not in session.env


def test_or_short_circuit_skips_rest():
    _, _, _, session, _, _ = _exec("true || export SKIP=yes")
    assert "SKIP" not in session.env


# ── semicolons + control flow ──────────────────


def test_semicolons_with_if():
    _, _, _, session, _, _ = _exec("export A=1; "
                                   "if true; then export B=2; fi; "
                                   "export C=3")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"
    assert session.env["C"] == "3"


def test_many_semicolons():
    _, _, _, session, _, _ = _exec("export A=1; export B=2; export C=3; "
                                   "export D=4; export E=5")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"
    assert session.env["C"] == "3"
    assert session.env["D"] == "4"
    assert session.env["E"] == "5"


# ── subshell nesting ───────────────────────────


def test_nested_subshell():
    session = _session(env={"X": "outer"})
    _exec("(export X=mid; (export X=inner))", session=session)
    assert session.env["X"] == "outer"


def test_subshell_in_if():
    session = _session(env={"V": "before"})
    _exec("if true; then (export V=inside); fi", session=session)
    assert session.env["V"] == "before"


# ── redirect + pipeline ────────────────────────


def test_redirect_with_pipeline():
    _, io, _, _, _, dispatch = _exec("echo hello | grep hello > /out.txt")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/out.txt"


# ── case patterns ──────────────────────────────


def test_case_wildcard():
    _, _, _, session, _, _ = _exec("case hello in "
                                   "world) export M=no;; "
                                   "hel*) export M=yes;; "
                                   "esac")
    assert session.env["M"] == "yes"


def test_case_fall_through_first_match():
    _, _, _, session, _, _ = _exec("case abc in "
                                   "abc) export M=first;; "
                                   "abc) export M=second;; "
                                   "esac")
    assert session.env["M"] == "first"


# ── function define + call ─────────────────────


def test_function_then_call():
    _, _, _, session, _, _ = _exec("myfunc() { export CALLED=yes; }; myfunc")
    assert session.env["CALLED"] == "yes"


def test_function_override():
    _, _, _, session, _, _ = _exec("f() { export V=first; }; "
                                   "f() { export V=second; }; "
                                   "f")
    assert session.env["V"] == "second"


# ── negation + control flow ────────────────────


def test_negated_in_if():
    _, _, _, session, _, _ = _exec("if ! false; then export N=yes; fi")
    assert session.env["N"] == "yes"


def test_negated_true_in_if():
    _, _, _, session, _, _ = _exec(
        "if ! true; then export N=yes; else export N=no; fi")
    assert session.env["N"] == "no"


# ── while with break / continue ────────────────


def test_while_true_with_break():
    _, io, _, session, _, _ = _exec(
        "while true; do export RAN=yes; break; done")
    assert session.env["RAN"] == "yes"
    assert io.exit_code == 0


def test_while_cap_emits_warning():
    """When _MAX_WHILE is hit, stderr carries a clear warning."""
    _, io, _, _, _, _ = _exec("while true; do export X=$X.; done",
                              env={"X": ""})
    assert io.stderr is not None
    stderr_bytes = io.stderr if isinstance(io.stderr, bytes) else b""
    assert b"terminated after" in stderr_bytes
    assert b"iterations" in stderr_bytes


def test_for_continue_skips_body():
    _, _, _, session, _, _ = _exec("for x in a b c; do "
                                   "continue; "
                                   "export NEVER=yes; "
                                   "done")
    assert "NEVER" not in session.env


# ── mixed operators ────────────────────────────


def test_and_or_semicolon_mix():
    _, _, _, session, _, _ = _exec(
        "export A=1; true && export B=2; false || export C=3")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"
    assert session.env["C"] == "3"


def test_pipeline_and_list():
    _, io, exec_node, session, _, _ = _exec("true | true && export OK=yes")
    assert exec_node.op == "&&"
    assert io.exit_code == 0
    assert session.env["OK"] == "yes"


# ── deeply nested ──────────────────────────────


def test_for_with_if_and_subshell():
    session = _session(env={"OUTER": "keep"})
    _exec("for x in a b; do "
          "if true; then (export OUTER=nope); fi; "
          "done",
          session=session)
    assert session.env["OUTER"] == "keep"


def test_if_with_for_and_break():
    _, _, _, session, _, _ = _exec("if true; then "
                                   "for x in 1 2 3; do "
                                   "export ITER=yes; break; "
                                   "done; "
                                   "fi")
    assert session.env["ITER"] == "yes"


def test_nested_for():
    _, _, _, session, _, _ = _exec("for a in x y; do "
                                   "for b in 1 2; do "
                                   "export INNER=ran; "
                                   "done; done")
    assert session.env["INNER"] == "ran"


def test_case_inside_for():
    _, _, _, session, _, _ = _exec("for x in hello world; do "
                                   "case x in "
                                   "hello) export H=yes;; "
                                   "world) export W=yes;; "
                                   "esac; done")
    assert "H" not in session.env


def test_if_and_or_chain():
    _, _, _, session, _, _ = _exec("if true && true; then export R=yes; fi")
    assert session.env["R"] == "yes"


def test_if_or_chain():
    _, _, _, session, _, _ = _exec("if false || true; then export R=yes; fi")
    assert session.env["R"] == "yes"


def test_subshell_with_pipeline():
    _, _, exec_node, _, _, _ = _exec("(echo a | grep a)")
    assert exec_node.op == "|"


def test_function_with_if():
    _, _, _, session, _, _ = _exec(
        "check() { if true; then export OK=yes; fi; }; check")
    assert session.env["OK"] == "yes"


def test_function_with_for():
    _, _, _, session, _, _ = _exec(
        "loop() { for x in a b; do export L=ran; done; }; loop")
    assert session.env["L"] == "ran"


# ═══════════════════════════════════════════════
# PathSpec / expansion flow tests
# ═══════════════════════════════════════════════

# ── command: paths become PathSpec ────────────


def test_command_file_becomes_globscope():
    """cat /data/file.txt → PathSpec(resolved=True)."""
    _, _, _, _, mount, _ = _exec("cat /data/file.txt")
    mount.execute_cmd.assert_called_once()
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].resolved is True
    assert "file.txt" in scopes[0].virtual


def test_command_glob_becomes_globscope():
    """cat /data/*.txt → unresolved PathSpec passed to resource."""
    _, _, _, _, mount, _ = _exec("cat /data/*.txt")
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    s = scopes[0]
    assert isinstance(s, PathSpec)
    assert s.pattern == "*.txt"
    assert s.resolved is False
    assert "/data/" in s.directory


def test_command_question_glob():
    """cat /data/file?.txt → unresolved PathSpec passed to resource."""
    _, _, _, _, mount, _ = _exec("cat /data/file?.txt")
    scopes = mount.execute_cmd.call_args[0][1]
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].pattern == "file?.txt"
    assert scopes[0].resolved is False


def test_command_bracket_glob():
    """cat /data/file[0-9].txt → unresolved PathSpec passed to resource."""
    _, _, _, _, mount, _ = _exec("cat /data/file[0-9].txt")
    scopes = mount.execute_cmd.call_args[0][1]
    assert isinstance(scopes[0], PathSpec)
    assert "[0-9]" in scopes[0].pattern
    assert scopes[0].resolved is False


def test_command_text_stays_string():
    """grep pattern /data/file → 'pattern' is text, path is PathSpec."""
    _, _, _, _, mount, _ = _exec("grep pattern /data/file.txt")
    args = mount.execute_cmd.call_args
    assert args[0][0] == "grep"
    paths = args[0][1]
    texts = args[0][2]
    assert len(paths) == 1
    assert isinstance(paths[0], PathSpec)
    assert "pattern" in texts


def test_command_multiple_paths():
    """diff /data/a.txt /data/b.txt → two PathSpecs with correct originals."""
    _, _, _, _, mount, _ = _exec("diff /data/a.txt /data/b.txt")
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 2
    assert all(isinstance(s, PathSpec) for s in scopes)
    assert "a.txt" in scopes[0].virtual
    assert "b.txt" in scopes[1].virtual


def test_command_no_paths():
    """echo hello world → builtin, returns stdout."""
    stdout, io, _, _, _, _ = _exec("echo hello world")
    assert io.exit_code == 0
    assert stdout == b"hello world\n"


def test_command_flags_stay_text():
    """grep -rn pattern /data/f → flags are text, path is PathSpec."""
    _, _, _, _, mount, _ = _exec("grep -rn pattern /data/f.txt")
    paths = mount.execute_cmd.call_args[0][1]
    texts = mount.execute_cmd.call_args[0][2]
    assert len(paths) == 1
    assert isinstance(paths[0], PathSpec)
    assert "-rn" in texts
    assert "pattern" in texts


def test_execute_cmd_receives_three_positional_args():
    """execute_cmd is called with (cmd_name, paths, texts, flag_kwargs)."""
    _, _, _, _, mount, _ = _exec("cat /data/file.txt")
    args = mount.execute_cmd.call_args[0]
    assert len(args) == 4
    assert args[0] == "cat"
    assert isinstance(args[1], list)
    assert isinstance(args[2], list)
    assert isinstance(args[3], dict)


def test_flag_kwargs_is_empty_without_spec():
    """Without a spec, flag_kwargs should be empty dict."""
    _, _, _, _, mount, _ = _exec("cat /data/file.txt")
    flag_kwargs = mount.execute_cmd.call_args[0][3]
    assert flag_kwargs == {}


def test_texts_separated_from_paths():
    """grep pattern /data/file → texts=['pattern'], paths=[PathSpec]."""
    _, _, _, _, mount, _ = _exec("grep pattern /data/file.txt")
    paths = mount.execute_cmd.call_args[0][1]
    texts = mount.execute_cmd.call_args[0][2]
    flag_kwargs = mount.execute_cmd.call_args[0][3]
    assert len(paths) == 1
    assert isinstance(paths[0], PathSpec)
    assert texts == ["pattern"]
    assert flag_kwargs == {}


def test_flags_and_texts_separated():
    """grep -rn pattern /data/f → flags in texts (no spec), paths separate."""
    _, _, _, _, mount, _ = _exec("grep -rn pattern /data/f.txt")
    paths = mount.execute_cmd.call_args[0][1]
    texts = mount.execute_cmd.call_args[0][2]
    assert len(paths) == 1
    assert "-rn" in texts
    assert "pattern" in texts


# ── variable expansion → classify ──────────────


def test_var_expands_to_file():
    """cat $FILE → PathSpec(resolved=True)."""
    _, _, _, _, mount, _ = _exec("cat $FILE", env={"FILE": "/data/x.txt"})
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].resolved is True
    assert "x.txt" in scopes[0].virtual


def test_var_expands_to_glob():
    """cat $P → unresolved glob PathSpec passed to resource."""
    _, _, _, _, mount, _ = _exec("cat $P", env={"P": "/data/*.csv"})
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].pattern == "*.csv"
    assert scopes[0].resolved is False


def test_var_expands_to_text():
    """echo $MSG → builtin, MSG expanded."""
    stdout, io, _, _, _, _ = _exec("echo $MSG", env={"MSG": "hello"})
    assert io.exit_code == 0
    assert stdout == b"hello\n"


def test_concatenation_var_path():
    """cat $DIR/file.txt → expanded + classified as PathSpec."""
    _, _, _, _, mount, _ = _exec("cat $DIR/file.txt", env={"DIR": "/data"})
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].resolved is True


def test_concatenation_var_glob():
    """cat $DIR/*.csv → unresolved glob PathSpec passed to resource."""
    _, _, _, _, mount, _ = _exec("cat $DIR/*.csv", env={"DIR": "/data"})
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].pattern == "*.csv"
    assert scopes[0].resolved is False


# ── export / local expansion ──────────────────


def test_export_expands_var_value():
    """export DIR=$BASE/sub → DIR=/data/sub."""
    _, _, _, session, _, _ = _exec("export DIR=$BASE/sub",
                                   env={"BASE": "/data"})
    assert session.env["DIR"] == "/data/sub"


def test_export_multiple_with_expansion():
    """export A=$X B=$Y → both expanded."""
    _, _, _, session, _, _ = _exec("export A=$X B=$Y",
                                   env={
                                       "X": "1",
                                       "Y": "2"
                                   })
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_declare_expands_value():
    """declare V=$BASE/file → V=/data/file.

    Spelled `declare` rather than `local` because `local` outside a
    function is refused, as GNU refuses it.
    """
    _, _, _, session, _, _ = _exec("declare V=$BASE/file",
                                   env={"BASE": "/data"})
    assert session.env["V"] == "/data/file"


# ── for loop expansion + classify ──────────────


def test_for_literal_values():
    """for x in a b c → iterates plain strings."""
    _, _, _, session, _, _ = _exec("for x in a b c; do export LAST=$x; done")
    assert session.env["LAST"] == "c"


def test_for_var_expansion():
    """for f in $A $B → expanded values."""
    _, _, _, session, _, _ = _exec("for f in $A $B; do export LAST=$f; done",
                                   env={
                                       "A": "first",
                                       "B": "second"
                                   })
    assert session.env["LAST"] == "second"


def test_for_path_becomes_globscope():
    """for f in /data/a.txt /data/b.txt → iterates PathSpec paths."""
    _, _, _, session, _, _ = _exec(
        "for f in /data/a.txt /data/b.txt; do export LAST=$f; done")
    assert session.env["LAST"] == "/data/b.txt"


def test_for_mixed_paths_and_text():
    """for f in /data/a.txt hello /data/b.txt → last is /data/b.txt."""
    _, _, _, session, _, _ = _exec(
        "for f in /data/a.txt hello /data/b.txt; do "
        "export N=$f; done")
    assert session.env["N"] == "/data/b.txt"


def test_for_glob_values():
    """for f in /data/*.csv → glob PathSpec, original preserved."""
    _, _, _, session, _, _ = _exec(
        "for f in /data/*.csv; do export LAST=$f; done")
    assert session.env["LAST"] == "/data/*.csv"


# ── case word expansion ────────────────────────


def test_case_var_match():
    """case $X in hello) → matches when X=hello."""
    _, _, _, session, _, _ = _exec("case $X in hello) export M=yes;; esac",
                                   env={"X": "hello"})
    assert session.env["M"] == "yes"


def test_case_var_no_match():
    """case $X in hello) → no match when X=world."""
    _, _, _, session, _, _ = _exec("case $X in hello) export M=yes;; esac",
                                   env={"X": "world"})
    assert "M" not in session.env


def test_case_var_wildcard():
    """case $X in hel*) → wildcard matches."""
    _, _, _, session, _, _ = _exec("case $X in hel*) export M=yes;; esac",
                                   env={"X": "hello"})
    assert session.env["M"] == "yes"


# ── redirect target expansion ──────────────────


def test_redirect_static_path():
    """echo hello > /data/out.txt → tee dispatched to /data/out.txt."""
    _, _, _, _, _, dispatch = _exec("echo hello > /data/out.txt")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) > 0
    assert write_calls[0][0][1].virtual == "/data/out.txt"


def test_redirect_var_target():
    """echo hello > $OUT → expanded to /data/out.txt."""
    _, _, _, _, _, dispatch = _exec("echo hello > $OUT",
                                    env={"OUT": "/data/out.txt"})
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) > 0
    assert write_calls[0][0][1].virtual == "/data/out.txt"


def test_redirect_concat_target():
    """echo hello > $DIR/out.txt → /data/out.txt."""
    _, _, _, _, _, dispatch = _exec("echo hello > $DIR/out.txt",
                                    env={"DIR": "/data"})
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) > 0
    assert write_calls[0][0][1].virtual == "/data/out.txt"


def test_redirect_append():
    """echo hello >> /data/out.txt → dispatch cat then tee."""
    _, _, _, _, _, dispatch = _exec("echo hello >> /data/out.txt")
    ops = [c[0][0] for c in dispatch.call_args_list]
    assert "write" in ops


def test_redirect_stdin():
    """sort < /data/input.txt → dispatch cat for input."""
    _, _, _, _, _, dispatch = _exec("sort < /data/input.txt")
    read_calls = [c for c in dispatch.call_args_list if c[0][0] == "read"]
    assert len(read_calls) > 0
    assert read_calls[0][0][1].virtual == "/data/input.txt"


# ── cd expansion ───────────────────────────────


def test_cd_var_expansion():
    """cd $DIR → cwd=/data."""
    dispatch = _mock_dispatch()
    stat = MagicMock()
    stat.type = "directory"
    dispatch.return_value = (stat, IOResult())
    _, io, _, session, _, _ = _exec("cd $DIR",
                                    dispatch=dispatch,
                                    env={"DIR": "/data"})
    assert io.exit_code == 0
    assert session.cwd == "/data"


def test_cd_concat_expansion():
    """cd $BASE/sub → cwd=/data/sub."""
    dispatch = _mock_dispatch()
    stat = MagicMock()
    stat.type = "directory"
    dispatch.return_value = (stat, IOResult())
    _, io, _, session, _, _ = _exec("cd $BASE/sub",
                                    dispatch=dispatch,
                                    env={"BASE": "/data"})
    assert io.exit_code == 0
    assert session.cwd == "/data/sub"


# ── source expansion ──────────────────────────


def test_source_var_expansion():
    """source $SCRIPT → dispatches cat on expanded path."""
    dispatch = _script_dispatch(b"export X=1")
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session(env={"SCRIPT": "/data/init.sh"})
    node = parse("source $SCRIPT")
    _run(execute_node(dispatch, reg, job_table, execute_fn, "a", node,
                      session))
    read_calls = [c for c in dispatch.call_args_list if c[0][0] == "read"]
    assert len(read_calls) > 0
    assert read_calls[0][0][1].virtual == "/data/init.sh"


# ── test command expansion ─────────────────────


def test_test_command_string_eq():
    """[ hello = hello ] → exit 0."""
    _, io, _, _, _, _ = _exec("test hello = hello")
    assert io.exit_code == 0


def test_test_command_string_neq():
    """[ hello = world ] → exit 1."""
    _, io, _, _, _, _ = _exec("test hello = world")
    assert io.exit_code == 1


# ── variable assignment expansion ──────────────


def test_bare_assignment_expands():
    """A=$B where B=hello → A=hello."""
    _, _, _, session, _, _ = _exec("A=$B", env={"B": "hello"})
    assert session.env["A"] == "hello"


def test_bare_assignment_concat():
    """A=$B/file where B=/data → A=/data/file."""
    _, _, _, session, _, _ = _exec("A=$B/file", env={"B": "/data"})
    assert session.env["A"] == "/data/file"


# ── arithmetic in commands ─────────────────────


def test_arithmetic_in_command():
    """echo $((2+3)) → expanded to 5."""
    stdout, io, _, _, _, _ = _exec("echo $((2 + 3))")
    assert io.exit_code == 0
    assert stdout == b"5\n"


# ── compound_statement (brace group) ───────────


def test_brace_group():
    """{ export A=1; export B=2; } → both execute."""
    _, _, _, session, _, _ = _exec("{ export A=1; export B=2; }")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_brace_group_updates_status_between_commands():
    """$? inside { ... } must reflect the previous command in the group."""
    stdout, io, _, _, _, _ = _exec("{ false; echo group=$?; }")
    assert io.exit_code == 0
    assert stdout == b"group=1\n"


def test_brace_group_status_after_true_then_false():
    stdout, io, _, _, _, _ = _exec("{ true; false; echo mid=$?; }")
    assert io.exit_code == 0
    assert stdout == b"mid=1\n"


def test_brace_group_with_if():
    _, _, _, session, _, _ = _exec(
        "{ if true; then export A=1; fi; export B=2; }")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_brace_group_in_pipeline():
    _, io, exec_node, _, _, _ = _exec("{ echo a; echo b; } | cat")
    assert exec_node.op == "|"
    assert len(exec_node.children) == 2
    assert io.exit_code == 0


# ── complex: pipeline + redirect + expansion ───


def test_pipeline_redirect_expansion():
    """cat $F | grep p > /data/out → tee to correct target."""
    _, _, _, _, _, dispatch = _exec("cat $F | grep p > /data/out",
                                    env={"F": "/data/input.txt"})
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/out"


def test_for_with_command_expansion():
    """for f in a b; do cat /data/$f.txt; done → 2 mount calls."""
    _, _, _, _, mount, _ = _exec("for f in a b; do cat /data/$f.txt; done")
    assert mount.execute_cmd.call_count == 2


def test_if_with_var_condition():
    """if [ $X = yes ]; then export R=ok; fi → expands $X."""
    _, _, _, session, _, _ = _exec("if true; then export R=$V; fi",
                                   env={"V": "expanded"})
    assert session.env["R"] == "expanded"


def test_nested_for_expansion():
    """Nested for — LAST set to last concatenated value."""
    _, _, _, session, _, _ = _exec("for a in x y; do "
                                   "for b in 1 2; do "
                                   "export LAST=$a$b; "
                                   "done; done")
    assert session.env["LAST"] == "y2"


# ═══════════════════════════════════════════════
# Comprehensive: all expansion + classify patterns
# ═══════════════════════════════════════════════

# ── for: glob + var + cmd sub ──────────────────


def test_for_glob_value_classified():
    """for f in /data/*.txt → PathSpec original as env value."""
    _, _, _, session, _, _ = _exec(
        "for f in /data/*.txt; do export LAST=$f; done")
    assert session.env["LAST"] == "/data/*.txt"


def test_for_var_expanded_to_path():
    """for f in $DIR → expanded to /data/sub, stored as path string."""
    _, _, _, session, _, _ = _exec("for f in $DIR; do export GOT=$f; done",
                                   env={"DIR": "/data/sub"})
    assert session.env["GOT"] == "/data/sub"


def test_for_cmd_sub_expanded():
    """for f in $(cmd) → cmd output split into values."""
    dispatch = _mock_dispatch()
    reg, mount = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock()
    io_result = IOResult()
    io_result.stdout = b"alpha\nbeta\ngamma\n"
    execute_fn.return_value = io_result
    session = _session()
    node = parse("for f in $(listcmd); do export LAST=$f; done")

    _run(execute_node(dispatch, reg, job_table, execute_fn, "a", node,
                      session))
    # $(listcmd) output split on \n → 3 iterations
    assert session.env["LAST"] == "gamma"


def test_for_mixed_glob_var_text():
    """for f in /s3/*.csv $DIR/file.txt hello → all classified correctly."""
    _, _, _, session, _, _ = _exec(
        "for f in /s3/*.csv $DIR/file.txt hello; do "
        "export LAST=$f; done",
        env={"DIR": "/data"})
    # 3 iterations: PathSpec, PathSpec, "hello"
    assert session.env["LAST"] == "hello"


# ── select: same as for ───────────────────────


def test_select_glob_value():
    """select f in /data/*.csv → PathSpec original as value."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "select f in /data/*.csv; do export GOT=$f; break; done", stdin=b"1\n")
    assert session.env["GOT"] == "/data/*.csv"


def test_select_var_expanded():
    """select f in $A $B → values expanded, choice 1 picks the first."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "select f in $A $B; do export LAST=$f; break; done",
        stdin=b"1\n",
        env={
            "A": "first",
            "B": "second"
        })
    assert session.env["LAST"] == "first"


# ── case: word expansion, pattern stays literal ─


def test_case_glob_pattern_matches():
    """case hello.txt in *.txt) → pattern is literal glob for fnmatch."""
    _, _, _, session, _, _ = _exec(
        "case hello.txt in *.txt) export M=yes;; esac")
    assert session.env["M"] == "yes"


def test_case_expanded_word_glob_pattern():
    """case $F in *.csv) → $F expanded, matched against pattern."""
    _, _, _, session, _, _ = _exec(
        "case $F in *.csv) export M=yes;; *.txt) export M=no;; esac",
        env={"F": "data.csv"})
    assert session.env["M"] == "yes"


def test_case_expanded_word_no_glob_match():
    """case $F in *.csv) → $F=data.txt doesn't match *.csv."""
    _, _, _, session, _, _ = _exec(
        "case $F in *.csv) export M=csv;; *.txt) export M=txt;; esac",
        env={"F": "data.txt"})
    assert session.env["M"] == "txt"


# ── command args: concat + glob + cmd sub ──────


def test_cmd_concat_var_glob():
    """cat $DIR/*.txt → unresolved glob PathSpec passed to resource."""
    _, _, _, _, mount, _ = _exec("cat $DIR/*.txt", env={"DIR": "/data"})
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].pattern == "*.txt"
    assert scopes[0].resolved is False


def test_cmd_concat_var_file():
    """cat $DIR/file.txt → /data/file.txt → PathSpec(resolved=True)."""
    _, _, _, _, mount, _ = _exec("cat $DIR/file.txt", env={"DIR": "/data"})
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert scopes[0].resolved is True
    assert "file.txt" in scopes[0].virtual


def test_cmd_cmd_sub_as_arg():
    """cat $(echo /data/file.txt) → cmd sub expanded → PathSpec."""
    dispatch = _mock_dispatch()
    reg, mount = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock()
    io_result = IOResult()
    io_result.stdout = b"/data/file.txt\n"
    execute_fn.return_value = io_result
    session = _session()
    node = parse("cat $(echo /data/file.txt)")

    _run(execute_node(dispatch, reg, job_table, execute_fn, "a", node,
                      session))
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 1
    assert isinstance(scopes[0], PathSpec)
    assert "file.txt" in scopes[0].virtual


def test_cmd_multiple_concat_paths():
    """diff $A/x.txt $B/y.txt → two PathSpecs with correct directories."""
    _, _, _, _, mount, _ = _exec("diff $A/x.txt $B/y.txt",
                                 env={
                                     "A": "/s3",
                                     "B": "/data"
                                 })
    scopes = mount.execute_cmd.call_args[0][1]
    assert len(scopes) == 2
    assert isinstance(scopes[0], PathSpec)
    assert isinstance(scopes[1], PathSpec)
    assert "/s3/" in scopes[0].directory
    assert "/data/" in scopes[1].directory


# ── assignment: cmd sub + concat ───────────────


def test_assign_cmd_sub():
    """VAR=$(cmd) → executes cmd, assigns output."""
    dispatch = _mock_dispatch()
    reg, mount = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock()
    io_result = IOResult()
    io_result.stdout = b"result_value\n"
    execute_fn.return_value = io_result
    session = _session()
    node = parse("VAR=$(echo result_value)")

    _run(execute_node(dispatch, reg, job_table, execute_fn, "a", node,
                      session))
    assert session.env["VAR"] == "result_value"


def test_assign_concat():
    """DIR=/data/$SUB → expanded to /data/files."""
    _, _, _, session, _, _ = _exec("DIR=/data/$SUB", env={"SUB": "files"})
    assert session.env["DIR"] == "/data/files"


def test_assign_nested_concat():
    """OUT=${BASE}/${SUB}/file.txt → fully expanded."""
    _, _, _, session, _, _ = _exec("OUT=${BASE}/${SUB}/file.txt",
                                   env={
                                       "BASE": "/data",
                                       "SUB": "reports"
                                   })
    assert session.env["OUT"] == "/data/reports/file.txt"


# ── redirect target: var + concat + cmd sub ────


def test_redirect_concat_var_target():
    """echo x > $DIR/out.txt → target expanded to /data/out.txt."""
    _, _, _, _, _, dispatch = _exec("echo x > $DIR/out.txt",
                                    env={"DIR": "/data"})
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/out.txt"


def test_redirect_cmd_sub_target():
    """echo x > $(echo /data/out.txt) → cmd sub expanded."""
    dispatch = _mock_dispatch()
    reg, mount = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock()
    io_result = IOResult()
    io_result.stdout = b"/data/out.txt\n"
    execute_fn.return_value = io_result
    session = _session()
    node = parse("echo x > $(echo /data/out.txt)")

    _run(execute_node(dispatch, reg, job_table, execute_fn, "a", node,
                      session))
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/out.txt"


def test_redirect_stderr_path():
    """cat missing 2> /data/err.log → stderr written to file."""
    dispatch = _mock_dispatch()
    reg, mount = _mock_registry()
    mount.execute_cmd = AsyncMock(side_effect=FileNotFoundError("missing"))
    mount.resolve_command = MagicMock(return_value=MagicMock())
    reg.mounts = MagicMock(return_value=[mount])
    stdout, io, _, _, _, dispatch = _exec(
        "cat /data/missing.txt 2> /data/err.log",
        dispatch=dispatch,
        registry=(reg, mount))
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/err.log"
    assert io.stderr is None


def test_redirect_append_var():
    """echo x >> $LOG → expanded, cat+tee dispatched."""
    _, _, _, _, _, dispatch = _exec("echo x >> $LOG",
                                    env={"LOG": "/data/app.log"})
    ops = [c[0][0] for c in dispatch.call_args_list]
    assert "read" in ops
    assert "write" in ops
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert write_calls[0][0][1].virtual == "/data/app.log"


def test_redirect_stdin_var():
    """sort < $INPUT → expanded, cat dispatched for input."""
    _, _, _, _, _, dispatch = _exec("sort < $INPUT",
                                    env={"INPUT": "/data/in.txt"})
    read_calls = [c for c in dispatch.call_args_list if c[0][0] == "read"]
    assert len(read_calls) == 1
    assert read_calls[0][0][1].virtual == "/data/in.txt"


# ── heredoc: no target expansion ───────────────


def test_heredoc_no_target_crash():
    """cat <<EOF\\nhello\\nEOF → dispatches cat, no crash."""
    _, io, _, _, mount, _ = _exec("cat <<EOF\nhello\nEOF")
    assert io.exit_code == 0
    mount.execute_cmd.assert_called_once()


# ── complex: full pipeline simulation ──────────


def test_full_pipeline_with_expansion():
    """cat $DIR/in.txt | grep $PAT > $DIR/out.txt → all expanded."""
    _, _, _, _, mount, dispatch = _exec(
        "cat $DIR/in.txt | grep $PAT > $DIR/out.txt",
        env={
            "DIR": "/data",
            "PAT": "error"
        })
    # cat should receive PathSpec for /data/in.txt
    # grep should receive "error" as text arg
    # redirect should tee to /data/out.txt
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/out.txt"


def test_for_with_redirect_expansion():
    """for f in a b; do echo $f > /data/$f.txt; done → 2 tee calls."""
    _, _, _, _, _, dispatch = _exec(
        "for f in a b; do echo $f > /data/$f.txt; done")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 2
    targets = {c[0][1].virtual for c in write_calls}
    assert "/data/a.txt" in targets
    assert "/data/b.txt" in targets


def test_nested_expansion_in_for_body():
    """for d in /s3 /data; do cat $d/file.txt; done → paths classified."""
    _, _, _, _, mount, _ = _exec(
        "for d in /s3 /data; do cat $d/file.txt; done")
    assert mount.execute_cmd.call_count == 2
    # Each call should have a PathSpec path
    for call in mount.execute_cmd.call_args_list:
        scopes = call[0][1]
        assert len(scopes) == 1
        assert isinstance(scopes[0], PathSpec)
        assert scopes[0].resolved is True


# ── multi-statement body (do_group) ────────────


def test_for_multi_statement_body():
    """for x in a b; do export V=$x; export W=done; done → both run."""
    _, _, _, session, _, _ = _exec(
        "for x in a b; do export V=$x; export W=done; done")
    assert session.env["V"] == "b"
    assert session.env["W"] == "done"


def test_for_break_in_multi_body():
    """for x in a b c; do export V=$x; break; done → breaks after first."""
    _, _, _, session, _, _ = _exec(
        "for x in a b c; do export V=$x; break; done")
    assert session.env["V"] == "a"


def test_for_continue_in_multi_body():
    """for x in a b c; do continue; export SKIP=yes; done → SKIP never set."""
    _, _, _, session, _, _ = _exec(
        "for x in a b c; do continue; export SKIP=yes; done")
    assert "SKIP" not in session.env


def test_for_break_after_condition():
    """Conditional break in multi-statement for body."""
    _, _, _, session, _, _ = _exec("for x in a b c; do "
                                   "export V=$x; "
                                   "if true; then break; fi; "
                                   "export AFTER=no; "
                                   "done")
    assert session.env["V"] == "a"
    assert "AFTER" not in session.env


def test_while_multi_statement_body():
    """while true; do export RAN=yes; break; done → both run, then break."""
    _, _, _, session, _, _ = _exec(
        "while true; do export RAN=yes; break; done")
    assert session.env["RAN"] == "yes"


def test_while_continue_in_multi_body():
    """while true; do export N=yes; continue; done → N set, hits MAX_WHILE."""
    _, io, _, session, _, _ = _exec(
        "while true; do export N=yes; continue; done")
    assert session.env["N"] == "yes"
    assert io.exit_code == 0


def test_select_break_in_multi_body():
    """select f in a b c → choice 2 sets the variable, break ends it."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "select f in a b c; do export V=$f; break; done", stdin=b"2\n")
    assert session.env["V"] == "b"


def test_for_multi_with_redirect():
    """Multi-statement for body with redirect."""
    _, _, _, session, _, dispatch = _exec(
        "for f in a b; do echo $f > /data/$f.txt; export DONE=yes; done")
    assert session.env["DONE"] == "yes"
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 2


def test_for_multi_with_pipeline():
    """Multi-statement for body with pipeline."""
    _, _, _, session, _, _ = _exec(
        "for f in a b; do echo $f | cat; export DONE=yes; done")
    assert session.env["DONE"] == "yes"


# ═══════════════════════════════════════════════
# stdin tests: bytes, AsyncIterator, None
# ═══════════════════════════════════════════════


async def _async_iter(data: bytes):
    """Create an async iterator yielding data in chunks."""
    for i in range(0, len(data), 4):
        yield data[i:i + 4]


def test_read_from_bytes_stdin():
    """read VAR with stdin=b'...' → VAR set."""
    session = _session()
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("read VAR")

    _, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=b"hello world\n"))
    assert io.exit_code == 0
    assert session.env["VAR"] == "hello world"


def test_read_from_async_iterator_stdin():
    """read VAR with stdin=AsyncIterator → VAR set from stream."""
    session = _session()
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("read VAR")

    stdin_stream = _async_iter(b"streamed data\n")
    _, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=stdin_stream))
    assert io.exit_code == 0
    assert session.env["VAR"] == "streamed data"


def test_read_from_none_stdin():
    """read VAR with stdin=None → VAR empty, exit code 1 (no input)."""
    session = _session()
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("read VAR")

    _, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=None))
    assert io.exit_code == 1
    assert session.env["VAR"] == ""


def test_read_multivar_from_bytes():
    """read A B C with stdin=b'x y z' → A=x B=y C=z."""
    session = _session()
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("read A B C")

    _, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=b"x y z"))
    assert session.env["A"] == "x"
    assert session.env["B"] == "y"
    assert session.env["C"] == "z"


def test_read_multivar_from_stream():
    """read A B with stdin=AsyncIterator → splits correctly."""
    session = _session()
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("read A B")

    stdin_stream = _async_iter(b"first second third")
    _, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=stdin_stream))
    assert session.env["A"] == "first"
    assert session.env["B"] == "second third"


def test_read_multiline_takes_first():
    """read VAR from multi-line input → takes first line only."""
    session = _session()
    dispatch = _mock_dispatch()
    reg, _ = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("read VAR")

    _, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=b"line1\nline2\nline3"))
    assert session.env["VAR"] == "line1"


def test_pipeline_passes_stdout_as_stdin():
    """echo hello | cat → cat receives echo's stdout."""
    _, _, exec_node, _, mount, _ = _exec("echo hello | cat")
    assert exec_node.op == "|"
    assert mount.execute_cmd.call_count == 1


def test_redirect_stdin_from_file():
    """sort < /data/input.txt → dispatch read_bytes, sort gets data."""
    dispatch = _mock_dispatch()
    dispatch.return_value = (b"line2\nline1\n", IOResult())
    reg, mount = _mock_registry()
    mount.execute_cmd = AsyncMock(side_effect=_sort_execute_cmd)
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("sort < /data/input.txt")

    stdout, io, _ = _run(
        execute_node(dispatch, reg, job_table, execute_fn, "a", node, session))
    read_calls = [c for c in dispatch.call_args_list if c[0][0] == "read"]
    assert len(read_calls) == 1
    assert read_calls[0][0][1].virtual == "/data/input.txt"
    assert io.exit_code == 0
    assert stdout == b"line1\nline2\n"


# ═══════════════════════════════════════════════
# stdin materialization in loops
# ═══════════════════════════════════════════════


def _exec_with_stdin(cmd, stdin, env=None):
    """Execute a command with explicit stdin."""

    async def _inner():
        session = _session(env=env or {})
        dispatch = _mock_dispatch()
        reg, mount = _mock_registry()
        job_table = JobTable()
        execute_fn = AsyncMock(return_value=IOResult())
        node = parse(cmd)
        stdout, io, exec_node = await execute_node(dispatch,
                                                   reg,
                                                   job_table,
                                                   execute_fn,
                                                   "a",
                                                   node,
                                                   session,
                                                   stdin=stdin)
        stdout = await apply_barrier(stdout, io, BarrierPolicy.VALUE)
        return stdout, io, exec_node, session, mount, dispatch

    return _run(_inner())


def test_for_stdin_bytes_shared_across_iterations():
    """for x in a b; do read V; done with bytes stdin.

    Each iteration reads from the materialized buffer.
    read consumes one line per call.
    """
    _, _, _, session, _, _ = _exec_with_stdin(
        "for x in a b; do read LINE; done", stdin=b"first\nsecond\n")
    # read takes first line, then second — both accessible
    assert "LINE" in session.env


def test_for_stdin_async_iterator_materialized():
    """for x in a b; do read V; done with async iterator.

    Iterator is materialized before loop, so all iterations
    can access the data.
    """

    async def _stream():
        yield b"alpha\n"
        yield b"beta\n"

    _, _, _, session, _, _ = _exec_with_stdin(
        "for x in a b; do read LINE; done", stdin=_stream())
    assert "LINE" in session.env


def test_while_read_consumes_lazily():
    """while read line; do ... loop pulls one line at a time.

    Producer should only yield as many chunks as needed for the lines
    actually consumed, not eagerly materialize the whole stream.
    """
    pulls = 0

    async def _lazy_stream():
        nonlocal pulls
        for i in range(1000):
            pulls += 1
            yield f"line{i}\n".encode()

    _, _, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do export LAST=$LINE; done; export N_PULLS=done",
        stdin=_lazy_stream())
    assert session.env.get("LAST") == "line999"
    assert pulls == 1000


def test_while_read_break_stops_pulling():
    """Early break should leave the producer un-pulled past break point.

    Proves laziness: with the old eager materialize, all 1000 chunks
    would be pulled before the loop body runs. With lazy buffering,
    only the consumed-up-to-break chunks get pulled.
    """
    pulls = 0

    async def _lazy_stream():
        nonlocal pulls
        for i in range(1000):
            pulls += 1
            yield f"line{i}\n".encode()

    _, _, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do "
        "  export LAST=$LINE; "
        "  if [ \"$LINE\" = \"line4\" ]; then break; fi; "
        "done",
        stdin=_lazy_stream())
    assert session.env.get("LAST") == "line4"
    assert pulls < 50, f"expected lazy pulls (~5), got {pulls}"


def test_for_stdin_none_no_crash():
    """for x in a b; do read V; done with stdin=None."""
    _, io, _, session, _, _ = _exec_with_stdin(
        "for x in a b; do read LINE; done", stdin=None)
    assert session.env.get("LINE", "") == ""


def test_while_stdin_bytes():
    """while loop materializes stdin before iterating."""
    _, _, _, session, _, _ = _exec_with_stdin("while false; do read V; done",
                                              stdin=b"data\n")
    # while false → body never executes, but no crash
    assert "V" not in session.env


def test_while_stdin_async_iterator():
    """while loop with async iterator stdin — materialized."""

    async def _stream():
        yield b"line1\n"
        yield b"line2\n"

    _, _, _, session, _, _ = _exec_with_stdin("while false; do read V; done",
                                              stdin=_stream())
    assert "V" not in session.env


def test_body_sequential_reads_advance_buffer():
    """read A; read B — each read consumes one line from buffer.

    First read sets A=hello, advances buffer.
    Second read sets B=world from remaining buffer.
    """
    _, _, _, session, mount, _ = _exec_with_stdin("read A; read B",
                                                  stdin=b"hello\nworld\n")
    assert session.env["A"] == "hello"
    assert session.env["B"] == "world"


def test_pipeline_stdin_flows_through():
    """echo data | cat → pipe connects stdout to stdin."""
    _, _, exec_node, _, mount, _ = _exec_with_stdin("echo data | cat",
                                                    stdin=None)
    assert exec_node.op == "|"
    assert mount.execute_cmd.call_count == 1


def test_select_stdin_materialized():
    """select reads its choice from stdin; a non-numeric reply sets
    the variable empty and lands raw in REPLY, like bash."""

    async def _stream():
        yield b"choice\n"

    _, _, _, session, _, _ = _exec_with_stdin(
        "select f in a b; do export GOT=$f; break; done", stdin=_stream())
    assert session.env["GOT"] == ""
    assert session.env["REPLY"] == "choice"


def test_for_with_cmd_using_stdin():
    """for x in a b; do cat; done with stdin."""
    _, _, _, _, mount, _ = _exec_with_stdin(
        "for x in a b; do cat /data/f.txt; done", stdin=b"piped data")
    assert mount.execute_cmd.call_count == 2


def test_subshell_stdin_passthrough():
    """(read V) with stdin → read gets stdin inside subshell."""
    _, _, _, session, _, _ = _exec_with_stdin("(read V)",
                                              stdin=b"subshell data")
    # subshell restores env, so V is lost — but no crash
    assert "V" not in session.env


def test_redirect_stdin_with_async_iterator():
    """sort < /data/f with async stdin → stdin from redirect, not pipe."""

    async def _stream():
        yield b"ignored\n"

    dispatch = _mock_dispatch()
    dispatch.return_value = (b"b\na\n", IOResult())
    reg, mount = _mock_registry()
    mount.execute_cmd = AsyncMock(side_effect=_sort_execute_cmd)
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    session = _session()
    node = parse("sort < /data/input.txt")

    stdout, io, _ = _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=_stream()))
    read_calls = [c for c in dispatch.call_args_list if c[0][0] == "read"]
    assert len(read_calls) == 1
    assert io.exit_code == 0
    assert stdout == b"a\nb\n"


# ═══════════════════════════════════════════════
# stdin buffer advancement (read consumes lines)
# ═══════════════════════════════════════════════


def test_sequential_reads_advance_through_lines():
    """read A; read B; read C with 3 lines → each gets one line."""
    _, _, _, session, _, _ = _exec_with_stdin("read A; read B; read C",
                                              stdin=b"first\nsecond\nthird\n")
    assert session.env["A"] == "first"
    assert session.env["B"] == "second"
    assert session.env["C"] == "third"


def test_read_past_end_returns_empty():
    """read A; read B with 1 line → A=line, B=empty, exit=1."""
    _, io, _, session, _, _ = _exec_with_stdin("read A; read B",
                                               stdin=b"only\n")
    assert session.env["A"] == "only"
    assert session.env["B"] == ""


def test_for_loop_read_advances_buffer():
    """for x in 1 2 3; do read LINE; done with 3 lines.

    Each iteration reads the next line from materialized buffer.
    """
    _, _, _, session, _, _ = _exec_with_stdin(
        "for x in 1 2 3; do read LINE; export L_$x=$LINE; done",
        stdin=b"alpha\nbeta\ngamma\n")
    # export L_$x doesn't work ($ in key), but LINE advances
    assert session.env.get("LINE") == "gamma"


def test_while_read_pattern():
    """while read LINE; do export LAST=$LINE; done.

    Classic bash pattern. Reads lines until EOF,
    then read returns 1 → while exits.
    """
    _, io, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do export LAST=$LINE; done",
        stdin=b"aaa\nbbb\nccc\n")
    assert session.env["LAST"] == "ccc"
    assert io.exit_code == 0


def test_while_read_single_line():
    """while read LINE; do ...; done with one line."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do export GOT=$LINE; done", stdin=b"only_line\n")
    assert session.env["GOT"] == "only_line"


def test_while_read_empty_stdin():
    """while read LINE; do ...; done with empty stdin → body never runs."""
    _, io, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do export RAN=yes; done", stdin=b"")
    assert "RAN" not in session.env


def test_while_read_async_iterator():
    """while read LINE; do ...; done with async iterator."""

    async def _stream():
        yield b"x\n"
        yield b"y\n"

    _, _, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do export LAST=$LINE; done", stdin=_stream())
    assert session.env["LAST"] == "y"


def test_buffer_reset_after_loop():
    """Buffer is restored after loop exits."""
    session = _session()
    session._stdin_buffer = b"outer\n"
    dispatch = _mock_dispatch()
    reg, mount = _mock_registry()
    job_table = JobTable()
    execute_fn = AsyncMock(return_value=IOResult())
    node = parse("for x in a; do read V; done")

    _run(
        execute_node(dispatch,
                     reg,
                     job_table,
                     execute_fn,
                     "a",
                     node,
                     session,
                     stdin=b"inner\n"))
    # Buffer should be restored to outer value after loop
    assert session._stdin_buffer == b"outer\n"


# ═══════════════════════════════════════════════
# call_stack tests
# ═══════════════════════════════════════════════


def test_function_receives_positional_args():
    """f() { export A=$1; }; f hello → A=hello."""
    _, _, _, session, _, _ = _exec("f() { export A=$1; }; f hello")
    assert session.env["A"] == "hello"


def test_function_multiple_args():
    """f() { export A=$1; export B=$2; }; f x y → A=x B=y."""
    _, _, _, session, _, _ = _exec("f() { export A=$1; export B=$2; }; f x y")
    assert session.env["A"] == "x"
    assert session.env["B"] == "y"


def test_function_dollar_at():
    """f() { export ALL=$@; }; f a b c → ALL='a b c'."""
    _, _, _, session, _, _ = _exec("f() { export ALL=$@; }; f a b c")
    assert session.env["ALL"] == "a b c"


def test_function_dollar_hash():
    """f() { export N=$#; }; f a b → N=2."""
    _, _, _, session, _, _ = _exec("f() { export N=$#; }; f a b")
    assert session.env["N"] == "2"


def test_function_no_args():
    """f() { export N=$#; }; f → N=0."""
    _, _, _, session, _, _ = _exec("f() { export N=$#; }; f")
    assert session.env["N"] == "0"


def test_shift_removes_positional():
    """f() { shift; export A=$1; }; f x y → A=y."""
    _, _, _, session, _, _ = _exec("f() { shift; export A=$1; }; f x y")
    assert session.env["A"] == "y"


def test_nested_function_calls():
    """inner() { export V=$1; }; outer() { inner hello; }; outer."""
    _, _, _, session, _, _ = _exec(
        "inner() { export V=$1; }; outer() { inner hello; }; outer")
    assert session.env["V"] == "hello"


def test_function_args_isolated():
    """Args from outer don't leak into inner.

    outer() { inner; }; inner() { export A=$1; }; outer hello
    → inner has no args, A=''
    """
    _, _, _, session, _, _ = _exec(
        "inner() { export A=$1; }; outer() { inner; }; outer hello")
    assert session.env["A"] == ""


def test_dollar_zero():
    """$0 is always 'mirage'."""
    stdout, _, _, _, _, _ = _exec("echo $0")
    assert stdout == b"mirage\n"


# ── multi-statement if/else/function bodies ────


def test_if_multi_statement_body():
    """if true; then export A=1; export B=2; fi → both execute."""
    _, _, _, session, _, _ = _exec("if true; then export A=1; export B=2; fi")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_if_else_multi_statement():
    """if false; then ...; else export A=1; export B=2; fi."""
    _, _, _, session, _, _ = _exec(
        "if false; then export X=no; else export A=1; export B=2; fi")
    assert "X" not in session.env
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_elif_multi_statement():
    """elif branch with multiple statements."""
    _, _, _, session, _, _ = _exec("if false; then export X=no; "
                                   "elif true; then export A=1; export B=2; "
                                   "fi")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_function_multi_statement_body():
    """f() { export A=1; export B=2; }; f → both execute."""
    _, _, _, session, _, _ = _exec("f() { export A=1; export B=2; }; f")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


def test_function_shift_then_use():
    """f() { shift; export A=$1; }; f x y → A=y."""
    _, _, _, session, _, _ = _exec("f() { shift; export A=$1; }; f x y")
    assert session.env["A"] == "y"


def test_function_multi_with_control():
    """f() { export A=1; if true; then export B=2; fi; export C=3; }; f."""
    _, _, _, session, _, _ = _exec(
        "f() { export A=1; if true; then export B=2; fi; export C=3; }; f")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"
    assert session.env["C"] == "3"


# ── while read pattern (the classic) ──────────


def test_while_read_counts_lines():
    """while read LINE; do export N=...; done — count iterations."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do export LAST=$LINE; done", stdin=b"a\nb\nc\n")
    assert session.env["LAST"] == "c"


def test_while_read_with_break():
    """while read LINE; do ...; break; done — reads one line then stops."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do export FIRST=$LINE; break; done",
        stdin=b"line1\nline2\n")
    assert session.env["FIRST"] == "line1"


def test_while_read_multivar():
    """while read A B; do ...; done — splits each line."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "while read K V; do export LAST_K=$K; export LAST_V=$V; done",
        stdin=b"key1 val1\nkey2 val2\n")
    assert session.env["LAST_K"] == "key2"
    assert session.env["LAST_V"] == "val2"


# ═══════════════════════════════════════════════
# Complex nested shell scripts
# ═══════════════════════════════════════════════


def test_nested_if_while_for():
    """if + while + for nested three levels deep.

    $LINE expands with IFS splitting.
    """
    _, _, _, session, _, _ = _exec_with_stdin(
        "if true; then "
        "while read LINE; do "
        "for w in $LINE; do "
        "export LAST=$w; "
        "done; done; fi",
        stdin=b"hello world\n")
    assert session.env["LAST"] == "world"


def test_for_if_pipeline():
    """for loop with if condition and pipeline in body."""
    _, _, _, session, mount, _ = _exec("for x in a b c; do "
                                       "if true; then "
                                       "echo $x | cat; "
                                       "export DONE=$x; "
                                       "fi; done")
    assert session.env["DONE"] == "c"
    assert mount.execute_cmd.call_count == 3


def test_function_with_while_read():
    """Function using while read pattern."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "process() { while read LINE; do "
        "export LAST=$LINE; done; }; process",
        stdin=b"x\ny\nz\n")
    assert session.env["LAST"] == "z"


def test_nested_functions():
    """Three levels of function nesting."""
    _, _, _, session, _, _ = _exec("a() { export DEPTH=1; b; }; "
                                   "b() { export DEPTH=2; c; }; "
                                   "c() { export DEPTH=3; }; "
                                   "a")
    assert session.env["DEPTH"] == "3"


def test_function_with_args_and_loop():
    """Function with positional args, $@ splits on whitespace."""
    _, _, _, session, _, _ = _exec(
        "proc() { for f in $@; do export LAST=$f; done; }; "
        "proc alpha beta gamma")
    assert session.env["LAST"] == "gamma"


def test_case_in_for():
    """case inside for loop — pattern matching per iteration."""
    _, _, _, session, _, _ = _exec("for f in a.csv b.txt c.csv; do "
                                   "case $f in "
                                   "*.csv) export CSV=$f;; "
                                   "*.txt) export TXT=$f;; "
                                   "esac; done")
    assert session.env["CSV"] == "c.csv"
    assert session.env["TXT"] == "b.txt"


def test_if_and_or_in_for():
    """for loop with && and || in body."""
    _, _, _, session, _, _ = _exec("for x in a b; do "
                                   "true && export OK=$x; "
                                   "false || export FALL=$x; "
                                   "done")
    assert session.env["OK"] == "b"
    assert session.env["FALL"] == "b"


def test_subshell_in_while():
    """Subshell inside while — env isolated per iteration."""
    session = _session(env={"OUTER": "keep"})
    _exec_with_stdin("while read LINE; do "
                     "(export OUTER=nope); "
                     "done",
                     stdin=b"a\nb\n")
    assert session.env["OUTER"] == "keep"


def test_redirect_in_for_with_expansion():
    """for loop writing to different files via redirect."""
    _, _, _, _, _, dispatch = _exec("for name in alpha beta; do "
                                    "echo $name > /data/${name}.txt; "
                                    "done")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 2
    targets = {c[0][1].virtual for c in write_calls}
    assert "/data/alpha.txt" in targets
    assert "/data/beta.txt" in targets


def test_function_calling_function_with_redirect():
    """Function A calls function B, output redirected."""
    _, _, _, _, _, dispatch = _exec("inner() { echo result; }; "
                                    "outer() { inner > /data/out.txt; }; "
                                    "outer")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/out.txt"


def test_while_read_with_case():
    """while read + case — classic line processing."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "while read LINE; do "
        "case $LINE in "
        "err*) export ERR=$LINE;; "
        "*) export OK=$LINE;; "
        "esac; done",
        stdin=b"error: bad\ninfo: good\n")
    assert session.env["ERR"] == "error: bad"
    assert session.env["OK"] == "info: good"


def test_export_in_elif_chain():
    """Multi-branch elif with exports."""
    _, _, _, session, _, _ = _exec(
        "if false; then export R=a; "
        "elif false; then export R=b; "
        "elif false; then export R=c; "
        "elif true; then export R=d; export S=also; "
        "else export R=e; fi")
    assert session.env["R"] == "d"
    assert session.env["S"] == "also"


def test_nested_while_loops():
    """Nested while loops with different stdin buffers."""
    _, _, _, session, _, _ = _exec_with_stdin(
        "while read OUTER; do "
        "export LAST_OUTER=$OUTER; "
        "break; done",
        stdin=b"line1\nline2\n")
    assert session.env["LAST_OUTER"] == "line1"


# ═══════════════════════════════════════════════
# Python command shell-parser routing
# ═══════════════════════════════════════════════
#
# python3 is now a registered general command (see
# mirage.commands.builtin.general.python). End-to-end behavior
# (script read via dispatch, env passthrough, exit codes) is covered
# by tests/workspace/test_workspace.py against a real workspace.
# Tests here only verify the shell parser routes python3 like any
# other command (heredocs, pipelines, redirects, fan-out).


def test_python3_dispatched_to_mount():
    """python3 -c 'code' → dispatched as a command to mount."""
    _, _, _, _, mount, _ = _exec("python3 -c 'print(42)'")
    mount.execute_cmd.assert_called_once()
    assert mount.execute_cmd.call_args[0][0] == "python3"


def test_python3_heredoc_parsed():
    """python3 <<PYEOF\\ncode\\nPYEOF → redirected_statement."""
    _, io, _, _, _, _ = _exec("python3 <<PYEOF\nprint('hi')\nPYEOF")
    assert io.exit_code is not None


def test_python3_heredoc_quoted():
    """python3 <<'SCRIPT'\\ncode\\nSCRIPT → parsed correctly."""
    _, io, _, _, _, _ = _exec("python3 <<'SCRIPT'\nprint(1)\nSCRIPT")
    assert io.exit_code is not None


def test_python3_heredoc_double_quoted():
    """python3 <<"EOF"\\ncode\\nEOF → parsed correctly."""
    _, io, _, _, _, _ = _exec('python3 <<"EOF"\nimport os\nEOF')
    assert io.exit_code is not None


def test_python3_in_pipeline():
    """python3 -c ... | grep → pipeline with python."""
    _, _, exec_node, _, _, _ = _exec("python3 -c 'print(1)' | grep 1")
    assert exec_node.op == "|"


def test_python3_with_redirect():
    """python3 script.py > /data/out.txt → redirect works."""
    _, _, _, _, _, dispatch = _exec("python3 /data/script.py > /data/out.txt")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/out.txt"


# ═══════════════════════════════════════════════
# Builtin commands: echo, printf, sort, seq, expr, sleep
# ═══════════════════════════════════════════════

# ── echo ───────────────────────────────────────


def test_echo_basic():
    stdout, io, _, _, _, _ = _exec("echo hello")
    assert io.exit_code == 0
    assert stdout == b"hello\n"


def test_echo_multiple_args():
    stdout, _, _, _, _, _ = _exec("echo hello world")
    assert stdout == b"hello world\n"


def test_echo_no_args():
    stdout, _, _, _, _, _ = _exec("echo")
    assert stdout == b"\n"


def test_echo_n_flag():
    stdout, _, _, _, _, _ = _exec("echo -n hello")
    assert stdout == b"hello"


def test_echo_var_expansion():
    stdout, _, _, _, _, _ = _exec("echo $X", env={"X": "expanded"})
    assert stdout == b"expanded\n"


def test_echo_concat_expansion():
    stdout, _, _, _, _, _ = _exec("echo $A/$B", env={"A": "dir", "B": "file"})
    assert stdout == b"dir/file\n"


def test_echo_in_if():
    stdout, _, _, session, _, _ = _exec("if true; then echo yes; fi")
    assert stdout == b"yes\n"


def test_echo_in_for():
    stdout, _, _, _, _, _ = _exec("for x in a b c; do echo $x; done")
    assert b"a\n" in stdout
    assert b"c\n" in stdout


def test_echo_pipe_to_cat():
    _, _, exec_node, _, mount, _ = _exec("echo hello | cat")
    assert exec_node.op == "|"
    mount.execute_cmd.assert_called_once()
    assert mount.execute_cmd.call_args[0][0] == "cat"


def test_echo_redirect():
    _, _, _, _, _, dispatch = _exec("echo hello > /data/out.txt")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1
    assert write_calls[0][0][1].virtual == "/data/out.txt"


# ── printf ─────────────────────────────────────


def test_printf_basic():
    stdout, io, _, _, _, _ = _exec("printf hello")
    assert io.exit_code == 0
    assert stdout == b"hello"


def test_printf_newline():
    stdout, _, _, _, _, _ = _exec(r"printf 'hello\n'")
    assert stdout == b"hello\n"


def test_printf_no_args():
    stdout, io, _, _, _, _ = _exec("printf")
    assert io.exit_code == 0
    assert stdout == b""


def test_printf_in_pipeline():
    _, _, exec_node, _, _, _ = _exec("printf hello | cat")
    assert exec_node.op == "|"


# ── sort ───────────────────────────────────────


def test_sort_in_pipeline():
    _, _, exec_node, _, _, _ = _exec("echo hello | sort")
    assert exec_node.op == "|"


def test_sort_pipe_from_echo():
    stdout, io, _, _, _, _ = _exec("echo 'c\nb\na' | sort")
    assert io.exit_code == 0


# ── sleep ──────────────────────────────────────


def test_sleep_zero():
    _, io, _, _, _, _ = _exec("sleep 0")
    assert io.exit_code == 0


def test_sleep_small():
    _, io, _, _, _, _ = _exec("sleep 0.01")
    assert io.exit_code == 0


def test_sleep_invalid():
    _, io, _, _, _, _ = _exec("sleep abc")
    assert io.exit_code == 1


def test_sleep_no_args():
    _, io, _, _, _, _ = _exec("sleep")
    assert io.exit_code == 1
    assert io.stderr == b"sleep: missing operand\n"


# ── nested / combined builtins ─────────────────


def test_echo_pipe_sort():
    stdout, io, _, _, _, _ = _exec_with_stdin("echo hello | sort", stdin=None)
    assert io.exit_code == 0


def test_for_echo_collect():
    """for loop echoing values, all output captured."""
    stdout, _, _, _, _, _ = _exec("for x in c b a; do echo $x; done")
    assert b"c\n" in stdout
    assert b"b\n" in stdout
    assert b"a\n" in stdout


def test_echo_in_function():
    stdout, _, _, _, _, _ = _exec("greet() { echo hello $1; }; greet world")
    assert stdout == b"hello world\n"


def test_echo_in_while():
    stdout, _, _, _, _, _ = _exec("while false; do echo never; done")
    assert stdout is None or stdout == b""


def test_echo_and_export():
    stdout, _, _, session, _, _ = _exec(
        "echo before; export V=set; echo after")
    assert session.env["V"] == "set"
    assert b"after\n" in stdout


def test_echo_in_subshell():
    stdout, _, _, _, _, _ = _exec("(echo sub)")
    assert stdout == b"sub\n"


def test_echo_in_brace_group():
    stdout, _, _, _, _, _ = _exec("{ echo brace; }")
    assert stdout == b"brace\n"


def test_printf_in_function():
    stdout, _, _, _, _, _ = _exec("f() { printf 'hi %s' $1; }; f world")


def test_sort_in_while_read():
    stdout, _, _, session, _, _ = _exec_with_stdin(
        "sort | while read LINE; do export LAST=$LINE; done",
        stdin=b"banana\napple\n")
    assert session.env.get("LAST") is not None


def test_echo_redirect_then_cat():
    """echo writes to file, cat reads it back."""
    _, _, _, _, mount, dispatch = _exec("echo hello > /data/out.txt")
    write_calls = [c for c in dispatch.call_args_list if c[0][0] == "write"]
    assert len(write_calls) == 1


def test_echo_negated():
    _, io, _, _, _, _ = _exec("! echo hello")
    assert io.exit_code == 1


def test_echo_in_case():
    stdout, _, _, _, _, _ = _exec("case yes in yes) echo matched;; esac")
    assert stdout == b"matched\n"


def test_echo_and_chain():
    stdout, _, _, _, _, _ = _exec("echo first && echo second")
    assert b"second\n" in stdout


def test_echo_or_chain():
    stdout, _, _, _, _, _ = _exec("echo first || echo never")
    assert b"first\n" in stdout


# ═══════════════════════════════════════════════
# Background execution (&)
# ═══════════════════════════════════════════════


def test_background_simple():
    """sleep 0 & → runs in background, exits 0."""
    _, io, exec_node, _, _, _ = _exec("sleep 0 &")
    assert exec_node.op == "&"


def test_background_export_before():
    """export A=1 & → export runs in background."""
    _, io, _, session, _, _ = _exec("export A=1 &")
    assert io.exit_code == 0


def test_background_then_foreground():
    """sleep 0 & export B=fg → background + foreground."""
    _, io, _, session, _, _ = _exec("sleep 0 & export B=fg")
    assert session.env["B"] == "fg"


def test_background_echo():
    """echo hello & → echo in background."""
    _, io, exec_node, _, _, _ = _exec("echo hello &")
    assert exec_node.op == "&"
    assert io.exit_code == 0


def test_background_does_not_block():
    """sleep 0.01 & true → true runs immediately."""
    _, io, _, _, _, _ = _exec("sleep 0.01 & true")
    assert io.exit_code == 0


def test_multiple_background():
    """sleep 0 &; sleep 0 &; export DONE=yes."""
    _, _, _, session, _, _ = _exec("sleep 0 & sleep 0 & export DONE=yes")
    assert session.env["DONE"] == "yes"


def test_background_in_sequence():
    """export A=1; sleep 0 &; export B=2."""
    _, _, _, session, _, _ = _exec("export A=1; sleep 0 & export B=2")
    assert session.env["A"] == "1"
    assert session.env["B"] == "2"


# ── pwd builtin ────────────────────────────────


def test_pwd_returns_cwd():
    session = _session(cwd="/data/sub")
    stdout, io, _, _, _, _ = _exec("pwd", session=session)
    assert io.exit_code == 0
    assert stdout == b"/data/sub\n"


def test_pwd_returns_root():
    session = _session(cwd="/")
    stdout, io, _, _, _, _ = _exec("pwd", session=session)
    assert io.exit_code == 0
    assert stdout == b"/\n"


# ── cd builtin ─────────────────────────────────


def test_cd_tilde():
    session = _session(cwd="/data/sub", env={"HOME": "/"})
    dispatch = _mock_dispatch()
    dispatch.side_effect = FileNotFoundError("not found")
    _, io, _, session, _, _ = _exec("cd ~", session=session, dispatch=dispatch)
    assert session.cwd == "/"
