temp/features/steps/overlay_sandbox_coverage_r3_steps.py

"""Step definitions for overlay_sandbox_coverage_r3.feature.

Exercises uncovered lines in overlay.py including _is_overlayfs_available,
real overlay mode, _mount_overlay, _unmount_overlay, commit/rollback
error paths, and cleanup edge cases.

All step text uses the ``oscov3`` prefix to avoid AmbiguousStep errors.
"""

from __future__ import annotations

import contextlib
import io
import os
import shutil
import subprocess
import tempfile
from unittest.mock import MagicMock, patch

from behave import given, then, when

from cleveragents.infrastructure.sandbox.overlay import (
    OverlaySandbox,
    _is_overlayfs_available,
)
from cleveragents.infrastructure.sandbox.protocol import (
    SandboxCommitError,
    SandboxCreationError,
    SandboxRollbackError,
    SandboxStateError,
    SandboxStatus,
)

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _cleanup_patcher(context, attr_name):
    """Stop a patcher stored on context and remove the attribute."""
    patcher = getattr(context, attr_name, None)
    if patcher is not None:
        with contextlib.suppress(RuntimeError):
            patcher.stop()


# ===================================================================
# Group 1: _is_overlayfs_available() steps
# ===================================================================


@given("oscov3 the lru_cache for _is_overlayfs_available is cleared")
def given_oscov3_clear_lru_cache(context):
    """Clear the lru_cache so the function executes fresh."""
    _is_overlayfs_available.cache_clear()
    context.oscov3_error = None
    context.oscov3_result = None
    # Track patchers for cleanup
    context.oscov3_patchers = []


@given('oscov3 /proc/filesystems is mocked to contain "{content}"')
def given_oscov3_proc_fs_content(context, content):
    """Mock /proc/filesystems to contain specific text (no 'overlay')."""
    # os.path.isfile returns True for /proc/filesystems
    p1 = patch(
        "cleveragents.infrastructure.sandbox.overlay.os.path.isfile",
        return_value=True,
    )
    p1.start()
    context.oscov3_patchers.append(p1)
    context.add_cleanup(p1.stop)

    # builtins open returns the content
    p2 = patch(
        "builtins.open",
        return_value=io.StringIO(content),
    )
    p2.start()
    context.oscov3_patchers.append(p2)
    context.add_cleanup(p2.stop)


@given("oscov3 os.path.isfile is mocked to return False for /proc/filesystems")
def given_oscov3_isfile_false(context):
    """Mock os.path.isfile to return False."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.os.path.isfile",
        return_value=False,
    )
    p.start()
    context.oscov3_patchers.append(p)
    context.add_cleanup(p.stop)


@given("oscov3 open is mocked to raise OSError for /proc/filesystems")
def given_oscov3_open_raises(context):
    """Mock open and os.path.isfile so that reading /proc/filesystems raises."""
    p1 = patch(
        "cleveragents.infrastructure.sandbox.overlay.os.path.isfile",
        return_value=True,
    )
    p1.start()
    context.oscov3_patchers.append(p1)
    context.add_cleanup(p1.stop)

    p2 = patch(
        "builtins.open",
        side_effect=OSError("mocked OSError"),
    )
    p2.start()
    context.oscov3_patchers.append(p2)
    context.add_cleanup(p2.stop)


@given("oscov3 the system is mocked as root with overlay in proc filesystems")
def given_oscov3_root_with_overlay(context):
    """Mock system as root with overlay present in /proc/filesystems."""
    p1 = patch(
        "cleveragents.infrastructure.sandbox.overlay.os.path.isfile",
        return_value=True,
    )
    p1.start()
    context.oscov3_patchers.append(p1)
    context.add_cleanup(p1.stop)

    p2 = patch(
        "builtins.open",
        return_value=io.StringIO("nodev\toverlay\n"),
    )
    p2.start()
    context.oscov3_patchers.append(p2)
    context.add_cleanup(p2.stop)

    p3 = patch(
        "cleveragents.infrastructure.sandbox.overlay.os.geteuid",
        return_value=0,
    )
    p3.start()
    context.oscov3_patchers.append(p3)
    context.add_cleanup(p3.stop)


@given("oscov3 subprocess.run is mocked for successful mount probe")
def given_oscov3_mount_probe_success(context):
    """Mock subprocess.run for mount probe: mount returns 0, umount succeeds."""
    mount_result = MagicMock()
    mount_result.returncode = 0

    umount_result = MagicMock()
    umount_result.returncode = 0

    call_count = {"n": 0}

    def mock_run(cmd, **kwargs):
        call_count["n"] += 1
        if call_count["n"] == 1:
            return mount_result
        return umount_result

    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        side_effect=mock_run,
    )
    p.start()
    context.oscov3_patchers.append(p)
    context.add_cleanup(p.stop)


@given("oscov3 subprocess.run is mocked for failed mount probe")
def given_oscov3_mount_probe_fail(context):
    """Mock subprocess.run for mount probe: mount returns non-zero."""
    mount_result = MagicMock()
    mount_result.returncode = 1

    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        return_value=mount_result,
    )
    p.start()
    context.oscov3_patchers.append(p)
    context.add_cleanup(p.stop)


@given("oscov3 subprocess.run is mocked to raise OSError during probe")
def given_oscov3_probe_oserror(context):
    """Mock subprocess.run to raise OSError during mount probe."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        side_effect=OSError("mocked OSError"),
    )
    p.start()
    context.oscov3_patchers.append(p)
    context.add_cleanup(p.stop)


@given("oscov3 subprocess.run is mocked to raise TimeoutExpired during probe")
def given_oscov3_probe_timeout(context):
    """Mock subprocess.run to raise TimeoutExpired during mount probe."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        side_effect=subprocess.TimeoutExpired(cmd="mount", timeout=5),
    )
    p.start()
    context.oscov3_patchers.append(p)
    context.add_cleanup(p.stop)


@when("oscov3 _is_overlayfs_available is called")
def when_oscov3_is_overlayfs_called(context):
    """Call _is_overlayfs_available and store the result."""
    try:
        context.oscov3_result = _is_overlayfs_available()
    except Exception as exc:
        context.oscov3_error = exc
    finally:
        # Always clear cache after test to not affect other tests
        _is_overlayfs_available.cache_clear()


@then("oscov3 the result should be False")
def then_oscov3_result_false(context):
    """Assert result is False."""
    assert context.oscov3_result is False, (
        f"Expected False, got {context.oscov3_result}"
    )


@then("oscov3 the result should be True")
def then_oscov3_result_true(context):
    """Assert result is True."""
    assert context.oscov3_result is True, f"Expected True, got {context.oscov3_result}"


# ===================================================================
# Group 2: Real overlay in create
# ===================================================================


@given("oscov3 a test directory is initialised")
def given_oscov3_test_directory(context):
    """Create a temporary directory with test files."""
    context.oscov3_tmpdir = tempfile.mkdtemp(prefix="oscov3-test-")
    context.add_cleanup(shutil.rmtree, context.oscov3_tmpdir, True)
    context.oscov3_original = os.path.join(context.oscov3_tmpdir, "original")
    os.makedirs(context.oscov3_original)

    # Create a test file
    with open(os.path.join(context.oscov3_original, "existing.txt"), "w") as fh:
        fh.write("original content")

    context.oscov3_error = None
    context.oscov3_sandbox = None
    context.oscov3_commit_result = None
    context.oscov3_mount_mock = None
    context.oscov3_unmount_mock = None


@given("oscov3 a sandbox is created with _use_real_overlay forced to True")
def given_oscov3_sandbox_real_overlay(context):
    """Create sandbox and force _use_real_overlay to True before create()."""
    context.oscov3_sandbox = OverlaySandbox(
        resource_id="res-oscov3",
        original_path=context.oscov3_original,
    )
    context.oscov3_sandbox._use_real_overlay = True


@given("oscov3 _mount_overlay is mocked to succeed")
def given_oscov3_mount_overlay_mocked(context):
    """Mock _mount_overlay to be a no-op."""
    p = patch.object(context.oscov3_sandbox, "_mount_overlay")
    context.oscov3_mount_mock = p.start()
    context.add_cleanup(p.stop)


@when('oscov3 the sandbox create is called for plan "{plan_id}"')
def when_oscov3_create_called(context, plan_id):
    """Call create() on the sandbox."""
    try:
        context.oscov3_sandbox.create(plan_id)
    except Exception as exc:
        context.oscov3_error = exc


@then('oscov3 the sandbox should be in "{state}" state')
def then_oscov3_sandbox_state(context, state):
    """Assert the sandbox is in the expected state."""
    assert context.oscov3_sandbox.status.value == state, (
        f"Expected {state}, got {context.oscov3_sandbox.status.value}"
    )


@then("oscov3 _mount_overlay should have been called")
def then_oscov3_mount_called(context):
    """Assert _mount_overlay was invoked."""
    assert context.oscov3_mount_mock is not None
    context.oscov3_mount_mock.assert_called_once()


# ===================================================================
# Group 3: Commit edge cases
# ===================================================================


@given('oscov3 a sandbox is created and activated for plan "{plan_id}"')
def given_oscov3_sandbox_active(context, plan_id):
    """Create a sandbox and activate via get_path."""
    context.oscov3_sandbox = OverlaySandbox(
        resource_id="res-oscov3",
        original_path=context.oscov3_original,
    )
    context.oscov3_sandbox.create(plan_id=plan_id)
    context.oscov3_sandbox.get_path("existing.txt")


@given("oscov3 merged_dir is forced to None")
def given_oscov3_merged_dir_none(context):
    """Force the sandbox _merged_dir to None."""
    context.oscov3_sandbox._merged_dir = None


@when("oscov3 commit is attempted")
def when_oscov3_commit_attempted(context):
    """Attempt commit, catching errors."""
    try:
        context.oscov3_commit_result = context.oscov3_sandbox.commit()
    except Exception as exc:
        context.oscov3_error = exc


@then('oscov3 a SandboxStateError should have been raised with "{msg}"')
def then_oscov3_state_error_with_msg(context, msg):
    """Assert a SandboxStateError was raised with expected message."""
    assert context.oscov3_error is not None, (
        "Expected SandboxStateError but none raised"
    )
    assert isinstance(context.oscov3_error, SandboxStateError), (
        f"Expected SandboxStateError, got {type(context.oscov3_error).__name__}"
    )
    assert msg in str(context.oscov3_error), (
        f"Expected '{msg}' in '{context.oscov3_error}'"
    )


@given('oscov3 a file "{filename}" is added to the sandbox')
def given_oscov3_file_added(context, filename):
    """Create a file in the sandbox merged directory."""
    merged = context.oscov3_sandbox._merged_dir
    filepath = os.path.join(merged, filename)
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    with open(filepath, "w") as fh:
        fh.write("test content for " + filename)


@given("oscov3 backup_directory is mocked to raise RuntimeError")
def given_oscov3_backup_raises(context):
    """Mock backup_directory to raise RuntimeError."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.backup_directory",
        side_effect=RuntimeError("backup failed"),
    )
    p.start()
    context.add_cleanup(p.stop)


@then("oscov3 a SandboxCommitError should have been raised")
def then_oscov3_commit_error(context):
    """Assert a SandboxCommitError was raised."""
    assert context.oscov3_error is not None, "Expected SandboxCommitError"
    assert isinstance(context.oscov3_error, SandboxCommitError), (
        f"Expected SandboxCommitError, got {type(context.oscov3_error).__name__}"
    )


@given("oscov3 copy2 is mocked to fail after backup succeeds")
def given_oscov3_copy2_fails_after_backup(context):
    """Mock shutil.copy2 to raise OSError (backup_directory is NOT mocked)."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.shutil.copy2",
        side_effect=OSError("disk full during copy"),
    )
    p.start()
    context.add_cleanup(p.stop)


@given("oscov3 safe_restore is mocked to raise RuntimeError")
def given_oscov3_safe_restore_fails(context):
    """Mock safe_restore to raise RuntimeError (simulates restore failure)."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.safe_restore",
        side_effect=RuntimeError("restore failed"),
    )
    p.start()
    context.add_cleanup(p.stop)


# ===================================================================
# Group 4: Rollback edge cases
# ===================================================================


@when("oscov3 rollback is attempted")
def when_oscov3_rollback_attempted(context):
    """Attempt rollback, catching errors."""
    try:
        context.oscov3_sandbox.rollback()
    except Exception as exc:
        context.oscov3_error = exc


@given("oscov3 _use_real_overlay is forced to True")
def given_oscov3_force_real_overlay(context):
    """Force the sandbox to think it has real overlay support."""
    context.oscov3_sandbox._use_real_overlay = True


@given("oscov3 the sandbox is committed successfully")
def given_oscov3_committed(context):
    """Commit the sandbox (should succeed)."""
    context.oscov3_commit_result = context.oscov3_sandbox.commit()
    assert context.oscov3_sandbox.status == SandboxStatus.COMMITTED


@given("oscov3 subprocess.run is mocked for umount and mount during rollback")
def given_oscov3_subprocess_umount_mount(context):
    """Mock subprocess.run to handle umount (check=True) and _mount_overlay."""
    # For COMMITTED rollback with real overlay:
    # 1. umount is called via subprocess.run with check=True (line 437-441)
    # 2. _mount_overlay is called which also uses subprocess.run

    umount_result = MagicMock()
    umount_result.returncode = 0

    mount_result = MagicMock()
    mount_result.returncode = 0

    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        return_value=mount_result,
    )
    p.start()
    context.add_cleanup(p.stop)


@given("oscov3 subprocess.run is mocked to raise CalledProcessError for umount")
def given_oscov3_umount_raises_cpe(context):
    """Mock subprocess.run to raise CalledProcessError (umount failure)."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        side_effect=subprocess.CalledProcessError(
            returncode=1, cmd=["umount", "/tmp/merged"]
        ),
    )
    p.start()
    context.add_cleanup(p.stop)


@when("oscov3 rollback is called")
def when_oscov3_rollback_called(context):
    """Call rollback and catch any errors."""
    try:
        context.oscov3_sandbox.rollback()
    except Exception as exc:
        context.oscov3_error = exc


@then("oscov3 a SandboxRollbackError should have been raised")
def then_oscov3_rollback_error(context):
    """Assert a SandboxRollbackError was raised."""
    assert context.oscov3_error is not None, "Expected SandboxRollbackError"
    assert isinstance(context.oscov3_error, SandboxRollbackError), (
        f"Expected SandboxRollbackError, got {type(context.oscov3_error).__name__}"
    )


@given("oscov3 subprocess.run is mocked for unmount and mount during active rollback")
def given_oscov3_subprocess_active_rollback(context):
    """Mock subprocess.run for ACTIVE rollback with real overlay.

    _unmount_overlay calls subprocess.run with check=True.
    _mount_overlay also calls subprocess.run with check=True.
    """
    result = MagicMock()
    result.returncode = 0

    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        return_value=result,
    )
    p.start()
    context.add_cleanup(p.stop)


# ===================================================================
# Group 5: Cleanup
# ===================================================================


@given('oscov3 a sandbox is created for plan "{plan_id}"')
def given_oscov3_sandbox_created(context, plan_id):
    """Create and initialise a sandbox."""
    context.oscov3_sandbox = OverlaySandbox(
        resource_id="res-oscov3",
        original_path=context.oscov3_original,
    )
    context.oscov3_sandbox.create(plan_id=plan_id)


@given("oscov3 _unmount_overlay is mocked")
def given_oscov3_unmount_mocked(context):
    """Mock _unmount_overlay."""
    p = patch.object(context.oscov3_sandbox, "_unmount_overlay")
    context.oscov3_unmount_mock = p.start()
    context.add_cleanup(p.stop)


@when("oscov3 cleanup is called")
def when_oscov3_cleanup(context):
    """Call cleanup on the sandbox."""
    try:
        context.oscov3_sandbox.cleanup()
    except Exception as exc:
        context.oscov3_error = exc


@then("oscov3 _unmount_overlay should have been called")
def then_oscov3_unmount_called(context):
    """Assert _unmount_overlay was called."""
    assert context.oscov3_unmount_mock is not None
    context.oscov3_unmount_mock.assert_called_once()


@given("oscov3 a pre_commit_backup directory exists")
def given_oscov3_pre_commit_backup_exists(context):
    """Create a pre_commit_backup directory and set it on the sandbox."""
    backup_dir = tempfile.mkdtemp(prefix="oscov3-backup-", dir=context.oscov3_tmpdir)
    # Put a file in it so rmtree has something to clean
    with open(os.path.join(backup_dir, "backup.txt"), "w") as fh:
        fh.write("backup data")
    context.oscov3_sandbox._pre_commit_backup = backup_dir
    context.oscov3_backup_path = backup_dir


@then("oscov3 pre_commit_backup should be None")
def then_oscov3_backup_none(context):
    """Assert _pre_commit_backup is None after cleanup."""
    assert context.oscov3_sandbox._pre_commit_backup is None


@given("oscov3 pre_commit_backup is set to a non-existent path")
def given_oscov3_backup_nonexistent(context):
    """Set _pre_commit_backup to a path that does not exist."""
    context.oscov3_sandbox._pre_commit_backup = os.path.join(
        context.oscov3_tmpdir, "nonexistent-backup-dir"
    )


# ===================================================================
# Group 6: _mount_overlay / _unmount_overlay
# ===================================================================


@given("oscov3 _original_path contains a comma")
def given_oscov3_original_path_comma(context):
    """Set _original_path to contain a comma."""
    context.oscov3_sandbox._original_path = "/tmp/path,with,commas"


@given("oscov3 _upper_dir contains a comma")
def given_oscov3_upper_dir_comma(context):
    """Set _upper_dir to contain a comma."""
    context.oscov3_sandbox._upper_dir = "/tmp/upper,dir"


@when("oscov3 _mount_overlay is called directly")
def when_oscov3_mount_overlay_direct(context):
    """Call _mount_overlay directly and catch errors."""
    try:
        context.oscov3_sandbox._mount_overlay()
    except Exception as exc:
        context.oscov3_error = exc


@then('oscov3 a SandboxCreationError should have been raised with "{msg}"')
def then_oscov3_creation_error_msg(context, msg):
    """Assert a SandboxCreationError was raised with expected message."""
    assert context.oscov3_error is not None, "Expected SandboxCreationError"
    assert isinstance(context.oscov3_error, SandboxCreationError), (
        f"Expected SandboxCreationError, got {type(context.oscov3_error).__name__}"
    )
    assert msg.lower() in str(context.oscov3_error).lower(), (
        f"Expected '{msg}' in '{context.oscov3_error}'"
    )


@given("oscov3 subprocess.run is mocked for successful mount")
def given_oscov3_subprocess_mount_ok(context):
    """Mock subprocess.run to succeed for mount."""
    result = MagicMock()
    result.returncode = 0

    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        return_value=result,
    )
    p.start()
    context.add_cleanup(p.stop)


@given("oscov3 subprocess.run is mocked to raise CalledProcessError for mount")
def given_oscov3_subprocess_mount_cpe(context):
    """Mock subprocess.run to raise CalledProcessError."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        side_effect=subprocess.CalledProcessError(
            returncode=1, cmd=["mount", "-t", "overlay"]
        ),
    )
    p.start()
    context.add_cleanup(p.stop)


@given("oscov3 subprocess.run is mocked to raise TimeoutExpired for mount")
def given_oscov3_subprocess_mount_timeout(context):
    """Mock subprocess.run to raise TimeoutExpired."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        side_effect=subprocess.TimeoutExpired(cmd="mount", timeout=30),
    )
    p.start()
    context.add_cleanup(p.stop)


@given("oscov3 subprocess.run is mocked to raise OSError for mount")
def given_oscov3_subprocess_mount_oserror(context):
    """Mock subprocess.run to raise OSError."""
    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        side_effect=OSError("no such file"),
    )
    p.start()
    context.add_cleanup(p.stop)


@then("oscov3 no error should have been raised")
def then_oscov3_no_error(context):
    """Assert no error was raised."""
    assert context.oscov3_error is None, (
        f"Expected no error, got {context.oscov3_error}"
    )


@given("oscov3 subprocess.run is mocked for successful umount")
def given_oscov3_subprocess_umount_ok(context):
    """Mock subprocess.run to succeed for umount."""
    result = MagicMock()
    result.returncode = 0

    p = patch(
        "cleveragents.infrastructure.sandbox.overlay.subprocess.run",
        return_value=result,
    )
    p.start()
    context.add_cleanup(p.stop)


@when("oscov3 _unmount_overlay is called directly")
def when_oscov3_unmount_direct(context):
    """Call _unmount_overlay directly and catch errors."""
    try:
        context.oscov3_sandbox._unmount_overlay()
    except Exception as exc:
        context.oscov3_error = exc