wehub-resource-sync
512 lines
20 KiB
Python
512 lines
20 KiB
Python
"""Shared pytest configuration and fixtures for Omnigent tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
from collections.abc import Generator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
try:
|
|
import resource as _resource # POSIX-only; absent on Windows.
|
|
except ImportError:
|
|
_resource = None # type: ignore[assignment]
|
|
|
|
# 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; the
|
|
# pre-rename OMNIGENT_ACCOUNTS_ENABLED is still honored too) — 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
|
|
from tests import _model_pools
|
|
|
|
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."""
|
|
|
|
|
|
# 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 _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(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"
|
|
)
|
|
)
|
|
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}`"))
|
|
else:
|
|
conn.execute(_sa.text(f'DROP DATABASE IF EXISTS "{db_name}" WITH (FORCE)'))
|
|
root_engine2.dispose()
|
|
|
|
|
|
@pytest.fixture()
|
|
def db_uri(tmp_path: Path, _worker_db_uri: str) -> Generator[str, None, None]:
|
|
"""
|
|
Per-test database URI.
|
|
|
|
* **SQLite** (default): fresh file per test, fully isolated.
|
|
* **Postgres / MySQL** (``OMNIGENT_TEST_DB_URI`` set): reuses the
|
|
session-scoped worker database and truncates 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"
|
|
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.
|
|
with engine.begin() as conn:
|
|
for table in tables:
|
|
q = f"`{table}`" if dialect == "mysql" else f'"{table}"'
|
|
conn.execute(_sa.text(f"TRUNCATE TABLE {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)
|