sandbox.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mutation/sandbox.py
 
"""
Purpose
-------
Apply one mutant to a real file and guarantee the pristine bytes come back.
 
Responsibilities
----------------
- Refuse a target that git cannot restore before anything is written.
- Record the pristine bytes, write the mutant, and restore unconditionally.
- Preserve the target's declared source encoding when writing a mutant.
 
Diagnostics
-----------
Domain: OCT-MUTATION
Levels:
    L2 - lifecycle
    L3 - apply and restore details
    L4 - deep tracing
 
Contracts
---------
- Restoration runs in a ``finally``, so it also executes on ``KeyboardInterrupt``
  and on any exception raised inside the managed block.
- Restoration writes the recorded bytes back unconditionally, so it repairs a
  target that the managed block replaced rather than merely mutated.
- A refusal is raised before the first write, never after a partial one.
- This module makes no claim about processes it did not start: a child that
  outlives the managed block can still observe or rewrite the target, which is
  a named residual rather than a closed gap.
 
Dependencies
------------
- oct.core.git (repository discovery and working-tree state, read-only)
- oct.mutation.engine (mutant application)
- tokenize (declared source encodings)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="Restoration must not depend on diagnostics being initialised; a _dbg failure inside the finally would defeat the guarantee"
# OCT-LINT: disable-file=dbg_import reason="Restoration path is kept import-minimal by design"
 
from __future__ import annotations
 
import io
import tokenize
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator
 
from oct.core import git as core_git
from oct.mutation.engine import apply_mutant
 
 
class MutationRefusal(Exception):
    """A named refusal: the run stopped before any file was written."""
 
 
def ensure_recoverable_target(path: str | Path) -> Path:
    """Return the repository root for a target git can restore, or refuse.
 
    The refusal is the point of the function. Mutation rewrites a real file in
    place, so the only honest precondition is that a committed copy exists to
    fall back on if the process dies between the write and the restore. An
    untracked or already-modified target has no such copy.
    """
    resolved = Path(path).resolve()
    if not resolved.is_file():
        raise MutationRefusal(f"target is not a file: {resolved}")
    if not core_git.is_git_repo(resolved.parent):
        raise MutationRefusal(
            f"target is not inside a git repository: {resolved}. Mutation "
            "rewrites the file in place and will not touch a target that "
            "cannot be restored from git if this process dies."
        )
    try:
        repo_root = core_git.get_repo_root(resolved.parent)
    except core_git.GitError as exc:
        raise MutationRefusal(f"could not resolve the repository for {resolved}: {exc}") from exc
    relative = _repo_relative(resolved, repo_root)
    _refuse_untracked(relative, repo_root)
    _refuse_dirty(relative, repo_root)
    return repo_root
 
 
def _repo_relative(resolved: Path, repo_root: Path) -> str:
    """Return the target's repo-relative posix path for git plumbing calls."""
    try:
        return resolved.relative_to(repo_root).as_posix()
    except ValueError as exc:
        raise MutationRefusal(
            f"target {resolved} is not inside its own repository root {repo_root}"
        ) from exc
 
 
def _refuse_untracked(relative: str, repo_root: Path) -> None:
    """Refuse a target git does not track."""
    try:
        proc = core_git.git_run(
            ["ls-files", "--error-unmatch", "--", relative],
            cwd=repo_root,
            check=False,
        )
    except core_git.GitError as exc:
        raise MutationRefusal(f"could not query git for {relative}: {exc}") from exc
    if proc.returncode != 0:
        raise MutationRefusal(
            f"target is not tracked by git: {relative}. An untracked file has "
            "no committed copy to restore from, so it is never mutated."
        )
 
 
def _refuse_dirty(relative: str, repo_root: Path) -> None:
    """Refuse a target that carries uncommitted working-tree or index changes."""
    try:
        proc = core_git.git_run(
            ["status", "--porcelain", "--", relative],
            cwd=repo_root,
            check=False,
        )
    except core_git.GitError as exc:
        raise MutationRefusal(f"could not query git for {relative}: {exc}") from exc
    if proc.returncode != 0:
        raise MutationRefusal(
            f"git status failed for {relative}: {proc.stderr.strip()}"
        )
    if proc.stdout.strip():
        raise MutationRefusal(
            f"target has uncommitted changes: {relative}. Commit or stash them "
            "first; mutation must be able to restore the file from git."
        )
 
 
def read_target(path: str | Path) -> tuple[bytes, str, str]:
    """Return a target's pristine bytes, decoded source, and declared encoding."""
    resolved = Path(path)
    pristine = resolved.read_bytes()
    encoding, _ = tokenize.detect_encoding(io.BytesIO(pristine).readline)
    return pristine, pristine.decode(encoding), encoding
 
 
@contextmanager
def mutated_target(
    path: str | Path,
    mutant_id: str,
    relpath: str,
) -> Iterator[str]:
    """Write one mutant to ``path``, yield its text, and always restore.
 
    The pristine bytes are captured before anything else happens and written
    back in a ``finally``, so restoration survives an exception in the body, a
    ``KeyboardInterrupt``, and a body that replaced the file outright. The
    restore is a plain unconditional write rather than a conditional repair:
    comparing first would leave a window where a crash between the comparison
    and the write loses the original.
    """
    resolved = Path(path)
    pristine, source, encoding = read_target(resolved)
    try:
        mutated_source = apply_mutant(source, mutant_id, relpath)
        resolved.write_bytes(mutated_source.encode(encoding))
        yield mutated_source
    finally:
        resolved.write_bytes(pristine)