Skip to content
This repository was archived by the owner on May 26, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions tests/kora_cli/handlers/test_email_inbound_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,37 @@ def _isolate(tmp_path, monkeypatch):
return tmp_path


@pytest.fixture(autouse=True)
def _reset_operational_state_holder():
"""Self-protective reset of the ``OperationalStateHolder``
singleton — KR-TEST-STABILITY-XDIST.

The handler's state-gate (``_check_state_gate`` in
``email_inbound_handler.py``) calls ``get_holder()`` to decide
whether to drop the message as ``filtered_paused`` /
``filtered_stopped``. State-gate tests in this file (lines
151-200) ``patch`` the accessor; the other ~30 tests expect
the singleton to be ``None`` so ``get_holder()`` returns
``None`` and the handler proceeds.

Under pytest-xdist, OTHER test files in the same worker may
install a non-None holder (e.g.
``test_mcp_tools_stop_control.py`` from KR-MCP-STOP-CONTROL ST1
installed PAUSED holders without teardown reset before this
bucket landed). A bleed surfaces as flaky
``HANDLED_RECEIVED vs filtered_paused`` assertions.

This fixture is BELT-AND-SUSPENDERS: the leaking test files
now also reset their holders, but a future test author who
forgets the teardown won't poison email-handler tests.
"""
from agent import operational_state_holder as h_mod

h_mod._HOLDER = None
yield
h_mod._HOLDER = None


def _log_path(tmp_path: Path) -> Path:
return tmp_path / "email_inbound_log.jsonl"

Expand Down
20 changes: 20 additions & 0 deletions tests/kora_cli/test_listeners/test_mcp_audit_on_denial.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,26 @@ def _reset_caller_cache():
mcp_caller_auth._reset_cache_for_tests()


@pytest.fixture(autouse=True)
def _reset_operational_state_holder():
"""Reset the module-level ``OperationalStateHolder`` singleton
between tests — KR-TEST-STABILITY-XDIST.

``test_successful_call_does_not_emit_denial_audit`` installs an
ACTIVE holder via direct ``h_mod._HOLDER = ...`` so the pause
executor can transition. Without this autouse reset, that
holder persists across worker boundaries to subsequent tests in
the same xdist worker — surfacing as ``test_email_inbound_
handler.py`` flakes where the state-gate sees an unexpected
holder.
"""
from agent import operational_state_holder as h_mod

h_mod._HOLDER = None
yield
h_mod._HOLDER = None


@pytest.fixture
def empty_caps_token(monkeypatch, tmp_path):
"""Caller authenticated but with NO caps — every mutating tool denies."""
Expand Down
26 changes: 26 additions & 0 deletions tests/kora_cli/test_listeners/test_mcp_tools_stop_control.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,32 @@ def _reset_caller_cache():
mcp_caller_auth._reset_cache_for_tests()


@pytest.fixture(autouse=True)
def _reset_operational_state_holder():
"""Reset the module-level ``OperationalStateHolder`` singleton
between tests — KR-TEST-STABILITY-XDIST.

The ``_holder_with_state`` helper below installs a holder via
direct ``h_mod._HOLDER = holder`` assignment (rather than
``monkeypatch.setattr``) because most tests use the helper for
its side effect WITHOUT taking ``monkeypatch`` as a fixture arg.
Without this autouse reset, a test that sets the holder to
PAUSED leaks into the next test on the same xdist worker —
surfacing as ``test_email_inbound_handler.py`` flakes where the
state-gate sees a stale PAUSED holder and returns
``filtered_paused`` instead of ``received``.

Resetting at BOTH setup and teardown is intentional: a previous
test in the same worker may have left a dirty holder, AND this
test may dirty the holder. Either path catches the leak.
"""
from agent import operational_state_holder as h_mod

h_mod._HOLDER = None
yield
h_mod._HOLDER = None


@pytest.fixture
def authorized_token(monkeypatch, tmp_path):
"""Caller with BOTH pause + resume caps (no full transition cap)."""
Expand Down