scripts/ft.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""FlintTrade cross-platform task runner.
``scripts/ft.py`` is the single task entry point that behaves identically on
Windows, macOS and Linux. It replaces the shell-only parts of the Makefile, so
none of the following are required: ``make``, ``bash``, ``lsof``, ``ss``,
``find``, ``du``, ``grep``, ``awk`` or ``echo -e``. Nothing outside the Python
standard library is imported.
Usage::
python scripts/ft.py <start|stop|restart|status|dev|setup|test|test-fast|check|lint|clean|version|help|
desktop-test|desktop-build|desktop-package|desktop-dev>
After a FlintTrade install the shim exposes exactly the same interface as::
flinttrade <same subcommands>
Run ``python scripts/ft.py help`` for the full command table.
"""
from __future__ import annotations
import importlib.util
import json
import os
import platform
import re
import select
import shlex
import shutil
import signal
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from collections.abc import Callable, Iterator, Sequence
from pathlib import Path
from types import ModuleType, SimpleNamespace
def _load_module_from_path(module_name: str, module_path: str | Path) -> ModuleType:
"""Load a module from an exact path with normal ``sys.modules`` semantics.
An existing entry from the same file is reused. Name collisions from other
files receive a numeric suffix and are left untouched. Failed execution
removes only the partially initialized module created by this call.
"""
resolved_path = Path(module_path).resolve()
suffix = 0
while True:
candidate_name = module_name if suffix == 0 else f"{module_name}_{suffix}"
if candidate_name not in sys.modules:
break
existing = sys.modules[candidate_name]
existing_path = getattr(existing, "__file__", None)
if existing_path is not None:
try:
if Path(existing_path).resolve() == resolved_path:
return existing
except (OSError, TypeError, ValueError):
pass
suffix += 1
spec = importlib.util.spec_from_file_location(candidate_name, resolved_path)
if spec is None or spec.loader is None:
raise ImportError(f"cannot load module {candidate_name!r} from {resolved_path}")
module = importlib.util.module_from_spec(spec)
sys.modules[candidate_name] = module
try:
spec.loader.exec_module(module)
except BaseException:
if sys.modules.get(candidate_name) is module:
del sys.modules[candidate_name]
raise
return module
_BROKER_SDK_ENVIRONMENT_PATH = Path(__file__).with_name("broker_sdk_environment.py").resolve()
_BROKER_SDK_ENVIRONMENT = _load_module_from_path(
"_flinttrade_broker_sdk_environment", _BROKER_SDK_ENVIRONMENT_PATH
)
remove_kotak_distributions = _BROKER_SDK_ENVIRONMENT.remove_kotak_distributions
repair_kotakneo_environment = _BROKER_SDK_ENVIRONMENT.repair_kotakneo_environment
repair_broker_sdk_environment = _BROKER_SDK_ENVIRONMENT.repair_broker_sdk_environment
REPO_ROOT = Path(__file__).resolve().parents[1]
"""Repository root - ``scripts/ft.py`` lives one level below it."""
IS_WINDOWS = os.name == "nt"
BACKEND_HOST = os.environ.get("FLINTTRADE_BACKEND_HOST", "127.0.0.1")
BACKEND_PORT = int(os.environ.get("FLINTTRADE_BACKEND_PORT", "5100"))
BACKEND_URL = f"http://{BACKEND_HOST}:{BACKEND_PORT}"
CORE_SRC = REPO_ROOT / "packages" / "core" / "core" / "src"
DEV_LOG_DIR = REPO_ROOT / ".local" / "dev-logs"
DESKTOP_PKG = "packages/apps/desktop"
TERMINAL_PKG = "packages/apps/terminal"
PINNED_PNPM_VERSION = "10.34.5"
PYTEST_TIMEOUT_SECONDS = 120
"""Per-test wall-clock bound for local runs.
Higher than CI's 60 because a developer machine is usually running a browser, an
editor and often the app itself, so the honest ceiling is looser. The point is
not the exact number - it is that an unbounded local run turns a flaky
process-spawning test into an indefinite hang with no output at all.
"""
PYTEST_NATIVE_THREAD_ENV = dict.fromkeys((
"OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS",
"VECLIB_MAXIMUM_THREADS", "NUMEXPR_NUM_THREADS",
), "1")
"""Test-only native thread bounds; xdist already supplies process parallelism."""
# Directories that `clean` and the workspace walk never descend into: the repo
# virtualenv, the git object store, and the gitignored .local/ scratch tree
# (external test-dep clones and archives live there and are not ours to delete).
PRUNED_DIRS = frozenset({".git", ".venv", ".local"})
# Every key here MUST have a handler in HANDLERS and MUST appear in the
# Makefile header's "Via scripts/ft.py" list - tests/test_ft_runner.py pins all
# three surfaces together, because a command documented in only two of them is
# a command that exits 2 when a user runs it.
COMMANDS: dict[str, str] = {
"start": "Start the FlintTrade backend API on port 5100",
"start-gateway": "Alias for start (the standalone FlintTrade backend)",
"stop": "Request workspace stop (refused without verified safe-maintenance policy)",
"restart": "Request stop, then start only after confirmed safe shutdown",
"status": "Show backend, port-ownership, workspace and version status",
"dev": "Run the terminal dev server alongside the FlintTrade backend",
"setup": "First-time setup - install Python and Node dependencies",
"test": "Run pytest (paths/flags accepted); default also tests Rust when cargo is present",
"test-fast": "Run pytest (paths/flags accepted) and stop on the first failure",
"check": "Run affected checks; --full runs the exhaustive local gate; --dry-run prints commands",
"lint": "Run ruff over packages/ and tests/, then the terminal's react-hooks gate",
"clean": "Delete __pycache__, .pytest_cache and node_modules trees",
"version": "Print the FlintTrade version",
"help": "Show this command table",
"desktop-test": "Typecheck and test the Electron shell",
"desktop-build": "Verify and bundle the Electron shell",
"desktop-package": "Package and verify the Electron installer for this OS/arch",
"desktop-dev": "Run the Electron shell against its managed source bootstrap",
}
# ---------------------------------------------------------------------------
# Output helpers (ASCII only - Windows consoles default to a non-UTF-8 codepage)
# ---------------------------------------------------------------------------
def info(message: str) -> None:
"""Print an informational line to stdout.
Args:
message: The text to display.
"""
print(message, flush=True)
def warn(message: str) -> None:
"""Print a warning line to stderr.
Args:
message: The text to display.
"""
print(f"warning: {message}", file=sys.stderr, flush=True)
def fail(message: str) -> None:
"""Print an error line to stderr.
Args:
message: The text to display.
"""
print(f"error: {message}", file=sys.stderr, flush=True)
def header(title: str) -> None:
"""Print a section header.
Args:
title: The section name.
"""
print(f"\n=== {title} ===", flush=True)
# ---------------------------------------------------------------------------
# Interpreter resolution
# ---------------------------------------------------------------------------
def is_store_alias(candidate: str | Path) -> bool:
"""Recognise a Microsoft Store "App execution alias" stub.
Windows ships ``python.exe``/``python3.exe`` shims under ``WindowsApps`` that
exist on ``PATH`` but do not run anything: invoked non-interactively they
exit 49 and open the Store instead. They must never be selected silently.
Args:
candidate: Path to a candidate interpreter.
Returns:
True when the path lives inside a ``WindowsApps`` directory.
"""
if not IS_WINDOWS:
return False
return "\\windowsapps\\" in str(candidate).replace("/", "\\").lower()
def probe_python(candidate: str | Path) -> bool:
"""Check that *candidate* is a real, runnable CPython interpreter.
This is the check that rejects the Microsoft Store alias stub: the stub
exits 49 without executing the ``-c`` snippet, so a non-zero return code
means the candidate is unusable and the caller should fall through.
Args:
candidate: Path to a candidate interpreter.
Returns:
True when the interpreter ran the probe successfully.
"""
try:
proc = subprocess.run(
[str(candidate), "-c", "import sys; sys.exit(0)"],
capture_output=True,
timeout=30,
check=False,
)
except (OSError, subprocess.SubprocessError):
return False
return proc.returncode == 0
def resolve_python() -> str:
"""Resolve the interpreter used for every FlintTrade Python invocation.
Resolution order:
1. The repository virtualenv - ``.venv/Scripts/python.exe`` on Windows and
``.venv/bin/python`` on POSIX. ``uv`` creates only one of the two, so both
layouts are checked rather than assuming the POSIX one, but the host's own
layout is always tried first: a ``.venv`` shared with a Windows host (or
left behind by WSL) must never win over the layout that can actually run.
2. The interpreter currently running this script.
3. ``python3`` then ``python`` from ``PATH``.
Every ``PATH`` candidate is probed, which is what recognises and rejects the
Microsoft Store alias stub.
Returns:
An absolute (or ``PATH``-resolved) path to a working interpreter.
Raises:
SystemExit: When no usable interpreter could be found.
"""
venv = REPO_ROOT / ".venv"
windows_layout = (venv / "Scripts" / "python.exe",)
posix_layout = (venv / "bin" / "python", venv / "bin" / "python3")
venv_candidates = windows_layout + posix_layout if IS_WINDOWS else posix_layout + windows_layout
for candidate in venv_candidates:
if candidate.is_file() and probe_python(candidate):
return str(candidate)
if sys.executable and not is_store_alias(sys.executable):
return sys.executable
for name in ("python3", "python"):
found = shutil.which(name)
if not found:
continue
if is_store_alias(found):
# A Store alias may still be a genuine Store-installed Python; probe
# before discarding it, and fall through when it is the stub.
if not probe_python(found):
continue
return found
if probe_python(found):
return found
fail("No usable Python interpreter found.")
fail("Install Python 3.14+ from https://python.org (tick 'Add python.exe to PATH' on Windows),")
fail("or create the repo virtualenv with: uv sync --frozen --all-packages")
raise SystemExit(1)
def python_package_src_dirs() -> list[Path]:
"""Return every workspace Python package source root, core first.
The repository is a uv workspace of thirteen ``flinttrade_*`` distributions
nested at ``packages/<group>/<pkg>/src``. Only the directories that really
hold an importable ``flinttrade_*`` package are returned, so the JavaScript
packages (``packages/apps/terminal/src`` and friends) and the Rust crate are
skipped. ``flinttrade_core`` is listed first because every other package
imports it.
Returns:
Existing source roots, ``packages/core/core/src`` first and the rest
sorted for a stable, reproducible ``PYTHONPATH``.
"""
roots = {
init.parent.parent
for init in REPO_ROOT.glob("packages/*/*/src/flinttrade_*/__init__.py")
}
ordered = sorted(roots - {CORE_SRC})
if CORE_SRC.is_dir():
return [CORE_SRC, *ordered]
return ordered
def python_env(extra: dict[str, str] | None = None) -> dict[str, str]:
"""Build the child-process environment for FlintTrade Python commands.
Every workspace package source root is exposed, not just the core tree:
``flinttrade_core.app`` eagerly imports ``flinttrade_data`` and the other
sibling distributions, so a ``PYTHONPATH`` carrying core alone dies with
``ModuleNotFoundError`` on any environment where the workspace was not
installed (the no-uv ``setup`` fallback, notably).
``PYTHONPATH`` is joined with :data:`os.pathsep` so it is ``;`` on Windows
and ``:`` on POSIX - hardcoding ``:`` silently breaks every import on
Windows because drive letters then split the path.
Args:
extra: Additional environment entries to overlay.
Returns:
A complete environment mapping for :mod:`subprocess`.
"""
env = os.environ.copy()
parts = [str(path) for path in python_package_src_dirs()]
existing = env.get("PYTHONPATH", "")
if existing:
parts.append(existing)
env["PYTHONPATH"] = os.pathsep.join(parts)
env.setdefault("PYTHONIOENCODING", "utf-8")
env.setdefault("PYTHONUTF8", "1")
if extra:
env.update(extra)
return env
def pytest_env(extra: dict[str, str] | None = None) -> dict[str, str]:
"""Bound native test libraries without changing application Python commands.
Each xdist worker otherwise starts its own CPU-sized BLAS/OpenMP pool.
Force the measured test bounds even when the shell inherits larger values;
pytest worker options still control process-level concurrency independently.
"""
return python_env(extra) | PYTEST_NATIVE_THREAD_ENV
# ---------------------------------------------------------------------------
# Process helpers
# ---------------------------------------------------------------------------
def run(
argv: Sequence[str],
*,
env: dict[str, str] | None = None,
cwd: Path | None = None,
check: bool = True,
quiet: bool = False,
) -> int:
"""Run a command and optionally propagate its failure.
Args:
argv: The command and its arguments.
env: Environment for the child process; inherited when omitted.
cwd: Working directory; the repository root when omitted.
check: Raise :class:`SystemExit` with the child's code on failure.
quiet: Discard the child's stdout. stderr is NEVER discarded - a quiet
step that fails must still say why, otherwise the runner aborts with
a bare exit code and the user sees nothing at all.
Returns:
The child's exit code.
Raises:
SystemExit: When ``check`` is set and the command failed.
"""
proc = subprocess.run(
[str(part) for part in argv],
env=env,
cwd=str(cwd or REPO_ROOT),
stdout=subprocess.DEVNULL if quiet else None,
stderr=None,
check=False,
)
if check and proc.returncode != 0:
raise SystemExit(proc.returncode)
return proc.returncode
def capture(
argv: Sequence[str],
*,
env: dict[str, str] | None = None,
cwd: Path | None = None,
timeout: float = 30.0,
) -> str | None:
"""Run a command and return its stripped stdout.
Args:
argv: The command and its arguments.
env: Environment for the child process; inherited when omitted.
cwd: Working directory; the repository root when omitted.
timeout: Seconds to wait before giving up.
Returns:
The stripped stdout, or None when the command is missing or failed.
"""
try:
proc = subprocess.run(
[str(part) for part in argv],
env=env,
cwd=str(cwd or REPO_ROOT),
capture_output=True,
text=True,
timeout=timeout,
check=False,
)
except (OSError, subprocess.SubprocessError):
return None
if proc.returncode != 0:
return None
return proc.stdout.strip()
# ---------------------------------------------------------------------------
# Port ownership (no lsof dependency)
# ---------------------------------------------------------------------------
def port_is_live(port: int, host: str = BACKEND_HOST, timeout: float = 0.6) -> bool:
"""Test whether something is listening on a local TCP port.
A plain loopback connect works identically on every OS, so liveness never
depends on ``lsof``.
Args:
port: TCP port to probe.
host: Interface to connect to.
timeout: Connect timeout in seconds.
Returns:
True when the connect succeeded.
"""
try:
with socket.create_connection((host, port), timeout=timeout):
return True
except OSError:
return False
def _owner_pid_windows(port: int) -> int | None:
"""Resolve the listening PID on Windows via ``Get-NetTCPConnection``."""
shell = shutil.which("powershell") or shutil.which("pwsh")
if not shell:
return None
script = (
f"$c = Get-NetTCPConnection -LocalPort {port} -State Listen -ErrorAction SilentlyContinue | "
"Select-Object -First 1; if ($c) { $c.OwningProcess }"
)
out = capture([shell, "-NoProfile", "-NonInteractive", "-Command", script])
if not out:
return None
match = re.search(r"\d+", out)
return int(match.group(0)) if match else None
def _owner_pid_posix(port: int) -> int | None:
"""Resolve the listening PID on POSIX via ``lsof`` or ``ss``."""
lsof = shutil.which("lsof")
if lsof:
out = capture([lsof, f"-tiTCP:{port}", "-sTCP:LISTEN"])
if out:
first = out.splitlines()[0].strip()
if first.isdigit():
return int(first)
ss_bin = shutil.which("ss")
if ss_bin:
out = capture([ss_bin, "-ltnp", f"sport = :{port}"])
if out:
match = re.search(r"pid=(\d+)", out)
if match:
return int(match.group(1))
return None
def port_owner_pid(port: int) -> int | None:
"""Resolve the PID owning a listening TCP port.
Args:
port: TCP port to inspect.
Returns:
The owning PID, or None when it could not be determined. None means
"unknown" - never "free"; use :func:`port_is_live` for liveness.
"""
if IS_WINDOWS:
return _owner_pid_windows(port)
return _owner_pid_posix(port)
def describe_port(port: int) -> str:
"""Describe a port's state in one line.
Args:
port: TCP port to inspect.
Returns:
A human-readable state such as ``in use (PID 1234)`` or ``free``.
"""
if not port_is_live(port):
return "free"
pid = port_owner_pid(port)
if pid is None:
return "in use (owner unknown)"
return f"in use (PID {pid})"
def terminate_pid(pid: int) -> bool:
"""Terminate a process by PID, including its children on Windows.
Args:
pid: The process identifier.
Returns:
True when the terminate request was accepted.
"""
if IS_WINDOWS:
taskkill = shutil.which("taskkill")
if taskkill:
try:
proc = subprocess.run(
[taskkill, "/PID", str(pid), "/T", "/F"],
capture_output=True,
check=False,
timeout=3,
)
except (OSError, subprocess.TimeoutExpired):
return False
return proc.returncode == 0
try:
os.kill(pid, signal.SIGTERM)
except OSError:
return False
return True
# ---------------------------------------------------------------------------
# Canonical paths
# ---------------------------------------------------------------------------
def workspace_dir() -> Path:
"""Resolve the canonical per-OS FlintTrade workspace directory.
MIRROR: this is a byte-for-byte behavioural twin of
``flinttrade_core.workspace.workspace_dir`` / ``_default_home`` in
``packages/core/core/src/flinttrade_core/workspace.py``. The two MUST agree;
``tests/test_ft_runner.py`` compares them on every platform and both env
overrides. It is deliberately duplicated rather than imported because
``scripts/ft.py`` is stdlib-only and has to run before any dependency (and
therefore ``flinttrade_core``) is installed - it is what installs them.
Precedence matches the core resolver: ``FLINTTRADE_WORKSPACE_DIR`` beats
``FLINTTRADE_HOME``, which beats the platform default (``~/.flinttrade`` on
Linux, ``~/Library/Application Support/flinttrade`` on macOS,
``%APPDATA%/flinttrade`` on Windows). Both overrides are resolved, exactly as
the core does; the platform defaults are not.
Returns:
The workspace directory. Unlike the core resolver this helper never
creates the directory - ``status`` must be able to report its absence.
"""
for name in ("FLINTTRADE_WORKSPACE_DIR", "FLINTTRADE_HOME"):
value = os.environ.get(name)
if value:
return Path(value).expanduser().resolve()
system = platform.system()
if system == "Darwin":
return Path.home() / "Library" / "Application Support" / "flinttrade"
if system == "Windows":
appdata = os.environ.get("APPDATA")
if appdata:
return Path(appdata) / "flinttrade"
return Path.home() / "AppData" / "Roaming" / "flinttrade"
return Path.home() / ".flinttrade"
def dir_size_bytes(path: Path) -> int:
"""Total size of every regular file below *path*.
Replaces ``du -sh``, which does not exist on Windows.
Args:
path: Directory to measure.
Returns:
The total size in bytes; 0 when the directory is missing.
"""
total = 0
for root, _dirnames, filenames in os.walk(path, onerror=lambda _e: None):
for filename in filenames:
try:
total += (Path(root) / filename).stat().st_size
except OSError:
continue
return total
def format_size(num_bytes: int) -> str:
"""Render a byte count in human-readable units.
Args:
num_bytes: Size in bytes.
Returns:
A short string such as ``12.4MB``.
"""
size = float(num_bytes)
for unit in ("B", "KB", "MB", "GB"):
if size < 1024.0:
return f"{size:.0f}{unit}" if unit == "B" else f"{size:.1f}{unit}"
size /= 1024.0
return f"{size:.1f}TB"
def read_version() -> str:
"""Read the repository VERSION file.
Returns:
The version string, or ``unknown`` when the file is absent.
"""
version_file = REPO_ROOT / "VERSION"
try:
return version_file.read_text(encoding="utf-8").strip() or "unknown"
except OSError:
return "unknown"
# ---------------------------------------------------------------------------
# Node toolchain
# ---------------------------------------------------------------------------
def pnpm_argv() -> list[str] | None:
"""Resolve a pinned pnpm invocation prefix.
Mirrors the installer's resolution chain: corepack first, then a pnpm binary
already on ``PATH`` at the pinned version, then ``npx --yes pnpm@<pinned>``.
Returns:
The argv prefix to which pnpm arguments are appended, or None when no
Node toolchain is available.
"""
corepack = shutil.which("corepack")
if corepack:
return [corepack, "pnpm"]
pnpm = shutil.which("pnpm")
if pnpm and capture([pnpm, "--version"]) == PINNED_PNPM_VERSION:
return [pnpm]
npx = shutil.which("npx")
if npx:
return [npx, "--yes", f"pnpm@{PINNED_PNPM_VERSION}"]
return None
def require_pnpm() -> list[str]:
"""Resolve pnpm or abort with remediation text.
Returns:
The pnpm argv prefix.
Raises:
SystemExit: When no Node toolchain could be found.
"""
argv = pnpm_argv()
if argv is None:
fail("No Node toolchain found (corepack, pnpm or npx).")
fail("Install Node 22.23.4+ from https://nodejs.org/ and re-run.")
raise SystemExit(1)
return argv
def ci_env() -> dict[str, str]:
"""Build the environment used for Node commands.
``CI=true`` is set through the environment dictionary rather than a
``CI=true cmd`` shell prefix, which cmd.exe does not understand.
Returns:
A complete environment mapping with ``CI`` forced on.
"""
env = os.environ.copy()
env["CI"] = "true"
return env
# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------
INIT_ARGV: tuple[str, ...] = ("-m", "flinttrade_core.workspace_maintenance", "provision")
"""The first-run workspace/master-password provisioning step."""
INIT_VERBOSE_ARGV: tuple[str, ...] = (*INIT_ARGV, "--verbose")
"""The same step with the full traceback enabled, for the re-run hint."""
EXIT_PROVISION_DEGRADED = 3
"""Provisioning failed but left a usable vault secret; start anyway.
Mirrors ``flinttrade_core.cli.EXIT_PROVISION_DEGRADED``. Refusing to launch when
the secret the backend actually needs is present and hardened turns a transient
hiccup into an outage, which is precisely the first-run failure this guards.
"""
def provision_workspace(python: str, env: dict[str, str]) -> None:
"""Run the first-run workspace provisioning step.
The step's chatty stdout is suppressed (it runs before every ``start``,
``dev`` and ``setup``), but its stderr is left attached so a genuine failure
- a missing dependency, an unwritable workspace, a traceback - is visible.
Silence here reads as a hang, which is exactly the first-run experience this
guards against.
Args:
python: The interpreter resolved by :func:`resolve_python`.
env: The child environment from :func:`python_env`.
Raises:
SystemExit: When provisioning failed, after printing what to re-run.
"""
code = run([python, *INIT_ARGV], env=env, check=False, quiet=True)
if code == 0:
return
if code == EXIT_PROVISION_DEGRADED:
warn("Workspace provisioning did not complete, but the existing credential-vault")
warn("secret is present and hardened, so FlintTrade is starting anyway. The cause")
warn("is printed above; re-run the command below if it recurs.")
warn(f" {python} " + " ".join(INIT_VERBOSE_ARGV))
return
fail(f"Workspace initialisation failed (exit {code}); any error above came from that step.")
fail("The failure is often transient - a lock or a file still held by another")
fail("process - so re-running can simply succeed. Run it from this directory to")
fail("keep the same interpreter and module search path, and add the traceback:")
fail(f" {python} " + " ".join(INIT_VERBOSE_ARGV))
raise SystemExit(code)
def cmd_start(_args: list[str]) -> int:
"""Start the FlintTrade backend API.
Args:
_args: Unused positional arguments.
Returns:
The backend's exit code.
"""
python = resolve_python()
env = python_env()
header("Starting FlintTrade")
info(f" Backend: {BACKEND_URL}")
info("")
provision_workspace(python, env)
return run([python, "-m", "flinttrade_core.app"], env=env, check=False)
def cmd_stop(args: list[str]) -> int:
"""Stop the exact owner only after its concrete safe-maintenance decision.
A port/PID or executable name is never authority. The core helper owns the
bounded drain, native exit proof and retained pending-cleanup transaction.
"""
python = resolve_python()
return run(
[python, "-m", "flinttrade_core.workspace_maintenance", "stop", "--workspace", str(workspace_dir()), *args],
env=python_env(), check=False,
)
def cmd_restart(args: list[str]) -> int:
"""Stop then start the FlintTrade backend.
Args:
args: Forwarded to :func:`cmd_start`.
Returns:
The exit code of the failing step, or the backend's exit code.
"""
code = cmd_stop(args)
if code != 0:
fail("Restart aborted: the running backend could not be stopped.")
return code
return cmd_start(args)
def backend_responds() -> bool:
"""Check whether the backend answers its ping route.
Returns:
True when the backend produced any HTTP response.
"""
try:
with urllib.request.urlopen(f"{BACKEND_URL}/api/v1/ping", timeout=2): # noqa: S310 - fixed loopback URL
return True
except urllib.error.HTTPError:
# A 4xx/5xx still proves the backend is serving.
return True
except (urllib.error.URLError, OSError):
return False
def cmd_status(_args: list[str]) -> int:
"""Show backend, port, workspace and version status.
Args:
_args: Unused positional arguments.
Returns:
Always 0 - status reporting never fails the caller.
"""
header("FlintTrade Status")
info("\nFlintTrade backend:")
if backend_responds():
info(f" API: responding on {BACKEND_URL}")
else:
info(f" API: not responding on port {BACKEND_PORT}")
info("\nPorts:")
for port in (BACKEND_PORT, 5173, 3000):
info(f" :{port} - {describe_port(port)}")
info("\nWorkspace:")
workspace = workspace_dir()
if workspace.is_dir():
info(f" {workspace} - {format_size(dir_size_bytes(workspace))}")
else:
info(f" {workspace} - does not exist")
info(f"\nVersion: {read_version()}")
return 0
def _wait_dev_backend_exit(
proc: subprocess.Popen[bytes], *, stop_requested: Callable[[], bool] | None = None,
) -> None:
"""Observe POSIX exit without reaping the group leader and releasing its PID.
Python 3.14 is the supported floor, and it exposes ``waitid`` on macOS.
kqueue provides the same non-reaping notification when ``waitid`` is
absent. Polling lets recorded signals request cleanup without raising
asynchronously inside Popen, before we have recorded ownership of a child.
Windows retains Popen's wait and taskkill tree handling.
"""
if IS_WINDOWS:
proc.wait()
elif hasattr(os, "waitid"):
flags = os.WEXITED | os.WNOWAIT | (os.WNOHANG if stop_requested is not None else 0)
while stop_requested is None or not stop_requested():
if os.waitid(os.P_PID, proc.pid, flags) is not None:
return
time.sleep(0.05)
else:
with select.kqueue() as queue:
event = select.kevent(proc.pid, filter=select.KQ_FILTER_PROC,
flags=select.KQ_EV_ADD | select.KQ_EV_ONESHOT, fflags=select.KQ_NOTE_EXIT)
try:
changes = [event]
while stop_requested is None or not stop_requested():
if queue.control(changes, 1, 0.05 if stop_requested is not None else None):
return
changes = []
except ProcessLookupError:
pass # The still-unreaped child exited before registration.
def _stop_dev_processes(processes: Sequence[subprocess.Popen[bytes]], *, timeout: float = 3.0) -> None:
"""Stop only dev children created here, including their inherited process groups.
POSIX callers must spawn with ``start_new_session=True``: the recorded PID
then names our private group, never a group discovered from a port or an
arbitrary process. Retain exited leaders until every group signal is sent:
the unreaped child pins its PID/PGID against reuse. A reaped leader no longer
grants group authority. All groups share one graceful deadline.
"""
if IS_WINDOWS:
for proc in processes:
if proc.poll() is None:
terminate_pid(proc.pid)
else:
pending = {proc.pid for proc in processes if proc.returncode is None}
for pid in tuple(pending):
try:
os.killpg(pid, signal.SIGTERM)
except ProcessLookupError:
pending.remove(pid)
deadline = time.monotonic() + timeout
while pending and time.monotonic() < deadline:
for pid in tuple(pending):
try:
os.killpg(pid, 0)
except ProcessLookupError:
pending.remove(pid)
if pending:
time.sleep(min(0.05, max(0.0, deadline - time.monotonic())))
for pid in pending:
try:
os.killpg(pid, signal.SIGKILL)
except ProcessLookupError:
pass
for proc in processes:
try:
proc.wait(timeout=timeout)
except subprocess.TimeoutExpired:
proc.kill()
try:
proc.wait(timeout=timeout)
except subprocess.TimeoutExpired:
warn(f"Dev child PID {proc.pid} did not exit after forced termination.")
def cmd_dev(_args: list[str]) -> int:
"""Run the terminal dev server alongside the FlintTrade backend.
Args:
_args: Unused positional arguments.
Returns:
The backend's exit code, or 0 when interrupted.
"""
python = resolve_python()
env = python_env()
header("FlintTrade Dev Mode")
info(" Terminal: http://127.0.0.1:5173")
info(f" Backend: {BACKEND_URL}")
info("")
DEV_LOG_DIR.mkdir(parents=True, exist_ok=True)
provision_workspace(python, env)
backend_log = (DEV_LOG_DIR / "backend.log").open("wb")
terminal_log = (DEV_LOG_DIR / "terminal.log").open("wb")
processes: list[subprocess.Popen[bytes]] = []
received_signal: int | None = None
previous_handlers = {}
def request_stop(signum: int, _frame: object) -> None:
nonlocal received_signal
# Do not interrupt Popen before its child is recorded, or interrupt the
# bounded TERM/KILL cleanup on a repeated signal. The first reason wins.
if received_signal is None:
received_signal = signum
try:
if not IS_WINDOWS:
for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
# Respect the launcher's contract, notably nohup's ignored HUP.
if signal.getsignal(signum) == signal.SIG_IGN:
continue
previous_handlers[signum] = signal.signal(signum, request_stop)
backend = subprocess.Popen(
[python, "-m", "flinttrade_core.app"],
cwd=str(REPO_ROOT),
env=env,
stdout=backend_log,
stderr=subprocess.STDOUT,
start_new_session=not IS_WINDOWS,
)
processes.append(backend)
pnpm = pnpm_argv()
if pnpm is None:
warn("No Node toolchain found; the terminal dev server was not started.")
elif received_signal is None:
# Vite's dev server is interactive, so CI is deliberately not forced here.
terminal_env = os.environ.copy()
terminal_env.setdefault("VITE_FLINTTRADE_HOST", BACKEND_URL)
terminal = subprocess.Popen(
[*pnpm, "--dir", TERMINAL_PKG, "run", "dev"],
cwd=str(REPO_ROOT),
env=terminal_env,
stdout=terminal_log,
stderr=subprocess.STDOUT,
start_new_session=not IS_WINDOWS,
)
processes.append(terminal)
info(f"Started backend PID {backend.pid}.")
info("Logs: .local/dev-logs/backend.log and .local/dev-logs/terminal.log")
info("Press Ctrl+C to stop.")
try:
_wait_dev_backend_exit(backend, stop_requested=lambda: received_signal is not None)
except KeyboardInterrupt:
received_signal = signal.SIGINT
if received_signal is not None:
info("\nStopping dev processes...")
finally:
try:
_stop_dev_processes(processes)
finally:
for signum, handler in previous_handlers.items():
signal.signal(signum, handler)
for handle in (backend_log, terminal_log):
handle.close()
if received_signal == signal.SIGINT:
return 0
if received_signal is not None:
# Restore the original signal disposition only after our owned groups
# are gone; default handlers retain the CLI's signal exit status.
signal.raise_signal(received_signal)
return 128 + received_signal
return backend.returncode if backend.returncode is not None else 1
_IMPORT_FAILURE_MARKER = "FT-IMPORT-FAILURE "
"""Prefix the probe stamps on each failure line.
Importing thirteen packages can print anything at all, so the report is picked
out by marker rather than by assuming the child's stdout holds only the answer.
"""
_IMPORT_PROBE = """\
import io
import sys
marker = sys.argv[1]
failures = []
real_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
for name in sys.argv[2:]:
try:
__import__(name)
except BaseException as exc:
lines = str(exc).strip().splitlines()
summary = lines[0] if lines else ""
failures.append("%s%s: %s: %s" % (marker, name, type(exc).__name__, summary))
finally:
sys.stdout = real_stdout
for line in failures:
print(line)
"""
"""Really import each named workspace package and report what blew up.
This deliberately EXECUTES the packages instead of merely locating them.
``importlib.util.find_spec`` would answer a question nobody is asking: the names
come from :func:`workspace_module_names`, which discovers them by globbing the
very source roots :func:`python_env` puts on ``PYTHONPATH``, so a spec lookup can
only ever succeed - a tautology dressed up as a check. A real import is what
``start`` does, and it fails exactly where ``start`` would: on a third-party
dependency the install step did not deliver (``No module named 'dotenv'`` after a
partial ``pip install``), a broken native extension, or a package that raises on
import.
"""
def workspace_module_names() -> list[str]:
"""Return the importable ``flinttrade_*`` package names in this checkout.
Returns:
Sorted top-level module names, one per workspace Python distribution.
"""
return sorted(
{
package.name
for root in python_package_src_dirs()
for package in root.glob("flinttrade_*")
if (package / "__init__.py").is_file()
}
)
def missing_workspace_modules(python: str, env: dict[str, str]) -> list[str] | None:
"""Return one diagnostic per workspace package the prepared environment cannot import.
The probe runs against the interpreter and environment a real ``start``
uses - :func:`resolve_python` and :func:`python_env`, the same two the
backend is launched with - and imports each package for real, so a genuinely
broken environment fails here instead of at the user's next command.
Args:
python: The interpreter resolved by :func:`resolve_python`.
env: The child environment from :func:`python_env`.
Returns:
``[]`` when every package imports, a ``module: ErrorType: message`` line
per failure otherwise, or None when the probe itself could not be run
(interpreter unusable, or the imports outran the timeout) - an
unverifiable environment is not the same as a broken one.
"""
modules = workspace_module_names()
if not modules:
return []
argv = [python, "-c", _IMPORT_PROBE, _IMPORT_FAILURE_MARKER, *modules]
output = capture(argv, env=env, timeout=180.0)
if output is None:
return None
return [
line[len(_IMPORT_FAILURE_MARKER) :].strip()
for line in output.splitlines()
if line.startswith(_IMPORT_FAILURE_MARKER)
]
def cmd_setup(_args: list[str]) -> int:
"""Install Python and Node dependencies and initialise the workspace.
On POSIX this delegates to ``infra/scripts/setup.sh``. On Windows, where
that script's toolchain does not exist, an equivalent native sequence runs
instead.
Args:
_args: Unused positional arguments.
Returns:
0 on success, non-zero otherwise.
"""
setup_sh = REPO_ROOT / "infra" / "scripts" / "setup.sh"
bash = shutil.which("bash")
if not IS_WINDOWS and bash and setup_sh.is_file():
return run([bash, str(setup_sh)], check=False)
header("FlintTrade setup (native)")
python = resolve_python()
uv = shutil.which("uv")
if uv:
info("Syncing the Python workspace with uv...")
run([uv, "sync", "--frozen", "--all-packages", "--no-build-package", "growwapi"])
python = resolve_python()
repair_broker_sdk_environment(Path(python))
info("OK Python workspace synced")
else:
info("uv not found; falling back to the hash-verified pip baseline.")
lock = REPO_ROOT / "requirements.lock"
if not lock.is_file():
fail("requirements.lock is missing; install uv from https://docs.astral.sh/uv/ and re-run.")
return 1
run([python, "-m", "pip", "install", "--require-hashes", "--only-binary=growwapi", "-r", str(lock)])
repair_broker_sdk_environment(Path(python))
info("OK Third-party requirements and pinned broker SDKs installed")
# requirements.lock is exported with `--no-emit-workspace`, so it carries
# ONLY third-party dependencies: none of the thirteen flinttrade_* packages
# is installed by that command. They are reached through the PYTHONPATH
# that python_env() builds from every workspace source root, which is what
# keeps `start` from dying on `import flinttrade_data`. Git-pinned broker
# SDKs excluded from that registry-only export are installed and attested
# separately by broker_sdk_environment.py above.
info(f"OK Workspace packages linked on PYTHONPATH ({len(workspace_module_names())} packages)")
pnpm = pnpm_argv()
if pnpm is None:
warn("No Node toolchain found; the terminal, site and desktop packages were skipped.")
warn("Install Node 22.23.4+ from https://nodejs.org/ and re-run setup.")
else:
info("Installing the pnpm workspace...")
run([*pnpm, "install", "--frozen-lockfile"], env=ci_env())
info("OK workspace node_modules (pnpm --frozen-lockfile)")
# Verify before claiming success. A setup that reports OK and leaves an
# environment whose very next step (`start`) dies on ModuleNotFoundError is
# worse than a setup that fails: the user trusts the wrong thing. `python`
# and `env` here are the exact pair `cmd_start` hands to the backend, and
# the probe imports for real, so whatever would break `start` breaks this.
env = python_env()
failures = missing_workspace_modules(python, env)
if failures is None:
warn("Could not run the import check, so this setup is UNVERIFIED.")
warn(f'Confirm it by hand before relying on it: {python} -c "import flinttrade_core"')
elif failures:
fail("Setup did NOT produce a runnable environment; these packages fail to import:")
for line in failures:
fail(f" {line}")
fail("Install uv from https://docs.astral.sh/uv/ and re-run:")
fail(" uv sync --frozen --all-packages")
return 1
else:
info(f"OK Every workspace package imports cleanly ({len(workspace_module_names())} packages)")
info("Initialising the workspace...")
provision_workspace(python, env)
info(f"OK Workspace initialised: {workspace_dir()}")
info("")
info("Next steps:")
info(" 1. Run: python scripts/ft.py start")
info(" 2. Open the terminal app and finish setup")
info(" 3. Run: python scripts/ft.py status")
return 0
def pytest_paths() -> list[str]:
"""Collect the pytest target directories.
The repository nests packages three levels deep (``packages/<group>/<pkg>``),
so the glob is expanded here rather than relying on a shell that may not
exist.
Returns:
Repo-relative POSIX paths for every existing test directory.
"""
paths = sorted(
path.relative_to(REPO_ROOT).as_posix()
for path in REPO_ROOT.glob("packages/*/*/tests")
if path.is_dir()
)
if (REPO_ROOT / "tests").is_dir():
paths.append("tests")
# scripts/__tests__ does not match the packages glob and is not under tests/,
# so it ran nowhere - not here, not in CI - while holding 114 tests. One of
# them had been failing at head unnoticed because of it.
if (REPO_ROOT / "scripts" / "__tests__").is_dir():
paths.append("scripts/__tests__")
return paths
def _pytest_runner_options(args: Sequence[str]) -> tuple[list[str], int | None]:
"""Consume the runner's optional worker count while preserving pytest arguments."""
remaining: list[str] = []
workers: int | None = None
index = 0
while index < len(args):
arg = args[index]
if arg == "--":
remaining.extend(args[index:])
break
if arg == "--workers" or arg.startswith("--workers="):
if arg == "--workers":
index += 1
if index == len(args):
raise ValueError("--workers requires a non-negative integer")
value = args[index]
else:
value = arg.partition("=")[2]
try:
workers = int(value)
except ValueError as exc:
raise ValueError("--workers requires a non-negative integer") from exc
if workers < 0:
raise ValueError("--workers requires a non-negative integer")
else:
remaining.append(arg)
index += 1
return remaining, workers
_PYTEST_VALUE_OPTIONS = frozenset({
"-k", "-m", "-c", "--config-file", "-o", "-n", "--numprocesses", "--override-ini", "--maxfail",
"-W", "--pythonwarnings", "--max-warnings", "--pdbcls", "--lfnf", "--last-failed-no-failures",
"-r", "--report-chars", "--verbosity", "--pastebin", "--cache-show", "--debug",
"--deselect", "--ignore", "--ignore-glob", "--confcutdir", "--rootdir", "--basetemp",
"--doctest-report", "--doctest-glob",
"--junitxml", "--junit-xml", "--junitprefix", "--junit-prefix", "--capture", "--tb", "--show-capture", "--color",
"--code-highlight", "--durations", "--durations-min", "--timeout", "--timeout-method", "--timeout_method",
"--session-timeout", "--randomly-seed", "--dist", "--max-worker-restart", "--maxprocesses",
"--tx", "--px", "--testrunuid", "--cov", "--cov-report", "--cov-config", "--cov-fail-under", "--cov-context",
"--log-level", "--log-format", "--log-date-format", "--log-file", "--log-file-level", "--log-file-format",
"--log-file-date-format", "--log-file-mode", "--log-disable",
"--log-cli-level", "--log-cli-format", "--log-cli-date-format", "--log-auto-indent",
"-p", "--assert", "--import-mode", "--anyio-mode", "--asyncio-mode", "--asyncio-default-fixture-loop-scope",
"--asyncio-default-test-loop-scope", "--maxschedchunk", "--rsyncdir", "--rsyncignore",
"--ft-shard-index", "--ft-shard-count",
})
def _pytest_has_targets(args: Sequence[str]) -> bool:
"""Recognise explicit paths/node ids without mistaking selection values for paths."""
skip_value = False
positional_only = False
for arg in args:
if skip_value:
skip_value = False
continue
if arg == "--":
positional_only = True
continue
if not positional_only and arg in _PYTEST_VALUE_OPTIONS:
skip_value = True
continue
if not positional_only and arg.startswith("-"):
continue
target_path = arg.partition("::")[0]
if positional_only or "--pyargs" in args or "/" in target_path or "\\" in target_path:
return True
if target_path.endswith(".py") or (REPO_ROOT / target_path).exists():
return True
return False
def _pytest_selection_requested(args: Sequence[str]) -> bool:
"""Distinguish partial/display-only runs from full tests with output preferences."""
if _pytest_has_targets(args):
return True
selectors = {
"-k", "-m", "--deselect", "--ignore", "--ignore-glob", "--collect-only", "--collectonly", "--co",
"--lf", "--last-failed", "--lfnf", "--last-failed-no-failures", "--stepwise", "--sw",
"--stepwise-skip", "--sw-skip", "--stepwise-reset", "--sw-reset", "--fixtures", "--funcargs", "--fixtures-per-test",
"--setup-only", "--setuponly", "--setup-plan", "--setupplan", "--markers", "--cache-show",
"--ft-shard-index", "--ft-shard-count", "-h", "--help", "-V", "--version", "-f", "--looponfail",
}
skip_value = False
for arg in args:
if skip_value:
skip_value = False
continue
if arg == "--":
break
if arg.partition("=")[0] in selectors or arg.startswith(("-k", "-m")):
return True
if arg in _PYTEST_VALUE_OPTIONS:
skip_value = True
return False
def _pytest_requested_plugins(args: Sequence[str]) -> set[str]:
"""Read explicit plugin names without executing pytest's plugin configuration."""
plugins = set()
index = 0
while index < len(args):
arg = args[index]
if arg == "--":
break
if arg == "-p":
if index + 1 < len(args):
plugins.add(args[index + 1])
index += 1
elif arg.startswith("-p"):
plugins.add(arg[2:].removeprefix("="))
elif arg in _PYTEST_VALUE_OPTIONS:
index += 1
index += 1
return plugins
_PYTEST_PLUGIN_PROBE = """\
import json
import sys
from _pytest.config import get_config
from _pytest.config.exceptions import UsageError
args = sys.argv[1:]
config = get_config(args)
try:
# Parse configuration/plugins, but never import project conftests or collect
# tests. Unknown conftest-defined options are left to the real invocation.
try:
config.parse(['--noconftest', *args])
except UsageError:
pass
manager = config.pluginmanager
modules = {getattr(plugin, '__name__', '') for plugin in manager.get_plugins()}
print('FT-PYTEST-PLUGINS ' + json.dumps({
'xdist': 'xdist.plugin' in modules,
'timeout': 'pytest_timeout' in modules,
'timeout_blocked': any(manager.is_blocked(name) for name in ('timeout', 'pytest_timeout')),
}))
finally:
config._ensure_unconfigure()
"""
def _pytest_plugin_state(python: str, args: Sequence[str], env: dict[str, str]) -> dict[str, bool] | None:
"""Ask the selected pytest to resolve config, overrides and plugin load order.
The bootstrap runner remains stdlib-only. Keep this configuration probe
separate from optional module availability checks; importing xdist alone
says nothing about whether the selected configuration enables its hooks.
Invalid/unavailable interpreters or configurations still receive pytest's
own diagnostics when the real command runs.
"""
try:
result = subprocess.run(
[python, "-c", _PYTEST_PLUGIN_PROBE, *args], cwd=REPO_ROOT, env=env,
capture_output=True, text=True, check=False, timeout=10,
)
except (OSError, subprocess.TimeoutExpired):
return None
marker = "FT-PYTEST-PLUGINS "
for line in reversed(result.stdout.splitlines()):
if result.returncode == 0 and line.startswith(marker):
return json.loads(line[len(marker):])
return None
def pytest_argv(extra_flags: Sequence[str], *, env: dict[str, str] | None = None) -> list[str]:
"""Build a bounded pytest invocation, defaulting to every repository test root.
Args:
extra_flags: Flags appended after the target paths.
env: Exact child environment; defaults to the interactive pytest environment.
Returns:
The complete command, with optional xdist parallelism capped at four workers
unless ``--workers`` or ``FLINTTRADE_TEST_WORKERS`` overrides it.
"""
python = resolve_python()
env = pytest_env() if env is None else env
args, workers = _pytest_runner_options(extra_flags)
paths = [] if _pytest_has_targets(args) else pytest_paths()
if not paths and not _pytest_has_targets(args):
raise ValueError("No test directories found.")
parallel_args: list[str] = []
option_args = args[:args.index("--")] if "--" in args else args
explicit_xdist = any(arg == "-n" or arg.startswith(("-n", "--numprocesses")) for arg in option_args)
explicit_config = any(
arg.startswith(("-c", "--config-file=")) or arg == "--config-file"
for arg in option_args
)
configuration_args = [] if explicit_config else ["-c", str(REPO_ROOT / "pyproject.toml")]
state = _pytest_plugin_state(python, [*configuration_args, *paths, *args], env)
plugin_args = [*shlex.split(env.get("PYTEST_ADDOPTS", "")), *option_args]
plugins = _pytest_requested_plugins(plugin_args)
autoload_disabled = bool(env.get("PYTEST_DISABLE_PLUGIN_AUTOLOAD")) or (
"--disable-plugin-autoload" in plugin_args
)
# CLI -p and, from pytest 9.2, PYTEST_PLUGINS both accept installed
# plugin entry-point names (xdist, timeout) as well as importable module
# names (xdist.plugin, pytest_timeout). The live plugin probe is the
# source of truth; this fallback must use the same aliases so a failed
# probe does not omit -n or re-inject pytest-timeout when the selected
# pytest would already load the plugin.
environment_plugins = {name.strip() for name in env.get("PYTEST_PLUGINS", "").split(",") if name.strip()}
xdist_loaded = bool({"xdist", "xdist.plugin"} & (plugins | environment_plugins))
timeout_loaded = bool({"timeout", "pytest_timeout"} & (plugins | environment_plugins))
xdist_disabled = "no:xdist" in plugins or "no:xdist.plugin" in plugins or (autoload_disabled and not xdist_loaded)
timeout_blocked = bool({"no:timeout", "no:pytest_timeout"} & plugins)
if state is not None:
xdist_disabled = not state["xdist"]
timeout_loaded = state["timeout"]
timeout_blocked = not timeout_loaded and state["timeout_blocked"]
autoload_disabled = not timeout_loaded
if timeout_blocked:
raise ValueError("The test runner requires pytest-timeout; do not disable the timeout watchdog plugin.")
# A disabled plugin cannot register -n. Only automatic worker insertion is
# suppressed: an explicit, incompatible -n remains pytest's own decision.
if not explicit_xdist and not xdist_disabled:
if workers is None:
value = env.get("FLINTTRADE_TEST_WORKERS")
try:
workers = int(value) if value is not None else min(4, max(1, os.cpu_count() or 1))
except ValueError as exc:
raise ValueError("FLINTTRADE_TEST_WORKERS must be a non-negative integer") from exc
if workers < 0:
raise ValueError("FLINTTRADE_TEST_WORKERS must be a non-negative integer")
available = state["xdist"] if state is not None else (
capture([python, "-c", "import xdist"], env=env) is not None
)
if available:
# Zero must override -n inherited through pytest's config/addopts too.
parallel_args = ["-n", str(workers)]
elif workers > 0:
info("pytest-xdist unavailable - running pytest sequentially")
timeout_plugin = ["-p", "pytest_timeout"] if autoload_disabled and not timeout_loaded else []
bounds = [
*configuration_args,
*timeout_plugin,
*parallel_args,
"--tb=short",
"--import-mode=importlib",
# Match the CI invocation in .github/workflows/test.yml. Without these,
# a local run has NO time bound at all: a process-spawning test whose
# child never exits blocks the whole suite indefinitely, and because
# pytest prints its summary only at the end there is no output to say so
# - it looks identical to "still running". One such flake was measured
# holding a local run for 93 minutes, at 0.015s of CPU, before it was
# killed; the same suite finishes in 5 minutes once bounded.
#
# --timeout-method=thread uses a watchdog thread rather than SIGALRM,
# which cannot reliably interrupt CPython's thread.join(). That is the
# difference between a named test with a traceback and a silent hang.
f"--timeout={PYTEST_TIMEOUT_SECONDS}",
"--timeout-method=thread",
]
if "--" in args:
separator = args.index("--")
args = [*args[:separator], *bounds, *args[separator:]]
else:
args.extend(bounds)
return [python, "-m", "pytest", *paths, *args]
def _run_pytest(extra_flags: Sequence[str]) -> int:
"""Run selected pytest tests, preserving failures and the watchdog timeout."""
try:
argv = pytest_argv(extra_flags)
except ValueError as exc:
fail(str(exc))
return 2
return run(argv, env=pytest_env(), check=False)
def rust_test_env(python: str) -> dict[str, str]:
"""Pin PyO3 and expose the selected interpreter's Unix shared library directory.
uv's standalone Python keeps libpython outside the system loader search
path. Rust's test executable links that library, so it needs LIBDIR at
runtime as well as PYO3_PYTHON during compilation. Preserve any existing
loader paths and leave Windows/static-Python environments unchanged.
"""
env = python_env({"PYO3_PYTHON": python})
system = platform.system()
if system not in {"Linux", "Darwin"}:
return env
libdir = capture(
[python, "-I", "-c", "import sysconfig; print(sysconfig.get_config_var('LIBDIR') or '')"],
env=env,
)
if not libdir:
return env
directory = Path(libdir)
pattern = "libpython*.so*" if system == "Linux" else "libpython*.dylib"
try:
shared_library = directory.is_absolute() and any(path.is_file() for path in directory.glob(pattern))
except (OSError, ValueError):
shared_library = False
if shared_library:
variable = "LD_LIBRARY_PATH" if system == "Linux" else "DYLD_LIBRARY_PATH"
existing = env.get(variable, "")
# Both Unix loaders use ':', irrespective of the host running a platform test.
if libdir not in existing.split(":"):
env[variable] = libdir + (":" + existing if existing else "")
return env
def cmd_test(args: list[str]) -> int:
"""Run selected Python tests, or the full Python/Rust suites without selection.
Args:
args: Pytest paths/options, plus an optional runner ``--workers`` override.
Returns:
The first non-zero exit code, or 0.
"""
code = _run_pytest(["-v", *args])
if code != 0:
return code
selected_args, _workers = _pytest_runner_options(args)
if _pytest_selection_requested(selected_args):
return 0
cargo = shutil.which("cargo")
if cargo is None:
info("cargo not found - skipping Rust ticks tests")
return 0
# PyO3 searches the ambient PATH when PYO3_PYTHON is unset. The task runner
# can itself be launched from another tool's older virtualenv, so that search
# may select a different (and unsupported) Python than the repository venv
# which just ran pytest. Pin the Rust extension build to the same resolved
# interpreter instead of allowing PATH order to choose it implicitly.
python = resolve_python()
return run(
[cargo, "test", "--manifest-path", "packages/core/ticks/Cargo.toml"],
env=rust_test_env(python),
check=False,
)
def cmd_test_fast(args: list[str]) -> int:
"""Run pytest and stop on the first failure.
Args:
args: Pytest paths/options, plus an optional runner ``--workers`` override.
Returns:
pytest's exit code.
"""
args = list(args)
args.insert(args.index("--") if "--" in args else len(args), "-x")
return _run_pytest(args)
def cmd_check(args: list[str]) -> int:
"""Run the local check planner from its exact sibling path."""
checks = _load_module_from_path("_flinttrade_local_checks", Path(__file__).with_name("check_local.py"))
return checks.run_checks(args, runner=SimpleNamespace(**globals()))
def cmd_lint(_args: list[str]) -> int:
"""Run ruff over ``packages/`` and ``tests/``, then the terminal's hooks gate.
The two halves mirror the two languages: ruff for Python, and the terminal's
``lint`` script (``react-hooks/rules-of-hooks`` plus
``react-hooks/exhaustive-deps``) for the React sources. The hooks gate is the
same command CI runs in ``node-core-tests``, so a contributor hits it before
pushing rather than after.
A missing toolchain is reported with install guidance and is not a failure -
the JavaScript workspace is optional for a Python-only change, exactly as
ruff is optional for a terminal-only one. A genuine lint violation from
either half is propagated, and both halves always run so one report does not
hide the other.
Args:
_args: Unused positional arguments.
Returns:
The first non-zero exit code of the two lints, or 0 when both pass (or
are skipped for a missing toolchain).
"""
python = resolve_python()
env = python_env()
if capture([python, "-m", "ruff", "--version"]) is None:
info("ruff not installed. Install with: uv sync --frozen --all-packages (or: pip install ruff)")
ruff_code = 0
else:
ruff_code = run([python, "-m", "ruff", "check", "packages/", "tests/"], env=env, check=False)
pnpm = pnpm_argv()
if pnpm is None:
info("No Node toolchain found; skipped the terminal react-hooks lint.")
return ruff_code
if not (REPO_ROOT / TERMINAL_PKG / "node_modules").is_dir():
info("Terminal node_modules missing; run `pnpm install --frozen-lockfile` to include the hooks lint.")
return ruff_code
header("Linting the terminal (react-hooks)")
hooks_code = run([*pnpm, "--dir", TERMINAL_PKG, "run", "lint"], env=ci_env(), check=False)
return ruff_code or hooks_code
def _clean_targets() -> Iterator[Path]:
"""Yield every build-artefact directory that ``clean`` removes."""
for root, dirnames, _filenames in os.walk(REPO_ROOT):
root_path = Path(root)
for name in list(dirnames):
if name in {"__pycache__", ".pytest_cache", "node_modules"}:
dirnames.remove(name)
yield root_path / name
elif name in PRUNED_DIRS:
dirnames.remove(name)
def cmd_clean(args: list[str]) -> int:
"""Delete ``__pycache__``, ``.pytest_cache`` and ``node_modules`` trees.
Uses :func:`shutil.rmtree` rather than ``find -exec rm -rf``, so it behaves
identically on Windows.
Args:
args: Pass ``--yes`` (or ``-y``) to skip the confirmation prompt.
Returns:
0 on success or when the user declined; 1 when a tree could not be removed.
"""
assume_yes = any(arg in {"--yes", "-y"} for arg in args)
targets = list(_clean_targets())
if not targets:
info("Nothing to clean.")
return 0
info(f"This will remove {len(targets)} __pycache__, .pytest_cache and node_modules director(y/ies).")
if not assume_yes:
if not sys.stdin.isatty():
fail("Refusing to clean without confirmation; re-run with --yes.")
return 1
if input("Continue? [y/N] ").strip().lower() not in {"y", "yes"}:
info("Cancelled.")
return 0
failures = 0
for target in targets:
try:
shutil.rmtree(target)
except OSError as exc:
warn(f"could not remove {target}: {exc}")
failures += 1
if failures:
return 1
info("OK Cleaned")
return 0
def cmd_version(_args: list[str]) -> int:
"""Print the FlintTrade version.
Args:
_args: Unused positional arguments.
Returns:
Always 0.
"""
info(read_version())
return 0
def _make_targets() -> list[tuple[str, str]]:
"""Parse ``target: ## description`` pairs out of the Makefile.
This is the Python replacement for the Makefile's ``grep | sort | awk``
help pipeline, none of which exists on Windows.
Returns:
Sorted ``(target, description)`` pairs; empty when the Makefile is absent.
"""
makefile = REPO_ROOT / "Makefile"
try:
text = makefile.read_text(encoding="utf-8")
except OSError:
return []
pattern = re.compile(r"^([a-zA-Z_-]+):.*?## (.*)$", re.MULTILINE)
return sorted((match.group(1), match.group(2).strip()) for match in pattern.finditer(text))
def cmd_help(_args: list[str]) -> int:
"""Show the command table.
Args:
_args: Unused positional arguments.
Returns:
Always 0.
"""
info("FlintTrade - cross-platform task runner")
info("")
info(" python scripts/ft.py <command>")
info(" flinttrade <command> (after install, via the shim)")
info("")
info("Commands (Windows, macOS and Linux):")
for name, description in COMMANDS.items():
info(f" {name:<17} {description}")
targets = _make_targets()
if targets:
info("")
info("make targets (some are POSIX-only or need Docker - see the Makefile header):")
for name, description in targets:
info(f" {name:<17} {description}")
return 0
def _pnpm_desktop(script: str) -> int:
"""Run a pnpm script inside the desktop package.
Args:
script: The package.json script name.
Returns:
The command's exit code.
"""
pnpm = require_pnpm()
return run([*pnpm, "--dir", DESKTOP_PKG, "run", script], env=ci_env(), check=False)
def _pnpm_workspace_install() -> int:
"""Install the pnpm workspace with the committed lockfile.
Returns:
The command's exit code.
"""
pnpm = require_pnpm()
return run([*pnpm, "install", "--frozen-lockfile"], env=ci_env(), check=False)
def cmd_desktop_test(_args: list[str]) -> int:
"""Typecheck and test the Electron shell.
Args:
_args: Unused positional arguments.
Returns:
The first non-zero exit code, or 0.
"""
for step in (_pnpm_workspace_install, lambda: _pnpm_desktop("typecheck"), lambda: _pnpm_desktop("test:electron")):
code = step()
if code != 0:
return code
return 0
def cmd_desktop_build(_args: list[str]) -> int:
"""Verify and bundle the Electron shell.
Args:
_args: Unused positional arguments.
Returns:
The first non-zero exit code, or 0.
"""
code = _pnpm_workspace_install()
if code != 0:
return code
return _pnpm_desktop("build")
def desktop_pack_script() -> str:
"""Choose the electron-builder pack script for this host.
Returns:
One of ``pack:win``, ``pack:mac``, ``pack:linux:x64`` or ``pack:linux:arm64``.
Raises:
SystemExit: When the host OS/architecture is unsupported.
"""
system = platform.system()
if system == "Windows":
return "pack:win"
if system == "Darwin":
return "pack:mac"
machine = platform.machine().lower()
if system == "Linux":
if machine in {"x86_64", "amd64"}:
return "pack:linux:x64"
if machine in {"arm64", "aarch64"}:
return "pack:linux:arm64"
fail(f"Unsupported desktop package target: {system}/{platform.machine()}")
raise SystemExit(1)
def cmd_desktop_package(_args: list[str]) -> int:
"""Package and verify the Electron installer for this OS/arch.
The standalone pack scripts own preparation, typechecking, tests, bundling
and package verification. Running ``build`` first repeats that preparation
and replaces the native helpers after their tests have already passed.
Args:
_args: Unused positional arguments.
Returns:
The first non-zero exit code, or 0.
"""
code = _pnpm_workspace_install()
if code != 0:
return code
code = _pnpm_desktop(desktop_pack_script())
if code != 0:
return code
info("OK Electron package under packages/apps/desktop/release/electron/")
return 0
def cmd_desktop_dev(_args: list[str]) -> int:
"""Run the Electron shell against its managed source bootstrap.
Args:
_args: Unused positional arguments.
Returns:
The first non-zero exit code, or 0.
"""
code = _pnpm_workspace_install()
if code != 0:
return code
return _pnpm_desktop("dev")
HANDLERS: dict[str, Callable[[list[str]], int]] = {
"start": cmd_start,
# `make start-gateway` is a plain alias for `make start`; the runner mirrors
# it so the Makefile header's command list is dispatchable in full.
"start-gateway": cmd_start,
"stop": cmd_stop,
"restart": cmd_restart,
"status": cmd_status,
"dev": cmd_dev,
"setup": cmd_setup,
"test": cmd_test,
"test-fast": cmd_test_fast,
"check": cmd_check,
"lint": cmd_lint,
"clean": cmd_clean,
"version": cmd_version,
"help": cmd_help,
"desktop-test": cmd_desktop_test,
"desktop-build": cmd_desktop_build,
"desktop-package": cmd_desktop_package,
"desktop-dev": cmd_desktop_dev,
}
def main(argv: Sequence[str] | None = None) -> int:
"""Dispatch a subcommand.
Args:
argv: Argument vector without the program name; ``sys.argv[1:]`` when omitted.
Returns:
The subcommand's exit code; 2 for an unrecognised command.
"""
args = list(sys.argv[1:] if argv is None else argv)
if not args or args[0] in {"-h", "--help", "help"}:
return cmd_help([])
name, rest = args[0], args[1:]
handler = HANDLERS.get(name)
if handler is None:
fail(f"Unknown command: {name}")
info("")
cmd_help([])
return 2
try:
return handler(rest)
except KeyboardInterrupt:
info("\nInterrupted.")
return 130
if __name__ == "__main__":
raise SystemExit(main())