orchestrata_ultimate.verification.autofix.format_fixes

Reference › orchestrata_ultimate › verification › autofix › format_fixes

Format-related fixers for the autofix engine, its only importer. autofix_line_length runs black with a line length of 88 over the workspace through the given interpreter, with a 30-second timeout; a timeout or a missing interpreter is logged at debug level and skipped. autofix_f821_datetime finds files that use datetime.X without importing it and adds the import (helpers _has_datetime_import and _add_datetime_import), returning the number of files changed. autofix_parse_args patches parse_args() to accept an optional argument list. File access goes through _safe_read, which returns None on error, and _safe_write; files are found with walk_python_files. Traces Spec 031.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/autofix/format_fixes.py

"""Format-related autofix functions.

Fixes line length (via black), parse_args testability,
and missing datetime imports (F821).

Spec: 007 - Verification Agent
"""

import logging
import re
import subprocess

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.autofix.helpers import (
    walk_python_files,
)

_logger = logging.getLogger(__name__)

_DATETIME_USAGE_RE = (
    r"\bdatetime\.(now|utcnow|fromisoformat|" r"strptime|combine|min|max|today)\b"
)


@traces(spec="031", story="US-031-001", reqs=["REQ-031-012"])
def autofix_line_length(
    workspace: str,
    venv_python: str = "python3",
) -> None:
    """Auto-format source files with black to fix line length.

    Args:
        workspace: Root directory of the project.
        venv_python: Path to the Python interpreter.
    """
    try:
        subprocess.run(
            [
                venv_python,
                "-m",
                "black",
                "--line-length",
                "88",
                ".",
            ],
            cwd=workspace,
            capture_output=True,
            text=True,
            timeout=30,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
        _logger.debug("black autoformat skipped: %s", exc)


def autofix_parse_args(workspace: str) -> int:
    """Patch parse_args() to accept optional args parameter.

    Makes parse_args testable by accepting an args parameter
    instead of always reading sys.argv.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        content = _safe_read(fpath)
        if content is None:
            continue
        original = content
        content = re.sub(
            r"def parse_args\(\)\s*(->\s*\S+\s*)?:",
            r"def parse_args(args=None) \1:",
            content,
        )
        content = re.sub(
            r"(return\s+\w+\.parse_args)\(\)",
            r"\1(args)",
            content,
        )
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-004"])
def autofix_f821_datetime(workspace: str) -> int:
    """Add missing datetime import when datetime.X is used.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        content = _safe_read(fpath)
        if content is None:
            continue
        if not re.search(_DATETIME_USAGE_RE, content):
            continue
        if _has_datetime_import(content):
            continue
        original = content
        content = _add_datetime_import(content, original)
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-004"])
def _has_datetime_import(content: str) -> bool:
    """Check if datetime class is already imported."""
    if re.search(
        r"^from datetime import .*\bdatetime\b",
        content,
        re.MULTILINE,
    ):
        return True
    if re.search(
        r"^import datetime$",
        content,
        re.MULTILINE,
    ):
        return True
    return False


@traces(spec="031", story="US-031-001", reqs=["REQ-031-004"])
def _add_datetime_import(
    content: str,
    original: str,
) -> str:
    """Add datetime import to content."""
    # Case 1: extend existing from datetime import
    result = re.sub(
        r"^(from datetime import )((?!.*\bdatetime\b).*)",
        r"\1datetime, \2",
        content,
        count=1,
        flags=re.MULTILINE,
    )
    if result != original:
        return result
    # Case 2: add after first import
    result = re.sub(
        r"^(import os\b.*\n)",
        r"\1from datetime import datetime\n",
        content,
        count=1,
        flags=re.MULTILINE,
    )
    if result != original:
        return result
    # Case 3: prepend
    return "from datetime import datetime\n" + content


@traces(spec="031", story="US-031-001", reqs=["REQ-031-015"])
def _safe_read(fpath: str) -> str | None:
    """Read file, returning None on error."""
    try:
        with open(fpath, encoding="utf-8") as f:
            return f.read()
    except OSError, UnicodeDecodeError:
        return None


@traces(spec="031", story="US-031-001", reqs=["REQ-031-015"])
def _safe_write(fpath: str, content: str) -> None:
    """Write content to file."""
    with open(fpath, "w", encoding="utf-8") as f:
        f.write(content)