tests/conftest.py
Rama main · réplica de lectura del repositorio original.
"""Shared pytest configuration and fixtures for Omnigent tests."""
from __future__ import annotations
import contextlib
import os
import shutil
import sys
import tempfile
import time
from collections.abc import Callable, Generator
from pathlib import Path
import pytest
try:
import resource as _resource # POSIX-only; absent on Windows.
except ImportError:
_resource = None # type: ignore[assignment]
# Establish test data isolation before importing any Omnigent modules. This
# deliberately replaces ambient state; subprocesses inherit the safe override.
_TEST_OMNIGENT_DATA_DIR = Path(tempfile.mkdtemp(prefix="omnigent-pytest-")).resolve()
os.environ["OMNIGENT_DATA_DIR"] = str(_TEST_OMNIGENT_DATA_DIR)
# Servers and runners booted by tests otherwise sweep the developer's live
# harness root. Prefer /tmp for the AF_UNIX path-length limit; some sandboxes
# only allow the default temp dir.
try:
_TEST_HARNESS_TMP_PARENT = Path(tempfile.mkdtemp(prefix="omni-h-", dir="/tmp"))
except OSError:
_TEST_HARNESS_TMP_PARENT = Path(tempfile.mkdtemp(prefix="omni-h-"))
os.environ["OMNIGENT_HARNESS_TMP_PARENT"] = str(_TEST_HARNESS_TMP_PARENT)
# Tests and their subprocesses must not ship logs through the caller's sink.
# Sink tests configure their own credentials or sender explicitly.
for _test_env_name in tuple(os.environ):
if _test_env_name.startswith("OMNIGENT_DEBUG_LOG_"):
os.environ.pop(_test_env_name)
# Skip the synchronous api.litellm.ai/model_catalog HTTP fallback during
# tests. Hardened CI runners can't reach the public internet, so every
# workflow startup that misses litellm's local registry would otherwise
# block 5 s on the timeout. ``setdefault`` so a developer can opt
# back in by exporting the var with any other value when exercising the
# catalog code path explicitly.
os.environ.setdefault("OMNIGENT_DISABLE_CATALOG_LOOKUP", "1")
# Pin header mode for the whole suite. Header is the env-unset default,
# but a developer's shell often has OMNIGENT_AUTH_ENABLED=1 set (the
# multi-user opt-in they use to test the login flow locally) — and that
# enable switch would flip the env-unset default to accounts (or oidc, if
# the shell also exports OMNIGENT_OIDC_ISSUER), booting every server in
# multi-user mode and failing loud with "Missing required environment
# variable OMNIGENT_ACCOUNTS_COOKIE_SECRET" / "Authentication required"
# (401). An explicit AUTH_PROVIDER always wins over the enable switch, so
# pinning it here keeps tests deterministic regardless of the ambient
# shell. Accounts/OIDC-specific tests still opt in by monkeypatching the
# vars inside their own fixtures (tests/server/test_accounts.py,
# tests/server/test_oidc.py). Module-level setdefault rather than a fixture
# so subprocess-spawning tests (e2e shells out to `omnigent run`) inherit
# the pin via env.
os.environ.setdefault("OMNIGENT_AUTH_PROVIDER", "header")
# Mark the whole suite a single-user local runtime. Header mode now
# fails closed on a missing X-Forwarded-Email: a request
# without the header is rejected with 401 instead of resolving to the
# shared "local" identity. Test servers and the subprocesses they spawn
# have no proxy injecting the header and drive headerless traffic
# (runner-status polls, REPL turns, session CRUD), so they need the
# single-user fallback that the managed local-server spawn paths set in
# production. Pinned here (not per-fixture) so every spawned server
# inherits it via os.environ — the same chokepoint as the header pin
# above. Tests that specifically verify the strict (deployed
# multi-user) posture opt OUT by constructing
# UnifiedAuthProvider(source="header", local_single_user=False) or by
# monkeypatch.delenv-ing this var.
os.environ.setdefault("OMNIGENT_LOCAL_SINGLE_USER", "1")
from omnigent.db.utils import _engine_cache, _engine_lock, get_or_create_engine # noqa: E402
from omnigent.runtime.filesystem_registry import GitFilesystemRegistry # noqa: E402
from tests import _model_pools # noqa: E402
pytest_plugins = ["tests._token_usage"]
def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
"""Translate ``@pytest.mark.llm_flaky`` into a rerunfailures ``flaky``
marker; each rerun resolves to a different model via
:mod:`tests._model_pools` rotation.
"""
is_windows = os.name == "nt"
skip_posix = pytest.mark.skip(reason="POSIX-only test; skipped on Windows")
skip_windows = pytest.mark.skip(reason="Windows-only test; skipped on POSIX")
for item in items:
llm_flaky = item.get_closest_marker("llm_flaky")
if llm_flaky is not None:
# WARNING: never llm_flaky a heavy e2e test that can hit the
# CI --timeout=180 cap: thread-timeout kill + loadscope +
# rerun can crash the whole xdist shard.
reruns = int(llm_flaky.kwargs.get("reruns", 2))
delay = int(llm_flaky.kwargs.get("reruns_delay", 1))
item.add_marker(pytest.mark.flaky(reruns=reruns, reruns_delay=delay))
# Auto-skip platform-pinned tests on the wrong OS so the Linux suite
# is unchanged and a Windows run doesn't choke on POSIX-only tests.
if item.get_closest_marker("posix_only") is not None and is_windows:
item.add_marker(skip_posix)
if item.get_closest_marker("windows_only") is not None and not is_windows:
item.add_marker(skip_windows)
# Per-worker progress log path; resolved from
# ``PYTEST_PROGRESS_LOG_DIR`` in :func:`pytest_configure`. ``None``
# when env var is unset (local dev).
_PROGRESS_LOG_PATH: str | None = None
def pytest_configure(config: pytest.Config) -> None:
"""Resolve the per-worker progress log path and run guardrails."""
global _PROGRESS_LOG_PATH
log_dir = os.environ.get("PYTEST_PROGRESS_LOG_DIR")
if log_dir:
os.makedirs(log_dir, exist_ok=True)
worker = os.environ.get("PYTEST_XDIST_WORKER", "main")
_PROGRESS_LOG_PATH = os.path.join(log_dir, f"progress-{worker}.log")
_run_test_environment_guardrails(config)
def _run_test_environment_guardrails(config: pytest.Config) -> None:
"""Enforce test-environment guardrails at session start.
Hard-fail: :func:`check_test_environment` raises on anything that
looks like a real (non-test) DB or a base URL aimed at a dev/prod host
or port. Set ``OMNIGENT_DISABLE_TEST_GUARDRAILS=1`` to temporarily
downgrade violations to warn-only for deliberate integration runs.
"""
from omnigent.testing.guardrails import check_test_environment
db_uri = os.environ.get("OMNIGENT_DATABASE_URI", "")
base_url = config.getoption("--omnigent-server-url", default=None)
check_test_environment(db_uri=db_uri, base_url=base_url, warn_only=False)
def pytest_unconfigure(config: pytest.Config) -> None:
"""Clean up per-session resources.
Reap before removing the data dir: tests spawn real detached Omnigent
processes (host daemons, local servers, runner zygotes) that would
otherwise outlive the session — squatting port 6767 and serving a
deleted database. Reaping first also lets attribution use the live
directory path while orphans still reference it.
"""
from omnigent.testing.process_reaper import reap_leaked_omnigent_processes
reaped, survivors = reap_leaked_omnigent_processes(_TEST_OMNIGENT_DATA_DIR)
for cmdline in reaped:
print(f"\nreaped leaked omnigent process: {cmdline}", file=sys.stderr)
for cmdline in survivors:
print(f"\nUNREAPED omnigent process survived SIGKILL: {cmdline}", file=sys.stderr)
shutil.rmtree(_TEST_OMNIGENT_DATA_DIR, ignore_errors=True)
shutil.rmtree(_TEST_HARNESS_TMP_PARENT, ignore_errors=True)
# Per-worker progress logger: fsync'd START/END lines so a
# wedged worker leaves the last test on disk. END lines also carry
# peak RSS; `pytest_terminal_summary` prints the top tests by RSS
# delta to flag OOM-shaped hangs.
def _process_peak_rss_kb() -> int | None:
"""Peak RSS in KB. None on Windows. ru_maxrss is bytes on macOS."""
if _resource is None:
return None
rss = _resource.getrusage(_resource.RUSAGE_SELF).ru_maxrss
if sys.platform == "darwin":
rss //= 1024
return int(rss)
_TEST_RSS_RECORDS: list[tuple[str, int]] = []
def _write_progress_event(event: str, nodeid: str, rss_kb: int | None = None) -> None:
"""Append ``<timestamp>\\t<event>\\t<nodeid>[\\t<rss_kb>]\\n`` and fsync."""
if _PROGRESS_LOG_PATH is None:
return
line = f"{time.time():.3f}\t{event}\t{nodeid}"
if rss_kb is not None:
line += f"\t{rss_kb}"
line += "\n"
with open(_PROGRESS_LOG_PATH, "a") as f:
f.write(line)
f.flush()
os.fsync(f.fileno())
def pytest_runtest_logstart(nodeid: str, location: tuple[str, int | None, str]) -> None:
_write_progress_event("START", nodeid)
def pytest_runtest_logfinish(nodeid: str, location: tuple[str, int | None, str]) -> None:
rss_kb = _process_peak_rss_kb()
if rss_kb is not None:
_TEST_RSS_RECORDS.append((nodeid, rss_kb))
_write_progress_event("END", nodeid, rss_kb=rss_kb)
# Clear so resolutions outside any test pass through unchanged.
_model_pools.set_current_test(None)
def pytest_runtest_setup(item: pytest.Item) -> None:
"""Stamp the model-pool context for this test attempt.
Runs once per rerunfailures attempt (``item.execution_count`` is
bumped before each), so reruns rotate to a different model.
:param item: The test item about to run.
"""
# execution_count is 1-based and absent without a flaky marker.
attempt = getattr(item, "execution_count", 1) - 1
_model_pools.set_current_test(
item.nodeid,
attempt=attempt,
pinned=item.get_closest_marker("model_pinned") is not None,
)
def pytest_terminal_summary(
terminalreporter: pytest.TerminalReporter,
exitstatus: int,
config: pytest.Config,
) -> None:
"""Top tests by peak-RSS delta -- per worker."""
if len(_TEST_RSS_RECORDS) < 2:
return
baseline = 0
deltas: list[tuple[str, int, int]] = []
for nodeid, rss_kb in _TEST_RSS_RECORDS:
delta = rss_kb - baseline
if delta > 0:
deltas.append((nodeid, rss_kb, delta))
baseline = rss_kb
if not deltas:
return
deltas.sort(key=lambda row: row[2], reverse=True)
terminalreporter.write_sep("=", "Top tests by peak-RSS delta")
for nodeid, rss_kb, delta_kb in deltas[:20]:
terminalreporter.write_line(
f"+{delta_kb / 1024:7.1f} MB (now {rss_kb / 1024:8.1f} MB) {nodeid}"
)
def pytest_addoption(parser):
"""Register CLI flags consumed across the suite.
:param parser: the pytest option parser.
"""
parser.addoption(
"--integration",
action="store_true",
default=False,
help="Run integration tests (requires real LLM credentials)",
)
parser.addoption(
"--model",
action="store",
default="databricks-claude-sonnet-4-6",
help="Model name for integration tests (default: databricks-claude-sonnet-4-6)",
)
parser.addoption(
"--harness",
action="store",
default="databricks",
help=(
"Harness type: 'databricks', 'claude-sdk', 'open-responses', "
"'openai-agents', or 'codex' (default: databricks)"
),
)
parser.addoption(
"--profile",
action="store",
default="",
help="Databricks config profile for integration tests",
)
parser.addoption(
"--llm-api-key",
action="store",
default=None,
help=(
"LLM API key for integration / e2e tests. Required when running "
"tests/e2e/ (those tests assert it's non-None via the live_server "
"fixture); optional for tests/frontends/ integration tests "
"(those skip gracefully when the key is absent)."
),
)
parser.addoption(
"--omnigent-server-url",
action="store",
default=None,
help=(
"Base URL of an externally-managed `omnigent.cli server` to run "
"e2e tests against, e.g. `http://localhost:8080`. When set, "
"server-fixtures skip the spawn step and yield this URL. Useful "
"for iterating on tests against a long-running dev server "
"(server logs stay visible, breakpoints stick across runs). When "
"unset, fixtures spawn a fresh subprocess as before. The fixture "
"consumer is responsible for ensuring the external server is "
"configured with the credentials/profile the test needs."
),
)
@pytest.fixture(autouse=True)
def _no_workspace_probe(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep tests off the network: the host's workspace probe answers "unknown".
Any CLI path that starts or checks a host daemon on a server URL probes
that server. Tests that need an answer patch the probe themselves.
:param monkeypatch: Pytest monkeypatch fixture.
"""
monkeypatch.setattr(
"omnigent.host.daemon_lifecycle.probe_workspace_id", lambda url, **_kwargs: None
)
@pytest.fixture
def frozen_recurrence_clock(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep a slow runner from tripping the RRULE search's CPU-time limit.
Searches then end only on the deterministic work budget. Tests of the time
limit patch this clock themselves, which overrides the frozen one.
:param monkeypatch: Pytest monkeypatch fixture.
"""
monkeypatch.setattr("omnigent.server.scheduled._bounded_rrule.thread_time", lambda: 0.0)
@pytest.fixture(autouse=True)
def _reset_runner_catalog_cache() -> Generator[None, None, None]:
"""
Clear smart routing's per-session runner-catalog cache after every test.
The cache is process-global and keyed by session id, and the suite reuses
a handful of fixed ids (``conv_123`` and friends) across unrelated tests —
so one test's catalog would answer another test's fetch and its counting
stub would never be called.
:returns: None.
"""
yield
# sys.modules lookup, not an import: the spec lane blocks omnigent imports
# inside some tests, and a lane that never touched smart_routing should not
# pay for loading it in every teardown.
module = sys.modules.get("omnigent.server.smart_routing")
if module is not None:
module._runner_catalog_cache.clear()
@pytest.fixture(autouse=True)
def _reset_databricks_auth_memory() -> None:
"""
Start every test without shared Databricks token sources or remembered profiles.
Both are meant to outlive a single client: the token sources are
process-global and keyed by server URL, which tests reuse, and the
remembered profiles live in the shared test data dir. Either would let one
test's resolution answer another's. The reset runs before the test, not
at teardown, where a test's ``Path.unlink`` patch can still be active.
:returns: None.
"""
chat = sys.modules.get("omnigent.chat")
if chat is not None:
chat._databricks_auth_cache.clear()
executor = sys.modules.get("omnigent.inner.databricks_executor")
if executor is not None:
with contextlib.suppress(FileNotFoundError):
os.remove(_TEST_OMNIGENT_DATA_DIR / executor._REMEMBERED_PROFILES_FILE)
@pytest.fixture(autouse=True)
def _isolate_claude_native_state(
tmp_path_factory: pytest.TempPathFactory,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
Redirect claude-native client-side persistent state to a tmp dir.
The ``omnigent claude`` wrapper writes per-conversation
launch state (the cwd a session was created in) under
``~/.omnigent/claude-native/<hash>/launch.json``. Any test
that drives the wrapper -- directly or indirectly via test
fakes that invoke its helpers -- would otherwise write to the
developer's real ``~/.omnigent`` directory and pollute it
across test runs.
The state module honors :data:`OMNIGENT_CLAUDE_NATIVE_STATE_DIR`
as a root override. ``autouse=True`` because the alternative
(opt-in fixture per test) leaves us one missed test away from
re-polluting the user's home; the override has no side effects
on tests that don't touch claude-native state at all.
Using ``tmp_path_factory.mktemp`` rather than the request-scoped
``tmp_path`` so the override fires before any other fixture or
test body picks up the env -- ``tmp_path`` materializes lazily
per test, and we want the redirect to be in effect from the
moment the test session starts.
:param tmp_path_factory: Pytest's session-scoped temp factory.
:param monkeypatch: Pytest monkeypatch fixture; auto-restores
the env var at teardown.
:returns: None.
"""
state_dir = tmp_path_factory.mktemp("claude-native-state")
monkeypatch.setenv("OMNIGENT_CLAUDE_NATIVE_STATE_DIR", str(state_dir))
@pytest.fixture(autouse=True)
def _isolate_codex_native_state(
tmp_path_factory: pytest.TempPathFactory,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
Redirect codex-native client-side persistent state to a tmp dir.
The ``omnigent codex`` wrapper writes per-conversation launch
state under ``~/.omnigent/codex-native/<hash>/launch.json``.
Tests that drive the wrapper should never write to or read from
the developer's real persistent resume state.
The state module honors :data:`OMNIGENT_CODEX_NATIVE_STATE_DIR`
as a root override. ``autouse=True`` keeps test isolation as the
default even for indirect wrapper tests that do not explicitly
request a Codex state fixture.
:param tmp_path_factory: Pytest's session-scoped temp factory.
:param monkeypatch: Pytest monkeypatch fixture; auto-restores
the env var at teardown.
:returns: None.
"""
state_dir = tmp_path_factory.mktemp("codex-native-state")
monkeypatch.setenv("OMNIGENT_CODEX_NATIVE_STATE_DIR", str(state_dir))
@pytest.fixture(autouse=True)
def _no_configured_codex_app_server_command(
monkeypatch: pytest.MonkeyPatch,
) -> Callable[[], tuple[str, ...]]:
"""
Run Codex directly unless a test configures an app-server wrapper.
The wrapper is read from the user's ``~/.omnigent/config.yaml``, so without
this every native Codex session launch would depend on the machine.
:param monkeypatch: Fixture used to stub the config read.
:returns: The real config reader, for a test that exercises it.
"""
from omnigent.harnesses.codex_native import app_server
real = app_server._configured_app_server_command
monkeypatch.setattr(app_server, "_configured_app_server_command", lambda: ())
return real
@pytest.fixture()
def claude_managed_settings() -> None:
"""Opt back in to the host's real Claude Code managed-settings chain.
Request this alongside a test that must read the machine's actual
``managed-settings.json``; :func:`_isolate_claude_managed_settings` then
leaves the path list alone.
:returns: None.
"""
@pytest.fixture(autouse=True)
def _isolate_claude_managed_settings(
request: pytest.FixtureRequest,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Keep the host's Claude Code managed settings out of ambient detection.
``omnigent.onboarding.ambient`` reads Claude Code's enterprise managed
settings to detect a gateway the CLI authenticates itself. Those paths are
**absolute and machine-global** (``/Library/Application Support/ClaudeCode/…``,
``/etc/claude-code/…``), so unlike ``~/.claude`` they are NOT redirected by
a test's tmp ``$HOME`` — a developer's or CI runner's real enterprise install
would otherwise add a phantom Claude credential to every detection assertion.
``claude_native`` resolves this same tuple at call time, so neutralizing it
here covers its Smart-Routing gateway check and managed model overrides too.
``autouse=True`` because the alternative leaves us one missed test away from
host-dependent failures that reproduce only on configured machines. Tests
that exercise the detection pass an explicit ``paths=`` argument (the readers
take one) or request :func:`claude_managed_settings`.
:param request: Pytest request, inspected for the opt-in fixture.
:param monkeypatch: Pytest monkeypatch fixture; restores the tuple at
teardown.
:returns: None.
"""
if "claude_managed_settings" in request.fixturenames:
return
monkeypatch.setattr("omnigent.onboarding.ambient.CLAUDE_CODE_MANAGED_SETTINGS_PATHS", ())
@pytest.fixture()
def pi_host_mcp_servers() -> None:
"""Opt back in to adding the host's MCP inventory to managed Pi agent dirs.
:returns: None.
"""
@pytest.fixture(autouse=True)
def _isolate_pi_host_mcp_servers(
request: pytest.FixtureRequest,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Keep the machine's Claude, Codex, and Cursor MCP servers out of managed Pi dirs.
Every managed Pi launch reads the host MCP inventory from the real ``$HOME``,
so without this a developer's configured servers would change which
``mcp.json`` a test sees. Tests of that behavior request
:func:`pi_host_mcp_servers` and point the inventory at a temporary home.
:param request: Pytest request, inspected for the opt-in fixture.
:param monkeypatch: Pytest monkeypatch fixture; restores the reader at teardown.
:returns: None.
"""
if "pi_host_mcp_servers" in request.fixturenames:
return
monkeypatch.setattr("omnigent.inner.pi_mcp._host_mcp_servers", list)
@pytest.fixture()
def untracked_cache_start() -> None:
"""Opt back in to the real :meth:`GitFilesystemRegistry.start`.
Request this alongside the tests that exercise the optimization
worker itself; :func:`_stub_untracked_cache_start` then leaves the
method alone.
:returns: None.
"""
@pytest.fixture(autouse=True)
def _stub_untracked_cache_start(
request: pytest.FixtureRequest,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Keep the git untracked-cache worker out of unrelated tests.
:meth:`GitFilesystemRegistry.start` spawns a daemon thread that shells
out to git. It lands at an unpredictable moment, so a test that swaps
the process-global ``subprocess.run`` can record the worker's argv as
one of its own calls and fail on an assertion about a wholly unrelated
command. Stubbing the method by default removes the race for every
such test; the worker's own tests request ``untracked_cache_start``.
:param request: Pytest request, inspected for the opt-in fixture.
:param monkeypatch: Pytest monkeypatch fixture; restores the method
at teardown.
:returns: None.
"""
if "untracked_cache_start" in request.fixturenames:
return
monkeypatch.setattr(GitFilesystemRegistry, "start", lambda self: None)
@pytest.fixture(autouse=True)
def _isolate_origin_identity(
request: pytest.FixtureRequest,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Keep debug-log workspace attribution off the network and out of later tests.
Host startup and CLI server resolution probe a workspace-hosted server once
when nothing records its workspace id. Unstubbed, that probe would consume a
response a test scripted into a patched ``httpx.get`` (or hit the real host),
and the published env would carry into later tests. The probe's own tests
request ``origin_identity_probe``.
:param request: Pytest request, inspected for the opt-in fixture.
:param monkeypatch: Pytest monkeypatch fixture; restores the env and probe
at teardown.
:returns: None.
"""
from omnigent import debug_logging
monkeypatch.delenv(debug_logging.ORIGIN_WORKSPACE_ID_ENV_VAR, raising=False)
monkeypatch.setattr(debug_logging, "_probed_workspace_ids", {})
if "origin_identity_probe" in request.fixturenames:
return
monkeypatch.setattr(debug_logging, "_probe_workspace_id", lambda api_base: None)
@pytest.fixture(autouse=True)
def _isolate_databricks_org_pin(monkeypatch: pytest.MonkeyPatch) -> None:
"""Register the org pin with monkeypatch so direct os.environ writes are undone."""
from omnigent.cli_auth import PINNED_DATABRICKS_ORG_ENV_VAR
monkeypatch.setenv(PINNED_DATABRICKS_ORG_ENV_VAR, "")
monkeypatch.delenv(PINNED_DATABRICKS_ORG_ENV_VAR)
@pytest.fixture
def origin_identity_probe() -> None:
"""Opt in to the real workspace-id probe (see ``_isolate_origin_identity``)."""
@pytest.fixture(scope="session", autouse=True)
def cleanup_snapshot_failures(pytestconfig: pytest.Config) -> Generator[None, None, None]:
"""Give every xdist worker its own snapshot-failures directory.
The pytest-playwright-visual-snapshot plugin ships a session-scoped
autouse fixture of this same name that ``rmtree``s then ``mkdir``s a
single static path (``playwright_visual_snapshot_failures_path``) at
session start. That fixture runs in *every* pytest session — including
the non-visual unit shards — and under ``-n`` all workers target the one
path: the rmtree/mkdir sequence is non-atomic, so one worker's
``mkdir(exist_ok=True)`` re-raises ``FileExistsError`` when another
worker deletes the dir in the window between them, and that fixture error
cascades to every test on the worker. This override (a conftest fixture
shadows the plugin fixture of the same name for the whole ``tests/`` tree)
keys the leaf off ``PYTEST_XDIST_WORKER`` so no two workers ever touch the
same directory — the race is gone by construction, with no retries or
sleeps. The shared parent is only ever created (never deleted), so the
plugin's delete-then-create-the-same-dir window cannot recur.
Without xdist (the serial ``ui-snapshot.yml`` visual gate) the worker id
is unset and the base path is used unchanged, so the committed snapshot
layout and the CI artifact upload are unaffected.
"""
import shutil
from pytest_playwright_visual_snapshot.plugin import SnapshotPaths, _get_option
root_dir = Path(pytestconfig.rootdir) # type: ignore[arg-type]
SnapshotPaths.snapshots_path = Path(
_get_option(pytestconfig, "playwright_visual_snapshots_path", cast=str)
or (root_dir / "__snapshots__")
)
base_failures_path = Path(
_get_option(pytestconfig, "playwright_visual_snapshot_failures_path", cast=str)
or (root_dir / "snapshot_failures")
)
# Per-worker leaf under xdist; the base path itself when run serially
# (master process / no xdist), keeping non-xdist output identical.
worker = os.environ.get("PYTEST_XDIST_WORKER")
failures_path = base_failures_path / worker if worker else base_failures_path
SnapshotPaths.failures_path = failures_path
# Only this worker's own leaf is ever removed, so the rmtree/mkdir pair is
# uncontended; parents=True only creates the shared parent, never deletes it.
shutil.rmtree(failures_path, ignore_errors=True)
failures_path.mkdir(parents=True, exist_ok=True)
yield
@pytest.fixture(scope="session")
def _worker_db_uri() -> Generator[str, None, None]:
"""
Session-scoped database URI — one DB per xdist worker, migrated once.
When ``OMNIGENT_TEST_DB_URI`` is set, creates one database per worker
(``omnigent_test_w0``, ``omnigent_test_w1``, …), runs Alembic migrations
exactly once per worker session, then tears the database down at the end.
This avoids migrating hundreds of times — one migration run per worker
instead of one per test.
For SQLite nothing is created here; ``db_uri`` handles per-test files.
"""
import re
import sqlalchemy as _sa
base_uri = os.environ.get("OMNIGENT_TEST_DB_URI", "")
if not base_uri:
yield ""
return
worker = os.environ.get("PYTEST_XDIST_WORKER", "w0")
db_name = f"omnigent_test_{worker}"
uri = re.sub(r"/[^/]*(\?.*)?$", f"/{db_name}", base_uri)
root_engine = _sa.create_engine(base_uri, isolation_level="AUTOCOMMIT")
dialect = root_engine.dialect.name
with root_engine.connect() as conn:
if dialect == "mysql":
conn.execute(
_sa.text(
f"CREATE DATABASE IF NOT EXISTS `{db_name}` "
"CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
)
)
elif dialect == "cockroachdb":
conn.execute(_sa.text(f'DROP DATABASE IF EXISTS "{db_name}" CASCADE'))
conn.execute(_sa.text(f'CREATE DATABASE "{db_name}"'))
else:
conn.execute(_sa.text(f'DROP DATABASE IF EXISTS "{db_name}"'))
conn.execute(_sa.text(f'CREATE DATABASE "{db_name}"'))
root_engine.dispose()
engine = get_or_create_engine(uri)
yield uri
with _engine_lock:
_engine_cache.pop(uri, None)
engine.dispose()
root_engine2 = _sa.create_engine(base_uri, isolation_level="AUTOCOMMIT")
with root_engine2.connect() as conn:
if dialect == "mysql":
conn.execute(_sa.text(f"DROP DATABASE IF EXISTS `{db_name}`"))
elif dialect == "cockroachdb":
conn.execute(_sa.text(f'DROP DATABASE IF EXISTS "{db_name}" CASCADE'))
else:
conn.execute(_sa.text(f'DROP DATABASE IF EXISTS "{db_name}" WITH (FORCE)'))
root_engine2.dispose()
@pytest.fixture(scope="session")
def _sqlite_db_template() -> Path | None:
"""Suites may supply a migrated SQLite snapshot; the default runs migrations."""
return None
@pytest.fixture()
def db_uri(
tmp_path: Path, _worker_db_uri: str, _sqlite_db_template: Path | None
) -> Generator[str, None, None]:
"""
Per-test database URI.
* **SQLite** (default): fresh file per test, optionally copied from
a suite's migrated template. Engines and data remain fully isolated.
* **Postgres / MySQL / CockroachDB** (``OMNIGENT_TEST_DB_URI`` set):
reuses the session-scoped worker database and clears all non-alembic
tables between tests so each test starts clean without re-migrating.
"""
import sqlalchemy as _sa
if not _worker_db_uri:
# SQLite: per-test file.
db_path = tmp_path / "test.db"
if _sqlite_db_template is not None:
shutil.copyfile(_sqlite_db_template, db_path)
uri = f"sqlite:///{db_path}"
engine = get_or_create_engine(uri)
yield uri
with _engine_lock:
_engine_cache.pop(uri, None)
engine.dispose()
return
engine = get_or_create_engine(_worker_db_uri)
dialect = engine.dialect.name
tables = [t for t in _sa.inspect(engine).get_table_names() if t != "alembic_version"]
# No FK constraints exist (dropped in p1a2b3c4d5e6) so no need to toggle
# FOREIGN_KEY_CHECKS — one less round-trip per test on MySQL.
operation = "DELETE FROM" if dialect == "cockroachdb" else "TRUNCATE TABLE"
with engine.begin() as conn:
for table in tables:
q = f"`{table}`" if dialect == "mysql" else f'"{table}"'
# CockroachDB implements TRUNCATE as a schema change. DELETE keeps
# per-test cleanup transactional and avoids seconds of DDL work.
conn.execute(_sa.text(f"{operation} {q}"))
yield _worker_db_uri
@pytest.fixture()
def lowered_idle_thresholds(monkeypatch: pytest.MonkeyPatch) -> None:
"""
Lower the terminal-idle thresholds so watcher tests don't burn
ten real seconds per assertion.
Mirrors :class:`tests.inner.test_terminal.TestTerminalIdleNotifications.setUp`
from the legacy class-based suite. Defaults marker substrings
to empty so tests that don't exercise the marker track see
pure diff semantics regardless of the production list — tests
that DO exercise markers can override locally with another
``monkeypatch.setattr``.
Shared between ``tests/inner/test_terminal.py`` (threaded /
asyncio watcher mechanics) and
``tests/tools/builtins/test_sys_terminal.py`` (AP-side
``notify_when_idle`` end-to-end). Promoted to root conftest
rather than duplicated per file so the threshold values stay
in lockstep — a future tuning change touches one location.
:param monkeypatch: Pytest's monkeypatch fixture; auto-restores
the original constants at teardown.
"""
from omnigent.inner import terminal as terminal_module
monkeypatch.setattr(terminal_module, "_IDLE_THRESHOLD_SECONDS", 0.4)
monkeypatch.setattr(terminal_module, "_IDLE_POLL_INTERVAL_SECONDS", 0.1)
monkeypatch.setattr(terminal_module, "_IDLE_MARKER_SUBSTRINGS", [])
monkeypatch.setattr(terminal_module, "_IDLE_MARKER_THRESHOLD_SECONDS", 0.4)