fix(teams): suppress SDK import-time dotenv instead of clearing environ
Teknium review on #62947: os.environ.clear()/update around deferred loaders is unsafe under concurrency and misses teams_pipeline's direct adapter import. Defer microsoft_teams binding in the Teams adapter, no-op dotenv.load_dotenv while the SDK imports, keep api_server explicit disable, and add SDK-import + load_gateway_config canaries. Fixes #62935
This commit is contained in:
Jaret Bottoms
Teknium
parent
a98c8eeed1
commit
eec6d3efde
|
|
@ -29,36 +29,12 @@ Usage (gateway side):
|
|||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Awaitable, Callable, Iterator, Optional
|
||||
from typing import Any, Awaitable, Callable, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _isolated_os_environ() -> Iterator[None]:
|
||||
"""Snapshot and restore ``os.environ`` around a deferred platform import.
|
||||
|
||||
Some third-party SDKs (notably ``microsoft-teams-apps``) call
|
||||
``load_dotenv(find_dotenv(usecwd=True))`` at module import time. That
|
||||
walks up from the process cwd and mutates the process-global environ —
|
||||
leaking root-profile secrets into secondary profiles and bypassing
|
||||
Hermes's own dotenv / secret-scope rules (#62935).
|
||||
|
||||
Deferred loaders are import-and-register only; any environ mutation they
|
||||
cause is treated as an unwanted side effect and discarded.
|
||||
"""
|
||||
saved = dict(os.environ)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
os.environ.clear()
|
||||
os.environ.update(saved)
|
||||
|
||||
|
||||
|
||||
@dataclass
|
||||
class PlatformEntry:
|
||||
"""Metadata and factory for a single platform adapter."""
|
||||
|
|
@ -229,10 +205,7 @@ class PlatformRegistry:
|
|||
if loader is None:
|
||||
return
|
||||
try:
|
||||
# Isolate environ so import-time dotenv / env mutation cannot leak
|
||||
# across profiles or override explicit YAML disables (#62935).
|
||||
with _isolated_os_environ():
|
||||
loader()
|
||||
loader()
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Deferred load of platform '%s' failed: %s",
|
||||
|
|
|
|||
|
|
@ -27,7 +27,8 @@ import html
|
|||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, Optional
|
||||
from contextlib import contextmanager
|
||||
from typing import Any, Dict, Iterator, Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
# httpx is imported lazily — only the ``_write_summary_via_incoming_webhook``
|
||||
|
|
@ -47,46 +48,37 @@ except ImportError:
|
|||
AIOHTTP_AVAILABLE = False
|
||||
web = None # type: ignore[assignment]
|
||||
|
||||
try:
|
||||
from microsoft_teams.apps import App, ActivityContext
|
||||
from microsoft_teams.common.http.client import ClientOptions
|
||||
from microsoft_teams.api import MessageActivity, ConversationReference
|
||||
from microsoft_teams.api.activities.typing import TypingActivityInput
|
||||
from microsoft_teams.api.activities.invoke.adaptive_card import AdaptiveCardInvokeActivity
|
||||
from microsoft_teams.api.models.adaptive_card import (
|
||||
AdaptiveCardActionCardResponse,
|
||||
AdaptiveCardActionMessageResponse,
|
||||
)
|
||||
from microsoft_teams.api.models.invoke_response import InvokeResponse, AdaptiveCardInvokeResponse
|
||||
from microsoft_teams.apps.http.adapter import (
|
||||
HttpMethod,
|
||||
HttpRequest,
|
||||
HttpResponse,
|
||||
HttpRouteHandler,
|
||||
)
|
||||
from microsoft_teams.cards import AdaptiveCard, ExecuteAction, TextBlock
|
||||
# microsoft-teams-apps calls ``load_dotenv(find_dotenv(usecwd=True))`` at
|
||||
# ``microsoft_teams.apps.app`` import time. Importing it during plugin discovery
|
||||
# / ``TeamsSummaryWriter`` imports would pollute process ``os.environ`` from a
|
||||
# cwd-discovered ``.env`` (#62935). Detect presence via find_spec only; bind
|
||||
# symbols in ``check_teams_requirements()`` behind a dotenv no-op.
|
||||
import importlib.util
|
||||
import sys as _sys
|
||||
|
||||
TEAMS_SDK_AVAILABLE = True
|
||||
except ImportError:
|
||||
TEAMS_SDK_AVAILABLE = False
|
||||
ClientOptions = None # type: ignore[assignment,misc]
|
||||
App = None # type: ignore[assignment,misc]
|
||||
ActivityContext = None # type: ignore[assignment,misc]
|
||||
MessageActivity = None # type: ignore[assignment,misc]
|
||||
ConversationReference = None # type: ignore[assignment,misc]
|
||||
TypingActivityInput = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardInvokeActivity = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardActionCardResponse = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardActionMessageResponse = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardInvokeResponse = None # type: ignore[assignment,misc,union-attr]
|
||||
InvokeResponse = None # type: ignore[assignment,misc]
|
||||
HttpMethod = str # type: ignore[assignment,misc]
|
||||
HttpRequest = None # type: ignore[assignment,misc]
|
||||
HttpResponse = None # type: ignore[assignment,misc]
|
||||
HttpRouteHandler = None # type: ignore[assignment,misc]
|
||||
AdaptiveCard = None # type: ignore[assignment,misc]
|
||||
ExecuteAction = None # type: ignore[assignment,misc]
|
||||
TextBlock = None # type: ignore[assignment,misc]
|
||||
try:
|
||||
TEAMS_SDK_AVAILABLE = importlib.util.find_spec("microsoft_teams") is not None
|
||||
except ValueError:
|
||||
# Test stubs may inject a module without ``__spec__``.
|
||||
TEAMS_SDK_AVAILABLE = "microsoft_teams" in _sys.modules
|
||||
ClientOptions = None # type: ignore[assignment,misc]
|
||||
App = None # type: ignore[assignment,misc]
|
||||
ActivityContext = None # type: ignore[assignment,misc]
|
||||
MessageActivity = None # type: ignore[assignment,misc]
|
||||
ConversationReference = None # type: ignore[assignment,misc]
|
||||
TypingActivityInput = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardInvokeActivity = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardActionCardResponse = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardActionMessageResponse = None # type: ignore[assignment,misc]
|
||||
AdaptiveCardInvokeResponse = None # type: ignore[assignment,misc,union-attr]
|
||||
InvokeResponse = None # type: ignore[assignment,misc]
|
||||
HttpMethod = str # type: ignore[assignment,misc]
|
||||
HttpRequest = None # type: ignore[assignment,misc]
|
||||
HttpResponse = None # type: ignore[assignment,misc]
|
||||
HttpRouteHandler = None # type: ignore[assignment,misc]
|
||||
AdaptiveCard = None # type: ignore[assignment,misc]
|
||||
ExecuteAction = None # type: ignore[assignment,misc]
|
||||
TextBlock = None # type: ignore[assignment,misc]
|
||||
|
||||
from gateway.config import Platform, PlatformConfig
|
||||
from gateway.platforms.helpers import MessageDeduplicator
|
||||
|
|
@ -631,6 +623,31 @@ async def _standalone_send(
|
|||
# install. ``check_teams_requirements`` is the ACTIVE lazy-installer called
|
||||
# from ``connect()``; it installs ``platform.teams`` on demand and rebinds the
|
||||
# SDK globals, mirroring ``check_slack_requirements`` in gateway/platforms/slack.py.
|
||||
@contextmanager
|
||||
def _suppress_third_party_dotenv() -> Iterator[None]:
|
||||
"""No-op ``dotenv.load_dotenv`` while importing the Teams SDK (#62935).
|
||||
|
||||
``microsoft_teams.apps.app`` calls ``load_dotenv(find_dotenv(usecwd=True))``
|
||||
at module import time. That mutates process-global ``os.environ`` from
|
||||
whatever ``.env`` sits above cwd — typically a root profile's secrets.
|
||||
Hermes owns dotenv loading; third-party import side effects must not.
|
||||
"""
|
||||
try:
|
||||
import dotenv as _dotenv
|
||||
except ImportError:
|
||||
yield
|
||||
return
|
||||
original = getattr(_dotenv, "load_dotenv", None)
|
||||
if original is None:
|
||||
yield
|
||||
return
|
||||
_dotenv.load_dotenv = lambda *args, **kwargs: False # type: ignore[assignment]
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_dotenv.load_dotenv = original # type: ignore[assignment]
|
||||
|
||||
|
||||
def check_teams_requirements() -> bool:
|
||||
"""Ensure the Teams SDK is importable, lazy-installing it on first use.
|
||||
|
||||
|
|
@ -638,34 +655,39 @@ def check_teams_requirements() -> bool:
|
|||
``tools.lazy_deps.ensure("platform.teams")`` if not present, then rebinds
|
||||
all module-level SDK globals on success. Returns True once the SDK (and
|
||||
aiohttp) are importable, False if they couldn't be installed/imported.
|
||||
|
||||
``App is not None`` means symbols are already bound — ``TEAMS_SDK_AVAILABLE``
|
||||
alone can be True from ``find_spec`` without an import having run yet.
|
||||
"""
|
||||
if TEAMS_SDK_AVAILABLE and AIOHTTP_AVAILABLE:
|
||||
if App is not None and AIOHTTP_AVAILABLE:
|
||||
return True
|
||||
|
||||
def _import() -> dict:
|
||||
from aiohttp import web as _web
|
||||
from microsoft_teams.apps import App, ActivityContext
|
||||
from microsoft_teams.common.http.client import ClientOptions
|
||||
from microsoft_teams.api import MessageActivity, ConversationReference
|
||||
from microsoft_teams.api.activities.typing import TypingActivityInput
|
||||
from microsoft_teams.api.activities.invoke.adaptive_card import (
|
||||
AdaptiveCardInvokeActivity,
|
||||
)
|
||||
from microsoft_teams.api.models.adaptive_card import (
|
||||
AdaptiveCardActionCardResponse,
|
||||
AdaptiveCardActionMessageResponse,
|
||||
)
|
||||
from microsoft_teams.api.models.invoke_response import (
|
||||
InvokeResponse,
|
||||
AdaptiveCardInvokeResponse,
|
||||
)
|
||||
from microsoft_teams.apps.http.adapter import (
|
||||
HttpMethod,
|
||||
HttpRequest,
|
||||
HttpResponse,
|
||||
HttpRouteHandler,
|
||||
)
|
||||
from microsoft_teams.cards import AdaptiveCard, ExecuteAction, TextBlock
|
||||
|
||||
with _suppress_third_party_dotenv():
|
||||
from microsoft_teams.apps import App, ActivityContext
|
||||
from microsoft_teams.common.http.client import ClientOptions
|
||||
from microsoft_teams.api import MessageActivity, ConversationReference
|
||||
from microsoft_teams.api.activities.typing import TypingActivityInput
|
||||
from microsoft_teams.api.activities.invoke.adaptive_card import (
|
||||
AdaptiveCardInvokeActivity,
|
||||
)
|
||||
from microsoft_teams.api.models.adaptive_card import (
|
||||
AdaptiveCardActionCardResponse,
|
||||
AdaptiveCardActionMessageResponse,
|
||||
)
|
||||
from microsoft_teams.api.models.invoke_response import (
|
||||
InvokeResponse,
|
||||
AdaptiveCardInvokeResponse,
|
||||
)
|
||||
from microsoft_teams.apps.http.adapter import (
|
||||
HttpMethod,
|
||||
HttpRequest,
|
||||
HttpResponse,
|
||||
HttpRouteHandler,
|
||||
)
|
||||
from microsoft_teams.cards import AdaptiveCard, ExecuteAction, TextBlock
|
||||
|
||||
return {
|
||||
"web": _web,
|
||||
|
|
|
|||
|
|
@ -494,41 +494,3 @@ class TestPluginEnablementGate:
|
|||
)
|
||||
finally:
|
||||
_reg.unregister("myrejectedplat")
|
||||
|
||||
|
||||
|
||||
class TestDeferredLoadEnvironIsolation:
|
||||
"""Deferred platform imports must not leak os.environ mutations (#62935)."""
|
||||
|
||||
def test_deferred_loader_environ_mutations_are_discarded(self, monkeypatch):
|
||||
monkeypatch.delenv("API_SERVER_ENABLED", raising=False)
|
||||
monkeypatch.delenv("LEAKED_FROM_IMPORT", raising=False)
|
||||
assert "API_SERVER_ENABLED" not in os.environ
|
||||
|
||||
reg = PlatformRegistry()
|
||||
|
||||
def _polluting_loader():
|
||||
os.environ["API_SERVER_ENABLED"] = "true"
|
||||
os.environ["LEAKED_FROM_IMPORT"] = "1"
|
||||
entry, _ = TestPlatformRegistry()._make_entry("polluter")
|
||||
reg.register(entry)
|
||||
|
||||
reg.register_deferred("polluter", _polluting_loader)
|
||||
assert reg.get("polluter") is not None
|
||||
assert "API_SERVER_ENABLED" not in os.environ
|
||||
assert "LEAKED_FROM_IMPORT" not in os.environ
|
||||
|
||||
def test_deferred_loader_preserves_preexisting_environ(self, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_KEEP", "alive")
|
||||
reg = PlatformRegistry()
|
||||
|
||||
def _loader():
|
||||
os.environ["HERMES_KEEP"] = "clobbered"
|
||||
os.environ["TEMP_LEAK"] = "1"
|
||||
entry, _ = TestPlatformRegistry()._make_entry("keeper")
|
||||
reg.register(entry)
|
||||
|
||||
reg.register_deferred("keeper", _loader)
|
||||
assert reg.get("keeper") is not None
|
||||
assert os.environ.get("HERMES_KEEP") == "alive"
|
||||
assert "TEMP_LEAK" not in os.environ
|
||||
|
|
|
|||
|
|
@ -168,8 +168,10 @@ _ensure_teams_mock()
|
|||
# (plugin_adapter_teams) so it cannot collide with sibling plugin adapters.
|
||||
_teams_mod = load_plugin_adapter("teams")
|
||||
|
||||
_teams_mod.TEAMS_SDK_AVAILABLE = True
|
||||
_teams_mod.AIOHTTP_AVAILABLE = True
|
||||
# SDK import is deferred (#62935); bind mocked symbols the same way connect() does.
|
||||
assert _teams_mod.check_teams_requirements() is True
|
||||
_teams_mod.TEAMS_SDK_AVAILABLE = True
|
||||
|
||||
# Ensure SDK symbols that were None (import failed on Python <3.12) are
|
||||
# replaced with the mocked versions so runtime calls don't silently no-op.
|
||||
|
|
@ -207,9 +209,9 @@ class TestTeamsRequirements:
|
|||
assert check_requirements() is True
|
||||
|
||||
def test_check_teams_requirements_shortcircuits_when_present(self, monkeypatch):
|
||||
# When the SDK + aiohttp are already importable, the active lazy-
|
||||
# installer returns True immediately without attempting an install.
|
||||
monkeypatch.setattr(_teams_mod, "TEAMS_SDK_AVAILABLE", True)
|
||||
# When SDK symbols are already bound and aiohttp is available, the
|
||||
# active lazy-installer returns True immediately without re-importing.
|
||||
monkeypatch.setattr(_teams_mod, "App", object())
|
||||
monkeypatch.setattr(_teams_mod, "AIOHTTP_AVAILABLE", True)
|
||||
called = {"ensure_and_bind": 0}
|
||||
|
||||
|
|
@ -223,6 +225,29 @@ class TestTeamsRequirements:
|
|||
assert check_teams_requirements() is True
|
||||
assert called["ensure_and_bind"] == 0
|
||||
|
||||
def test_check_teams_requirements_lazy_installs_when_missing(self, monkeypatch):
|
||||
# When deps are missing, the active installer delegates to
|
||||
# ensure_and_bind("platform.teams", ...) — parity with Slack/Discord.
|
||||
monkeypatch.setattr(_teams_mod, "App", None)
|
||||
monkeypatch.setattr(_teams_mod, "TEAMS_SDK_AVAILABLE", False)
|
||||
monkeypatch.setattr(_teams_mod, "AIOHTTP_AVAILABLE", False)
|
||||
seen = {}
|
||||
|
||||
def _fake_ensure_and_bind(feature, importer, target_globals, **kwargs):
|
||||
seen["feature"] = feature
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tools.lazy_deps.ensure_and_bind", _fake_ensure_and_bind
|
||||
)
|
||||
assert check_teams_requirements() is True
|
||||
assert seen["feature"] == "platform.teams"
|
||||
|
||||
def test_validate_config_with_env(self, monkeypatch):
|
||||
monkeypatch.setenv("TEAMS_CLIENT_ID", "test-id")
|
||||
monkeypatch.setenv("TEAMS_CLIENT_SECRET", "test-secret")
|
||||
monkeypatch.setenv("TEAMS_TENANT_ID", "test-tenant")
|
||||
assert validate_config(_make_config()) is True
|
||||
|
||||
def test_validate_config_from_extra(self, monkeypatch):
|
||||
monkeypatch.delenv("TEAMS_CLIENT_ID", raising=False)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,223 @@
|
|||
"""Canaries for Teams SDK import-time dotenv isolation (#62935 / #62947)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
CANARY_KEY = "HERMES_TEAMS_DOTENV_CANARY"
|
||||
|
||||
|
||||
def _plant_cwd_dotenv(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Plant a cwd ``.env`` that would leak if Teams SDK dotenv ran unguarded."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / ".env").write_text(f"{CANARY_KEY}=leaked-from-cwd\n", encoding="utf-8")
|
||||
monkeypatch.delenv(CANARY_KEY, raising=False)
|
||||
|
||||
|
||||
def _install_fake_teams_sdk(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Minimal microsoft_teams package tree for adapter imports."""
|
||||
microsoft_teams = types.ModuleType("microsoft_teams")
|
||||
microsoft_teams.__path__ = [] # type: ignore[attr-defined]
|
||||
|
||||
apps = types.ModuleType("microsoft_teams.apps")
|
||||
apps.__path__ = [] # type: ignore[attr-defined]
|
||||
apps.App = type("App", (), {})
|
||||
apps.ActivityContext = type("ActivityContext", (), {})
|
||||
|
||||
common = types.ModuleType("microsoft_teams.common")
|
||||
common.__path__ = [] # type: ignore[attr-defined]
|
||||
common_http = types.ModuleType("microsoft_teams.common.http")
|
||||
common_http.__path__ = [] # type: ignore[attr-defined]
|
||||
common_http_client = types.ModuleType("microsoft_teams.common.http.client")
|
||||
common_http_client.ClientOptions = type("ClientOptions", (), {})
|
||||
|
||||
api = types.ModuleType("microsoft_teams.api")
|
||||
api.__path__ = [] # type: ignore[attr-defined]
|
||||
api.MessageActivity = type("MessageActivity", (), {})
|
||||
api.ConversationReference = type("ConversationReference", (), {})
|
||||
|
||||
api_activities = types.ModuleType("microsoft_teams.api.activities")
|
||||
api_activities.__path__ = [] # type: ignore[attr-defined]
|
||||
api_typing = types.ModuleType("microsoft_teams.api.activities.typing")
|
||||
api_typing.TypingActivityInput = type("TypingActivityInput", (), {})
|
||||
api_invoke = types.ModuleType("microsoft_teams.api.activities.invoke")
|
||||
api_invoke.__path__ = [] # type: ignore[attr-defined]
|
||||
api_invoke_card = types.ModuleType(
|
||||
"microsoft_teams.api.activities.invoke.adaptive_card"
|
||||
)
|
||||
api_invoke_card.AdaptiveCardInvokeActivity = type(
|
||||
"AdaptiveCardInvokeActivity", (), {}
|
||||
)
|
||||
|
||||
api_models = types.ModuleType("microsoft_teams.api.models")
|
||||
api_models.__path__ = [] # type: ignore[attr-defined]
|
||||
api_models_card = types.ModuleType("microsoft_teams.api.models.adaptive_card")
|
||||
api_models_card.AdaptiveCardActionCardResponse = type(
|
||||
"AdaptiveCardActionCardResponse", (), {}
|
||||
)
|
||||
api_models_card.AdaptiveCardActionMessageResponse = type(
|
||||
"AdaptiveCardActionMessageResponse", (), {}
|
||||
)
|
||||
api_models_invoke = types.ModuleType(
|
||||
"microsoft_teams.api.models.invoke_response"
|
||||
)
|
||||
api_models_invoke.InvokeResponse = type("InvokeResponse", (), {})
|
||||
api_models_invoke.AdaptiveCardInvokeResponse = type(
|
||||
"AdaptiveCardInvokeResponse", (), {}
|
||||
)
|
||||
|
||||
apps_http = types.ModuleType("microsoft_teams.apps.http")
|
||||
apps_http.__path__ = [] # type: ignore[attr-defined]
|
||||
apps_http_adapter = types.ModuleType("microsoft_teams.apps.http.adapter")
|
||||
apps_http_adapter.HttpMethod = str
|
||||
apps_http_adapter.HttpRequest = type("HttpRequest", (), {})
|
||||
apps_http_adapter.HttpResponse = type("HttpResponse", (), {})
|
||||
apps_http_adapter.HttpRouteHandler = type("HttpRouteHandler", (), {})
|
||||
|
||||
cards = types.ModuleType("microsoft_teams.cards")
|
||||
cards.AdaptiveCard = type("AdaptiveCard", (), {})
|
||||
cards.ExecuteAction = type("ExecuteAction", (), {})
|
||||
cards.TextBlock = type("TextBlock", (), {})
|
||||
|
||||
for name, mod in {
|
||||
"microsoft_teams": microsoft_teams,
|
||||
"microsoft_teams.apps": apps,
|
||||
"microsoft_teams.common": common,
|
||||
"microsoft_teams.common.http": common_http,
|
||||
"microsoft_teams.common.http.client": common_http_client,
|
||||
"microsoft_teams.api": api,
|
||||
"microsoft_teams.api.activities": api_activities,
|
||||
"microsoft_teams.api.activities.typing": api_typing,
|
||||
"microsoft_teams.api.activities.invoke": api_invoke,
|
||||
"microsoft_teams.api.activities.invoke.adaptive_card": api_invoke_card,
|
||||
"microsoft_teams.api.models": api_models,
|
||||
"microsoft_teams.api.models.adaptive_card": api_models_card,
|
||||
"microsoft_teams.api.models.invoke_response": api_models_invoke,
|
||||
"microsoft_teams.apps.http": apps_http,
|
||||
"microsoft_teams.apps.http.adapter": apps_http_adapter,
|
||||
"microsoft_teams.cards": cards,
|
||||
}.items():
|
||||
monkeypatch.setitem(sys.modules, name, mod)
|
||||
|
||||
|
||||
def _purge_teams_adapter_modules() -> None:
|
||||
for name in list(sys.modules):
|
||||
if name == "plugins.platforms.teams" or name.startswith(
|
||||
"plugins.platforms.teams."
|
||||
):
|
||||
del sys.modules[name]
|
||||
|
||||
|
||||
class TestTeamsAdapterImportDoesNotLeakDotenv:
|
||||
def test_adapter_import_does_not_load_cwd_dotenv(self, tmp_path, monkeypatch):
|
||||
_plant_cwd_dotenv(tmp_path, monkeypatch)
|
||||
_install_fake_teams_sdk(monkeypatch)
|
||||
_purge_teams_adapter_modules()
|
||||
|
||||
import plugins.platforms.teams.adapter as teams_adapter
|
||||
|
||||
assert CANARY_KEY not in os.environ
|
||||
assert teams_adapter.App is None # SDK symbols deferred
|
||||
|
||||
def test_teams_summary_writer_import_does_not_load_cwd_dotenv(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
"""teams_pipeline/runtime.py imports TeamsSummaryWriter directly."""
|
||||
_plant_cwd_dotenv(tmp_path, monkeypatch)
|
||||
_install_fake_teams_sdk(monkeypatch)
|
||||
_purge_teams_adapter_modules()
|
||||
|
||||
from plugins.platforms.teams.adapter import TeamsSummaryWriter
|
||||
|
||||
assert CANARY_KEY not in os.environ
|
||||
assert TeamsSummaryWriter is not None
|
||||
|
||||
def test_sdk_import_path_suppresses_dotenv(self, tmp_path, monkeypatch):
|
||||
"""Guarded SDK bind must no-op dotenv.load_dotenv (SDK import side effect)."""
|
||||
_plant_cwd_dotenv(tmp_path, monkeypatch)
|
||||
_install_fake_teams_sdk(monkeypatch)
|
||||
_purge_teams_adapter_modules()
|
||||
|
||||
import dotenv
|
||||
|
||||
import plugins.platforms.teams.adapter as teams_adapter
|
||||
|
||||
def _marking_load_dotenv(*args, **kwargs):
|
||||
os.environ[CANARY_KEY] = "leaked-from-sdk-import"
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(dotenv, "load_dotenv", _marking_load_dotenv)
|
||||
|
||||
# Unguarded call would leak — establish the canary contract.
|
||||
dotenv.load_dotenv()
|
||||
assert os.environ.get(CANARY_KEY) == "leaked-from-sdk-import"
|
||||
monkeypatch.delenv(CANARY_KEY, raising=False)
|
||||
|
||||
with teams_adapter._suppress_third_party_dotenv():
|
||||
dotenv.load_dotenv(dotenv.find_dotenv(usecwd=True))
|
||||
# Also exercise the real bind importer under the same suppress.
|
||||
from microsoft_teams.apps import App # noqa: F401
|
||||
|
||||
assert CANARY_KEY not in os.environ
|
||||
|
||||
# Active requirements path must also stay clean.
|
||||
monkeypatch.setattr(teams_adapter, "App", None)
|
||||
monkeypatch.setattr(teams_adapter, "AIOHTTP_AVAILABLE", True)
|
||||
|
||||
def _fake_ensure_and_bind(feature, importer, target_globals, **kwargs):
|
||||
assert feature == "platform.teams"
|
||||
# Call dotenv the way microsoft_teams.apps.app does, but from inside
|
||||
# the importer which wraps SDK imports in _suppress_third_party_dotenv.
|
||||
# We inject the dotenv call by wrapping the importer.
|
||||
original_importer = importer
|
||||
|
||||
def _importer_with_sdk_dotenv():
|
||||
with teams_adapter._suppress_third_party_dotenv():
|
||||
dotenv.load_dotenv(dotenv.find_dotenv(usecwd=True))
|
||||
return original_importer()
|
||||
|
||||
bound = _importer_with_sdk_dotenv()
|
||||
target_globals.update(bound)
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tools.lazy_deps.ensure_and_bind", _fake_ensure_and_bind
|
||||
)
|
||||
assert teams_adapter.check_teams_requirements() is True
|
||||
assert CANARY_KEY not in os.environ
|
||||
assert teams_adapter.App is not None
|
||||
|
||||
|
||||
class TestLoadGatewayConfigApiServerExplicitDisable:
|
||||
def test_load_gateway_config_honors_explicit_api_server_disable(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "config.yaml").write_text(
|
||||
"platforms:\n"
|
||||
" api_server:\n"
|
||||
" enabled: false\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setenv("API_SERVER_ENABLED", "true")
|
||||
# Must satisfy _has_usable_api_server_key (min_length=16) — a weaker
|
||||
# key never enters the env-override branch on current main, which
|
||||
# would vacuously pass the enabled=False assertion below.
|
||||
monkeypatch.setenv("API_SERVER_KEY", "test-key-0123456789abcdef")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
from gateway.config import Platform, load_gateway_config
|
||||
|
||||
config = load_gateway_config()
|
||||
api_cfg = config.platforms.get(Platform.API_SERVER)
|
||||
assert api_cfg is not None
|
||||
assert api_cfg.enabled is False
|
||||
assert api_cfg.extra.get("key") == "test-key-0123456789abcdef"
|
||||
Loading…
Reference in New Issue