ccb_test

#!/usr/bin/env python3
"""Run the source checkout CCB only from an external test project.

This wrapper is the supported entrypoint for validating source changes without
pointing the user's installed `ccb` or the source checkout itself at live state.
"""

from __future__ import annotations

import time

_CCB_TEST_PROCESS_ENTRY_NS = time.perf_counter_ns()

import os
import sys
from pathlib import Path


SCRIPT_ROOT = Path(__file__).resolve().parent
SOURCE_CCB = SCRIPT_ROOT / "ccb.py"
_STARTUP_TRACE_ENV_KEYS = (
    "CCB_STARTUP_TIMING_TRACE",
    "CCB_STARTUP_TRACE_ID",
    "CCB_STARTUP_TRACE_SPAWN_NS",
    "CCB_STARTUP_TRACE_WRAPPER_ENTRY_NS",
    "CCB_STARTUP_TRACE_WRAPPER_PRE_EXEC_NS",
)


def _is_diagnose(argv: list[str]) -> bool:
    return bool(argv) and argv[0] in {"--diagnose", "diagnose"}


def _is_safe_introspection(argv: list[str]) -> bool:
    if not argv:
        return False
    first = argv[0]
    return first in {"--help", "-h", "--print-version", "version"} or "--help" in argv or "-h" in argv


def _path_is_under(path: Path, root: Path) -> bool:
    try:
        resolved_path = path.resolve()
        resolved_root = root.resolve()
    except Exception:
        resolved_path = path.absolute()
        resolved_root = root.absolute()
    return resolved_path == resolved_root or resolved_root in resolved_path.parents


def _project_arg_paths(argv: list[str], cwd: Path) -> list[Path]:
    paths: list[Path] = []
    index = 0
    while index < len(argv):
        item = argv[index]
        value: str | None = None
        if item == "--project" and index + 1 < len(argv):
            value = argv[index + 1]
            index += 1
        elif item.startswith("--project="):
            value = item.split("=", 1)[1]
        if value:
            project_path = Path(value).expanduser()
            if not project_path.is_absolute():
                project_path = cwd / project_path
            paths.append(project_path)
        index += 1
    return paths


def _split_roots(value: str) -> list[Path]:
    roots: list[Path] = []
    for item in value.split(os.pathsep):
        text = item.strip()
        if text:
            roots.append(Path(text).expanduser())
    return roots


def _default_test_roots() -> list[Path]:
    parent = SCRIPT_ROOT.parent
    return [parent / "test_ccb2"]


def _allowed_roots() -> list[Path]:
    roots: list[Path] = []
    roots.extend(_split_roots(os.environ.get("CCB_TEST_ROOTS", "")))
    roots.extend(_split_roots(os.environ.get("CCB_SOURCE_ALLOWED_ROOTS", "")))
    roots.extend(_default_test_roots())

    unique: list[Path] = []
    seen: set[str] = set()
    for root in roots:
        try:
            key = str(root.resolve())
        except Exception:
            key = str(root.absolute())
        if key not in seen:
            seen.add(key)
            unique.append(root)
    return unique


def _render_paths(paths: list[Path]) -> str:
    return os.pathsep.join(str(path) for path in paths) if paths else "<none>"


def _diagnose(argv: list[str], cwd: Path) -> int:
    diagnose_args = argv[1:] if _is_diagnose(argv) else argv
    project_paths = _project_arg_paths(diagnose_args, cwd)
    default_roots = _default_test_roots()
    explicit_test_roots = _split_roots(os.environ.get("CCB_TEST_ROOTS", ""))
    explicit_source_roots = _split_roots(os.environ.get("CCB_SOURCE_ALLOWED_ROOTS", ""))
    effective_roots = _allowed_roots()
    check_paths = project_paths if project_paths else [cwd]
    source_checkout_cwd = _path_is_under(cwd, SCRIPT_ROOT)
    project_inside_source = any(_path_is_under(path, SCRIPT_ROOT) for path in project_paths)
    paths_allowed = all(
        any(_path_is_under(path, root) for root in effective_roots)
        for path in check_paths
    )
    allowed = paths_allowed and not source_checkout_cwd and not project_inside_source

    print(f"wrapper: {Path(__file__).resolve()}")
    print(f"source_ccb: {SOURCE_CCB}")
    print(f"cwd: {cwd}")
    print(f"project_paths: {_render_paths(project_paths)}")
    print(f"default_roots: {_render_paths(default_roots)}")
    print(f"env_CCB_TEST_ROOTS: {_render_paths(explicit_test_roots)}")
    print(f"env_CCB_SOURCE_ALLOWED_ROOTS: {_render_paths(explicit_source_roots)}")
    print(f"effective_roots: {_render_paths(effective_roots)}")
    print(f"checked_paths: {_render_paths(check_paths)}")
    print(f"source_checkout_cwd: {'yes' if source_checkout_cwd else 'no'}")
    print(f"project_inside_source: {'yes' if project_inside_source else 'no'}")
    print(f"allowed_source_test_project: {'yes' if allowed else 'no'}")
    return 0


def _reject(message: str) -> int:
    print(message, file=sys.stderr)
    return 1


def _preflight(argv: list[str], cwd: Path) -> tuple[bool, str, list[Path]]:
    if _is_safe_introspection(argv):
        return True, "", []

    project_paths = _project_arg_paths(argv, cwd)
    allowed_roots = _allowed_roots()
    if _path_is_under(cwd, SCRIPT_ROOT):
        return (
            False,
            "Refusing to run `ccb_test` from the CCB source checkout.\n"
            "Run source validation from an external test project, for example:\n"
            "  cd /home/bfly/yunwei/test_ccb2 && /home/bfly/yunwei/ccb_source/ccb_test config validate",
            project_paths,
        )
    for project_path in project_paths:
        if _path_is_under(project_path, SCRIPT_ROOT):
            return (
                False,
                "Refusing to run `ccb_test` against a project inside the CCB source checkout.\n"
                "Use the dedicated external test project /home/bfly/yunwei/test_ccb2, "
                "or set CCB_TEST_ROOTS/CCB_SOURCE_ALLOWED_ROOTS for an explicit external test project.",
                project_paths,
            )
    check_paths = project_paths if project_paths else [cwd]
    for path in check_paths:
        if not any(_path_is_under(path, root) for root in allowed_roots):
            rendered = ", ".join(str(root) for root in allowed_roots)
            return (
                False,
                "Refusing to run `ccb_test` outside an allowed source-test project.\n"
                "Run source validation from the dedicated test project, for example:\n"
                "  cd /home/bfly/yunwei/test_ccb2 && /home/bfly/yunwei/ccb_source/ccb_test config validate\n"
                f"Current directory: {cwd}\n"
                f"Checked project path: {path}\n"
                f"Allowed source-test roots: {rendered}\n"
                "Set CCB_TEST_ROOTS or CCB_SOURCE_ALLOWED_ROOTS only for explicit external test projects.",
                project_paths,
            )
    return True, "", project_paths


def main() -> int:
    argv = sys.argv[1:]
    cwd = Path.cwd()
    if _is_diagnose(argv):
        return _diagnose(argv, cwd)

    ok, reason, project_paths = _preflight(argv, cwd)
    if not ok:
        return _reject(reason)

    env = dict(os.environ)
    roots = _allowed_roots()
    env["CCB_SOURCE_ALLOWED_ROOTS"] = os.pathsep.join(str(root) for root in roots)
    env["CCB_TEST_ENTRYPOINT"] = "1"
    env.setdefault("CCB_SKIP_STARTUP_UPDATE_CHECK", "1")
    if _valid_startup_trace_envelope(env):
        env["CCB_STARTUP_TRACE_WRAPPER_ENTRY_NS"] = str(_CCB_TEST_PROCESS_ENTRY_NS)
        env["CCB_STARTUP_TRACE_WRAPPER_PRE_EXEC_NS"] = str(time.perf_counter_ns())
    else:
        for key in _STARTUP_TRACE_ENV_KEYS:
            env.pop(key, None)

    os.execvpe(sys.executable, [sys.executable, str(SOURCE_CCB), *argv], env)
    raise AssertionError("unreachable")


def _valid_startup_trace_envelope(env: dict[str, str]) -> bool:
    if env.get("CCB_STARTUP_TIMING_TRACE") != "1":
        return False
    trace_id = str(env.get("CCB_STARTUP_TRACE_ID") or "")
    suffix = trace_id.removeprefix("trace_")
    if not (
        len(trace_id) == 38
        and len(suffix) == 32
        and all(character in "0123456789abcdef" for character in suffix)
    ):
        return False
    try:
        spawn_ns = int(str(env.get("CCB_STARTUP_TRACE_SPAWN_NS") or ""))
    except ValueError:
        return False
    if spawn_ns <= 0 or spawn_ns > _CCB_TEST_PROCESS_ENTRY_NS:
        return False
    return not any(
        env.get(key)
        for key in (
            "CCB_STARTUP_TRACE_WRAPPER_ENTRY_NS",
            "CCB_STARTUP_TRACE_WRAPPER_PRE_EXEC_NS",
        )
    )


if __name__ == "__main__":
    raise SystemExit(main())