e84a91cdbd
In-chat cronjob tool reached cron.jobs whose path constants snapshot at import, so a chat under a named profile hit the DEFAULT profile's cron dirs. Adds a per-turn contextvar (_STREAMING_CRON_PROFILE_HOME, reset on every exit) + a tool-call-boundary wrapper that enters cron_profile_context_for_home() only while the cronjob handler runs (cron lock, restored immediately — no whole-turn global mutation). Verified: agent executor's propagate_context_to_thread uses unfiltered contextvars.copy_context() so the contextvar reaches the sync handler on the ThreadPool worker; reset-token symmetry intact. Code byte-faithful from PR head c3fd6ab22f8c. 12 own tests pass. Co-authored-by: nesquena-hermes <agent@nesquena-hermes> Co-authored-by: roian6 <roian6@users.noreply.github.com>
552 lines
20 KiB
Python
552 lines
20 KiB
Python
"""Regression test: /api/crons must read jobs.json from the *active profile*.
|
|
|
|
Before the fix, `cron.jobs.list_jobs()` resolved HERMES_HOME from os.environ
|
|
at call time, ignoring the WebUI's per-request thread-local profile. So the
|
|
Scheduled Jobs panel showed the process-default profile's jobs regardless of
|
|
which profile the user had selected in the cookie.
|
|
|
|
This test writes two distinct jobs.json files (default + a named profile),
|
|
then verifies `cron_profile_context` pins the cron.jobs call to the named
|
|
profile's file.
|
|
"""
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
import threading
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
# Ensure both repos are importable.
|
|
WEBUI_ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
AGENT_ROOT = pathlib.Path(os.environ.get("HERMES_AGENT_ROOT", pathlib.Path.home() / "hermes-agent"))
|
|
for p in (str(WEBUI_ROOT), str(AGENT_ROOT)):
|
|
if p not in sys.path:
|
|
sys.path.insert(0, p)
|
|
|
|
|
|
def _write_jobs(home: pathlib.Path, jobs: list):
|
|
cron_dir = home / "cron"
|
|
cron_dir.mkdir(parents=True, exist_ok=True)
|
|
(cron_dir / "jobs.json").write_text(
|
|
json.dumps({"jobs": jobs}), encoding="utf-8"
|
|
)
|
|
|
|
|
|
def test_cron_profile_context_pins_profile_home(tmp_path, monkeypatch):
|
|
"""The context manager should swap cron.jobs to read from the named profile."""
|
|
pytest.importorskip("cron.jobs") # auto-skip when hermes-agent is unavailable
|
|
|
|
default_home = tmp_path / "default_home"
|
|
meow_home = tmp_path / "default_home" / "profiles" / "meow"
|
|
|
|
_write_jobs(default_home, [{"id": "d1", "name": "default-job"}])
|
|
_write_jobs(meow_home, [{"id": "m1", "name": "meow-job"}])
|
|
|
|
# Point base at default_home; HERMES_HOME env starts at default.
|
|
monkeypatch.setenv("HERMES_HOME", str(default_home))
|
|
|
|
from api import profiles as p
|
|
|
|
monkeypatch.setattr(p, "_DEFAULT_HERMES_HOME", default_home)
|
|
|
|
# Baseline: no context → default profile.
|
|
from cron.jobs import list_jobs
|
|
# Force cron.jobs to re-evaluate its cached constants for this test run.
|
|
import cron.jobs as _cj
|
|
_cj.HERMES_DIR = default_home
|
|
_cj.CRON_DIR = default_home / "cron"
|
|
_cj.JOBS_FILE = _cj.CRON_DIR / "jobs.json"
|
|
_cj.OUTPUT_DIR = _cj.CRON_DIR / "output"
|
|
|
|
jobs_before = list_jobs(include_disabled=True)
|
|
assert any(j["id"] == "d1" for j in jobs_before), \
|
|
f"Expected default-profile job before entering context, got {jobs_before}"
|
|
|
|
# Simulate a request with TLS profile = 'meow'.
|
|
p.set_request_profile("meow")
|
|
try:
|
|
with p.cron_profile_context():
|
|
jobs_inside = list_jobs(include_disabled=True)
|
|
assert any(j["id"] == "m1" for j in jobs_inside), \
|
|
f"Expected meow-profile job inside context, got {jobs_inside}"
|
|
assert not any(j["id"] == "d1" for j in jobs_inside), \
|
|
"Default-profile job leaked into meow context"
|
|
finally:
|
|
p.clear_request_profile()
|
|
|
|
# After the context exits, we should be back to default.
|
|
jobs_after = list_jobs(include_disabled=True)
|
|
assert any(j["id"] == "d1" for j in jobs_after), \
|
|
f"Expected default-profile job after exiting context, got {jobs_after}"
|
|
|
|
|
|
def test_cron_profile_context_for_home_pins_explicit_home(tmp_path):
|
|
"""Thread variant: pin by explicit path (no TLS)."""
|
|
pytest.importorskip("cron.jobs") # auto-skip when hermes-agent is unavailable
|
|
|
|
home_a = tmp_path / "a"
|
|
home_b = tmp_path / "b"
|
|
_write_jobs(home_a, [{"id": "a1", "name": "A"}])
|
|
_write_jobs(home_b, [{"id": "b1", "name": "B"}])
|
|
|
|
# Start with env pointing at A.
|
|
prev = os.environ.get("HERMES_HOME")
|
|
os.environ["HERMES_HOME"] = str(home_a)
|
|
try:
|
|
import cron.jobs as _cj
|
|
_cj.HERMES_DIR = home_a
|
|
_cj.CRON_DIR = home_a / "cron"
|
|
_cj.JOBS_FILE = _cj.CRON_DIR / "jobs.json"
|
|
_cj.OUTPUT_DIR = _cj.CRON_DIR / "output"
|
|
|
|
from cron.jobs import list_jobs
|
|
from api.profiles import cron_profile_context_for_home
|
|
|
|
assert any(j["id"] == "a1" for j in list_jobs(include_disabled=True))
|
|
|
|
with cron_profile_context_for_home(home_b):
|
|
jobs_inside = list_jobs(include_disabled=True)
|
|
assert any(j["id"] == "b1" for j in jobs_inside), jobs_inside
|
|
assert not any(j["id"] == "a1" for j in jobs_inside), jobs_inside
|
|
|
|
# Restored to A.
|
|
assert any(j["id"] == "a1" for j in list_jobs(include_disabled=True))
|
|
finally:
|
|
if prev is None:
|
|
os.environ.pop("HERMES_HOME", None)
|
|
else:
|
|
os.environ["HERMES_HOME"] = prev
|
|
|
|
|
|
def test_cron_profile_context_serializes_concurrent_access(tmp_path):
|
|
"""The lock must prevent concurrent contexts from interleaving."""
|
|
from api.profiles import cron_profile_context_for_home
|
|
|
|
home_a = tmp_path / "a"
|
|
home_b = tmp_path / "b"
|
|
home_a.mkdir()
|
|
home_b.mkdir()
|
|
|
|
# Ensure the context lock is released between tests.
|
|
from api import profiles as p
|
|
assert not p._cron_env_lock.locked(), \
|
|
"Lock leaked from a previous test"
|
|
|
|
observed = []
|
|
barrier = threading.Barrier(2)
|
|
|
|
def worker(home, tag):
|
|
barrier.wait()
|
|
with cron_profile_context_for_home(home):
|
|
observed.append(("enter", tag, os.environ["HERMES_HOME"]))
|
|
# If serialization works, the partner thread cannot be inside
|
|
# its own context at this moment.
|
|
observed.append(("exit", tag))
|
|
|
|
t1 = threading.Thread(target=worker, args=(home_a, "A"))
|
|
t2 = threading.Thread(target=worker, args=(home_b, "B"))
|
|
t1.start(); t2.start()
|
|
t1.join(); t2.join()
|
|
|
|
# Every enter must be immediately followed by its matching exit (no
|
|
# interleaving), because the lock serializes the two contexts.
|
|
assert len(observed) == 4
|
|
first, second, third, fourth = observed
|
|
assert first[0] == "enter" and second[0] == "exit" and first[1] == second[1]
|
|
assert third[0] == "enter" and fourth[0] == "exit" and third[1] == fourth[1]
|
|
|
|
|
|
def test_cron_run_does_not_silently_swallow_profile_resolution_errors():
|
|
"""_handle_cron_run must NOT silently fall through to profile_home=None
|
|
when get_active_hermes_home() raises.
|
|
|
|
A silent fallback would re-introduce the exact bug #1573 fixes — the
|
|
worker thread would run unpinned against the process-global HERMES_HOME,
|
|
silently corrupting cross-profile state. We'd rather 500 the request
|
|
than risk that, since get_active_hermes_home() raising at all from
|
|
inside a request handler means api.profiles is in a state we shouldn't
|
|
be making cron decisions in.
|
|
|
|
Source-level assertion to catch any future re-introduction of the
|
|
over-broad except clause.
|
|
"""
|
|
from pathlib import Path
|
|
src = (Path(__file__).resolve().parent.parent / "api" / "routes.py").read_text(encoding="utf-8")
|
|
|
|
# Locate _handle_cron_run definition; assert the spawn block does NOT
|
|
# wrap get_active_hermes_home() in a bare except that falls back to None.
|
|
idx = src.find("def _handle_cron_run(handler, body):")
|
|
assert idx != -1, "_handle_cron_run not found"
|
|
body = src[idx : idx + 4000]
|
|
|
|
# The spawn site must call get_active_hermes_home() unguarded (no
|
|
# try/except around it specifically), because a silent fallback to None
|
|
# is exactly what would re-introduce #1573.
|
|
spawn_idx = body.find("threading.Thread(target=_run_cron_tracked")
|
|
assert spawn_idx != -1, "thread spawn not found in _handle_cron_run"
|
|
|
|
# Look at the 1500 chars before the spawn — should NOT contain the
|
|
# `_profile_home = None` fallback pattern.
|
|
pre_spawn = body[max(0, spawn_idx - 1500) : spawn_idx]
|
|
assert "_profile_home = None" not in pre_spawn, (
|
|
"_handle_cron_run silently falls back to _profile_home=None when "
|
|
"get_active_hermes_home() raises. That re-introduces bug #1573 — "
|
|
"the worker thread would run unpinned against the process-global "
|
|
"HERMES_HOME. Let the exception propagate (500 the request) rather "
|
|
"than corrupt cross-profile state silently."
|
|
)
|
|
|
|
|
|
def test_manual_cron_event_profile_uses_job_profile(monkeypatch):
|
|
from api import routes
|
|
|
|
monkeypatch.setattr(routes, "_available_cron_profile_names", lambda: {"default", "research"})
|
|
|
|
assert routes._event_profile_for_cron_job({"profile": "research"}) == "research"
|
|
assert routes._event_profile_for_cron_job({"profile": " default "}) == "default"
|
|
assert routes._event_profile_for_cron_job({"profile": ""}) is None
|
|
assert routes._event_profile_for_cron_job({"profile": "deleted"}) is None
|
|
|
|
|
|
def test_webui_installs_profile_context_on_in_process_scheduler_run_job(tmp_path, monkeypatch):
|
|
"""If WebUI ever runs cron.scheduler.tick in-process, scheduled run_job calls
|
|
must execute under the job's selected profile home, not the process-global
|
|
HERMES_HOME that happened to be active when the scheduler thread fired.
|
|
"""
|
|
import types
|
|
|
|
from api import profiles as p
|
|
|
|
default_home = tmp_path / "home"
|
|
research_home = default_home / "profiles" / "research"
|
|
research_home.mkdir(parents=True)
|
|
events = []
|
|
|
|
class Ctx:
|
|
def __init__(self, home):
|
|
self.home = str(home)
|
|
|
|
def __enter__(self):
|
|
events.append(("enter", self.home))
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
events.append(("exit", self.home))
|
|
return False
|
|
|
|
cron_pkg = types.ModuleType("cron")
|
|
cron_pkg.__path__ = []
|
|
cron_scheduler = types.ModuleType("cron.scheduler")
|
|
cron_scheduler.run_job = lambda job: events.append(("run", job["id"])) or "ok"
|
|
|
|
monkeypatch.setitem(sys.modules, "cron", cron_pkg)
|
|
monkeypatch.setitem(sys.modules, "cron.scheduler", cron_scheduler)
|
|
monkeypatch.setattr(p, "_DEFAULT_HERMES_HOME", default_home)
|
|
monkeypatch.setattr(p, "cron_profile_context_for_home", Ctx)
|
|
monkeypatch.setattr(p, "publish_session_list_changed", lambda reason: events.append(("publish", reason)))
|
|
|
|
p.install_cron_scheduler_profile_isolation()
|
|
|
|
assert cron_scheduler.run_job({"id": "job1575", "profile": "research"}) == "ok"
|
|
assert events == [
|
|
("enter", str(research_home)),
|
|
("run", "job1575"),
|
|
("exit", str(research_home)),
|
|
("publish", "cron_complete"),
|
|
]
|
|
|
|
|
|
def test_scheduler_run_job_wrapper_does_not_reenter_manual_cron_context(tmp_path, monkeypatch):
|
|
"""Manual /api/crons/run already pins run_job before calling it.
|
|
|
|
The scheduler safety wrapper must detect that existing context and delegate
|
|
directly, otherwise the non-reentrant env lock would deadlock or override the
|
|
manual execution profile.
|
|
"""
|
|
import types
|
|
|
|
from api import profiles as p
|
|
|
|
events = []
|
|
|
|
class Ctx:
|
|
def __init__(self, home):
|
|
self.home = str(home)
|
|
|
|
def __enter__(self):
|
|
events.append(("unexpected-enter", self.home))
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
events.append(("unexpected-exit", self.home))
|
|
return False
|
|
|
|
cron_pkg = types.ModuleType("cron")
|
|
cron_pkg.__path__ = []
|
|
cron_scheduler = types.ModuleType("cron.scheduler")
|
|
cron_scheduler.run_job = lambda job: events.append(("run", job["id"])) or "ok"
|
|
|
|
monkeypatch.setitem(sys.modules, "cron", cron_pkg)
|
|
monkeypatch.setitem(sys.modules, "cron.scheduler", cron_scheduler)
|
|
monkeypatch.setattr(p, "_DEFAULT_HERMES_HOME", tmp_path / "home")
|
|
monkeypatch.setattr(p, "cron_profile_context_for_home", Ctx)
|
|
monkeypatch.setattr(p, "publish_session_list_changed", lambda reason: events.append(("unexpected-publish", reason)))
|
|
monkeypatch.setattr(p._tls, "cron_profile_depth", 1, raising=False)
|
|
|
|
p.install_cron_scheduler_profile_isolation()
|
|
|
|
assert cron_scheduler.run_job({"id": "manual1575", "profile": "research"}) == "ok"
|
|
assert events == [("run", "manual1575")]
|
|
|
|
|
|
def test_cron_worker_does_not_silently_fall_back_on_profile_context_failure():
|
|
"""The subprocess target must not fall back to an unpinned cron run.
|
|
|
|
A silent fallback would leave the job running against process-global
|
|
HERMES_HOME, silently corrupting cross-profile state — the same class of bug
|
|
as #1573. The child process may report the exception to the parent, but it
|
|
must not continue into run_job outside the requested profile context.
|
|
"""
|
|
from pathlib import Path
|
|
src = (Path(__file__).resolve().parent.parent / "api" / "routes.py").read_text(encoding="utf-8")
|
|
|
|
idx = src.find("def _cron_job_subprocess_main(job")
|
|
assert idx != -1, "_cron_job_subprocess_main not found"
|
|
body = src[idx : idx + 2000]
|
|
|
|
assert "with cron_profile_context_for_home(execution_profile_home):" in body
|
|
assert "result = _run()" in body
|
|
assert "ctx = None" not in body
|
|
assert "except Exception" not in body[:body.find("with cron_profile_context_for_home")], (
|
|
"cron subprocess target appears to catch profile-context setup before "
|
|
"entering the context; do not fall back to an unpinned run_job call."
|
|
)
|
|
|
|
|
|
def test_streaming_cronjob_wrapper_uses_profile_context_only_for_tool_call(tmp_path, monkeypatch):
|
|
"""The chat cronjob fix must use the cron profile context at call time.
|
|
|
|
Holding cron.jobs module globals for an entire streaming run would race with
|
|
other profiles. The wrapper should instead enter the existing locked cron
|
|
context only while the cronjob tool handler itself executes.
|
|
"""
|
|
import types
|
|
|
|
from api import profiles as p
|
|
from api import streaming as st
|
|
|
|
profile_home = tmp_path / "home" / "profiles" / "ops"
|
|
events = []
|
|
|
|
class Ctx:
|
|
def __init__(self, home):
|
|
self.home = str(home)
|
|
|
|
def __enter__(self):
|
|
events.append(("enter", self.home))
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
events.append(("exit", self.home))
|
|
return False
|
|
|
|
def original_handler(args, **kwargs):
|
|
events.append(("handler", args.get("action"), st._STREAMING_CRON_PROFILE_HOME.get()))
|
|
return "ok"
|
|
|
|
class Entry:
|
|
name = "cronjob"
|
|
toolset = "cronjob"
|
|
schema = {"name": "cronjob"}
|
|
handler = staticmethod(original_handler)
|
|
check_fn = None
|
|
requires_env = []
|
|
is_async = False
|
|
description = ""
|
|
emoji = "⏰"
|
|
max_result_size_chars = None
|
|
dynamic_schema_overrides = None
|
|
|
|
entry = Entry()
|
|
|
|
class Registry:
|
|
def get_entry(self, name):
|
|
assert name == "cronjob"
|
|
return entry
|
|
|
|
def register(self, **kwargs):
|
|
events.append(("register", kwargs["name"]))
|
|
entry.handler = kwargs["handler"]
|
|
|
|
tools_pkg = types.ModuleType("tools")
|
|
registry_mod = types.ModuleType("tools.registry")
|
|
registry_mod.__dict__["registry"] = Registry()
|
|
monkeypatch.setitem(sys.modules, "tools", tools_pkg)
|
|
monkeypatch.setitem(sys.modules, "tools.registry", registry_mod)
|
|
monkeypatch.setattr(st, "_STREAMING_CRONJOB_WRAPPER_INSTALLED", False)
|
|
monkeypatch.setattr(p, "cron_profile_context_for_home", Ctx)
|
|
|
|
st._install_streaming_cronjob_profile_wrapper()
|
|
assert events == [("register", "cronjob")]
|
|
|
|
token = st._STREAMING_CRON_PROFILE_HOME.set(str(profile_home))
|
|
try:
|
|
assert entry.handler({"action": "list"}, task_id="t1") == "ok"
|
|
finally:
|
|
st._STREAMING_CRON_PROFILE_HOME.reset(token)
|
|
|
|
assert events == [
|
|
("register", "cronjob"),
|
|
("enter", str(profile_home)),
|
|
("handler", "list", str(profile_home)),
|
|
("exit", str(profile_home)),
|
|
]
|
|
|
|
|
|
def test_streaming_cronjob_wrapper_context_survives_threadpool_context_copy(tmp_path, monkeypatch):
|
|
"""Lock the cross-thread contextvar contract used by agent tool dispatch.
|
|
|
|
WebUI sets the active profile on the streaming thread, then the Hermes agent
|
|
dispatches sync tools on a ThreadPoolExecutor worker under a copied
|
|
contextvars context. The cronjob wrapper must still see the profile context
|
|
on that worker or it silently falls back to the default profile.
|
|
"""
|
|
import concurrent.futures
|
|
import contextvars
|
|
import types
|
|
|
|
from api import profiles as p
|
|
from api import streaming as st
|
|
|
|
profile_home = tmp_path / "home" / "profiles" / "ops"
|
|
events = []
|
|
|
|
class Ctx:
|
|
def __init__(self, home):
|
|
self.home = str(home)
|
|
|
|
def __enter__(self):
|
|
events.append(("enter", self.home))
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
events.append(("exit", self.home))
|
|
return False
|
|
|
|
def original_handler(args, **kwargs):
|
|
events.append(("handler", args.get("action"), st._STREAMING_CRON_PROFILE_HOME.get()))
|
|
return "ok"
|
|
|
|
class Entry:
|
|
name = "cronjob"
|
|
toolset = "cronjob"
|
|
schema = {"name": "cronjob"}
|
|
handler = staticmethod(original_handler)
|
|
check_fn = None
|
|
requires_env = []
|
|
is_async = False
|
|
description = ""
|
|
emoji = "⏰"
|
|
max_result_size_chars = None
|
|
dynamic_schema_overrides = None
|
|
|
|
entry = Entry()
|
|
|
|
class Registry:
|
|
def get_entry(self, name):
|
|
assert name == "cronjob"
|
|
return entry
|
|
|
|
def register(self, **kwargs):
|
|
events.append(("register", kwargs["name"]))
|
|
entry.handler = kwargs["handler"]
|
|
|
|
registry_mod = types.ModuleType("tools.registry")
|
|
registry_mod.__dict__["registry"] = Registry()
|
|
monkeypatch.setitem(sys.modules, "tools.registry", registry_mod)
|
|
monkeypatch.setattr(st, "_STREAMING_CRONJOB_WRAPPER_INSTALLED", False)
|
|
monkeypatch.setattr(p, "cron_profile_context_for_home", Ctx)
|
|
|
|
st._install_streaming_cronjob_profile_wrapper()
|
|
token = st._STREAMING_CRON_PROFILE_HOME.set(str(profile_home))
|
|
try:
|
|
copied_context = contextvars.copy_context()
|
|
finally:
|
|
st._STREAMING_CRON_PROFILE_HOME.reset(token)
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
|
|
future = executor.submit(copied_context.run, entry.handler, {"action": "list"})
|
|
assert future.result(timeout=5) == "ok"
|
|
|
|
assert events == [
|
|
("register", "cronjob"),
|
|
("enter", str(profile_home)),
|
|
("handler", "list", str(profile_home)),
|
|
("exit", str(profile_home)),
|
|
]
|
|
|
|
|
|
def test_streaming_cronjob_wrapper_leaves_calls_unwrapped_without_streaming_profile(monkeypatch):
|
|
"""CLI/default calls through the registered cronjob handler are unchanged."""
|
|
import types
|
|
|
|
from api import streaming as st
|
|
|
|
events = []
|
|
|
|
def original_handler(args, **kwargs):
|
|
events.append(("handler", args.get("action")))
|
|
return "ok"
|
|
|
|
class Entry:
|
|
name = "cronjob"
|
|
toolset = "cronjob"
|
|
schema = {"name": "cronjob"}
|
|
handler = staticmethod(original_handler)
|
|
check_fn = None
|
|
requires_env = []
|
|
is_async = False
|
|
description = ""
|
|
emoji = "⏰"
|
|
max_result_size_chars = None
|
|
dynamic_schema_overrides = None
|
|
|
|
entry = Entry()
|
|
|
|
class Registry:
|
|
def get_entry(self, name):
|
|
assert name == "cronjob"
|
|
return entry
|
|
|
|
def register(self, **kwargs):
|
|
entry.handler = kwargs["handler"]
|
|
|
|
registry_mod = types.ModuleType("tools.registry")
|
|
registry_mod.__dict__["registry"] = Registry()
|
|
monkeypatch.setitem(sys.modules, "tools.registry", registry_mod)
|
|
monkeypatch.setattr(st, "_STREAMING_CRONJOB_WRAPPER_INSTALLED", False)
|
|
|
|
st._install_streaming_cronjob_profile_wrapper()
|
|
|
|
assert entry.handler({"action": "list"}) == "ok"
|
|
assert events == [("handler", "list")]
|
|
|
|
|
|
def test_streaming_profile_home_mutation_avoids_long_lived_cron_cache_patch():
|
|
"""Guard the streaming integration seam for issue #4580.
|
|
|
|
Streaming still mutates process env briefly for legacy fallbacks, but cron
|
|
path caches must be scoped to the cronjob tool-call boundary via the
|
|
wrapper/contextvar path — not patched for the full agent turn.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
src = (Path(__file__).resolve().parent.parent / "api" / "streaming.py").read_text(encoding="utf-8")
|
|
assert "_install_streaming_cronjob_profile_wrapper()" in src
|
|
assert "_STREAMING_CRON_PROFILE_HOME.set(_profile_home)" in src
|
|
assert "_STREAMING_CRON_PROFILE_HOME.reset(_streaming_cron_profile_home_token)" in src
|
|
assert "def _patch_streaming_profile_module_caches" not in src
|
|
assert "old_profile_module_cache_snapshot" not in src
|