forked from cleveragents/cleveragents-core
63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
"""Shared utilities for Robot Framework helper scripts.
|
|
|
|
This module centralises the ``reset_global_state()`` function used by
|
|
``helper_m4_e2e_verification.py``,
|
|
``helper_m4_correction_subplan_smoke.py``, and
|
|
``helper_m3_decision_validation_smoke.py``.
|
|
|
|
Any helper script invoked via ``Run Process`` from a ``.robot`` suite
|
|
can import this module to reset process-wide singletons between
|
|
chained CLI invocations.
|
|
"""
|
|
|
|
import contextlib
|
|
import sys
|
|
|
|
|
|
def reset_global_state() -> None:
|
|
"""Reset process-wide singletons between CLI invocations.
|
|
|
|
Clears the Settings singleton, DI container, provider registry,
|
|
and in-memory SQLAlchemy engine cache so that chained CLI
|
|
invocations within the same helper process do not carry stale
|
|
state.
|
|
|
|
Each import is guarded by ``contextlib.suppress(ImportError)``
|
|
because this module lives in ``robot/`` and may be loaded in
|
|
contexts where not all application modules are on the path
|
|
(e.g. a minimal helper that only exercises domain models).
|
|
This is an intentional exception to the top-of-file import rule
|
|
per CONTRIBUTING.md §Import Guidelines.
|
|
"""
|
|
# Settings singleton
|
|
with contextlib.suppress(ImportError):
|
|
from cleveragents.config.settings import Settings
|
|
|
|
Settings._instance = None
|
|
|
|
# DI container singleton
|
|
with contextlib.suppress(ImportError):
|
|
from cleveragents.application.container import reset_container
|
|
|
|
reset_container()
|
|
|
|
# Provider registry singleton
|
|
with contextlib.suppress(ImportError):
|
|
from cleveragents.providers.registry import reset_provider_registry
|
|
|
|
reset_provider_registry()
|
|
|
|
# In-memory SQLAlchemy engine cache
|
|
with contextlib.suppress(ImportError):
|
|
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
|
|
|
|
for _url, engine in list(MEMORY_ENGINES.items()):
|
|
try:
|
|
engine.dispose()
|
|
except Exception as exc:
|
|
print(
|
|
f"[helpers_common] engine.dispose() suppressed: {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
MEMORY_ENGINES.clear()
|