honey-biscuit-workshop/tests/conftest.py

"""
Root conftest for HoneyBiscuitWorkshop system tests.

Provides core fixtures for:
- Locating the baker binary and examples
- Running subprocess commands with output capture
- Managing temporary working directories
"""
import os
import subprocess
import shutil
import tempfile
import pytest


def pytest_configure(config):
    """Register custom markers."""
    config.addinivalue_line("markers", "prebake: prebake stage tests")
    config.addinivalue_line("markers", "bake: bake stage tests")
    config.addinivalue_line("markers", "finalize: finalize stage tests")
    config.addinivalue_line("markers", "llm: tests requiring LLM API")
    config.addinivalue_line("markers", "slow: slow running tests")


@pytest.fixture(scope="session")
def baker_bin():
    """Path to the workshop-baker binary."""
    path = os.environ.get("BAKER_BIN", "/workshop/workshop-baker")
    if not os.path.isfile(path):
        pytest.skip(f"Baker binary not found at {path}")
    if not os.access(path, os.X_OK):
        pytest.skip(f"Baker binary not executable at {path}")
    return path


@pytest.fixture(scope="session")
def examples_dir():
    """Path to the examples directory (may be None)."""
    path = os.environ.get("EXAMPLES_DIR", "/workshop/examples")
    if not os.path.isdir(path):
        return None
    return path


@pytest.fixture(scope="session")
def llm_env():
    """LLM environment variables dict (from llm.env)."""
    env = {}
    for var in [
        "OPENAI_BASE_URL",
        "OPENAI_API_KEY",
        "ANTHROPIC_BASE_URL",
        "ANTHROPIC_API_KEY",
        "MODEL",
        "REASONING_EFFORT",
    ]:
        val = os.environ.get(var)
        if val:
            env[var] = val
    return env


@pytest.fixture
def work_dir():
    """Create a temporary working directory for a test, cleaned up after."""
    d = tempfile.mkdtemp(prefix="hbw-test-")
    yield d
    shutil.rmtree(d, ignore_errors=True)


@pytest.fixture
def run_cmd():
    """Fixture that provides a command runner function.

    Returns a function that runs a command and returns
    (stdout, stderr, returncode, success).
    """

    def _run(args, cwd=None, env=None, timeout=120):
        full_env = {**os.environ}
        if env:
            full_env.update(env)

        show_logs = os.environ.get("BAKER_LOG", "0") == "1"

        try:
            if show_logs:
                # Stream output to terminal in real-time while capturing
                import sys, select
                p = subprocess.Popen(
                    args, cwd=cwd, env=full_env,
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                    text=True,
                )
                stdout_chunks, stderr_chunks = [], []
                try:
                    p.wait(timeout=timeout)
                except subprocess.TimeoutExpired:
                    p.kill()
                    p.wait()
                    return {"stdout": "", "stderr": f"Command timed out after {timeout}s",
                            "returncode": -1, "success": False}
                for line in p.stdout:
                    sys.stderr.write("[baker] " + line)
                    stdout_chunks.append(line)
                for line in p.stderr:
                    sys.stderr.write("[baker:err] " + line)
                    stderr_chunks.append(line)
                return {
                    "stdout": "".join(stdout_chunks),
                    "stderr": "".join(stderr_chunks),
                    "returncode": p.returncode,
                    "success": p.returncode == 0,
                }
            else:
                result = subprocess.run(
                    args, cwd=cwd, env=full_env,
                    capture_output=True, text=True, timeout=timeout,
                )
                return {
                    "stdout": result.stdout,
                    "stderr": result.stderr,
                    "returncode": result.returncode,
                    "success": result.returncode == 0,
                }
        except subprocess.TimeoutExpired:
            return {
                "stdout": "",
                "stderr": f"Command timed out after {timeout}s",
                "returncode": -1,
                "success": False,
            }
        except Exception as e:
            return {
                "stdout": "",
                "stderr": str(e),
                "returncode": -1,
                "success": False,
            }

    return _run


@pytest.fixture
def run_baker(baker_bin, run_cmd):
    """Fixture that provides a baker command runner.

    Returns a function that runs workshop-baker with common flags.
    """

    def _run(subcommand, *args, cwd=None, env=None, timeout=120):
        cmd = [
            baker_bin,
            "-u", "testuser",
            "-p", "test-pipeline",
            "-b", "test-build-001",
            "bare",
            subcommand,
        ]
        cmd.extend(args)
        return run_cmd(cmd, cwd=cwd, env=env, timeout=timeout)

    return _run


@pytest.fixture
def fixtures_dir():
    """Path to the test fixtures directory."""
    return os.path.join(os.path.dirname(__file__), "fixtures")


@pytest.fixture
def bake_workspace():
    """Create /workspace for bake tests (baker writes temp scripts there)."""
    existed = os.path.exists("/workspace")
    if not existed:
        os.makedirs("/workspace", exist_ok=True)
    yield
    if not existed:
        shutil.rmtree("/workspace", ignore_errors=True)