46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
|
"""Pytest config: temp-dir isolation for state modules.
|
||
|
|
|
||
|
|
Each test gets a fresh dir for memory.json / schedule.json / etc so concurrent
|
||
|
|
runs don't pollute the real user data.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import importlib
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
# Make parent package importable.
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def isolated_state(tmp_path, monkeypatch):
|
||
|
|
"""Patch state modules to use a tmp_path so we don't touch real state files."""
|
||
|
|
|
||
|
|
def patch_module(mod_name, path_attrs):
|
||
|
|
"""Reimport a module and rewire its `_PATH` and friends to tmp_path."""
|
||
|
|
if mod_name in sys.modules:
|
||
|
|
del sys.modules[mod_name]
|
||
|
|
mod = importlib.import_module(mod_name)
|
||
|
|
for attr in path_attrs:
|
||
|
|
if not hasattr(mod, attr):
|
||
|
|
continue
|
||
|
|
original = getattr(mod, attr)
|
||
|
|
new_val = os.path.join(str(tmp_path), os.path.basename(original))
|
||
|
|
monkeypatch.setattr(mod, attr, new_val)
|
||
|
|
# Also reset any cached _loaded / _initialized flags
|
||
|
|
for flag in ('_loaded', '_initialized', '_thread_started'):
|
||
|
|
if hasattr(mod, flag):
|
||
|
|
monkeypatch.setattr(mod, flag, False)
|
||
|
|
# Reset stores
|
||
|
|
for store_attr in ('_store', '_tasks'):
|
||
|
|
if hasattr(mod, store_attr):
|
||
|
|
cur = getattr(mod, store_attr)
|
||
|
|
if isinstance(cur, dict):
|
||
|
|
monkeypatch.setattr(mod, store_attr, {})
|
||
|
|
elif isinstance(cur, list):
|
||
|
|
monkeypatch.setattr(mod, store_attr, [])
|
||
|
|
return mod
|
||
|
|
|
||
|
|
yield patch_module
|