fix(ci): eliminate test_set_key flakiness — v0.50.161

Root cause: test_profile_env_isolation.py and test_profile_path_security.py called sys.modules.pop() without restoring, poisoning subsequent tests. Fix: monkeypatch.delitem so pytest auto-restores. Also holds _ENV_LOCK for full I/O cycle in _write_env_file and creates .env at 0600 via os.open. Reviewed by Opus (no independent review needed — test/providers fix only).
This commit is contained in:
nesquena-hermes
2026-04-22 19:09:37 -07:00
committed by GitHub
parent cc025aab79
commit 0f1b232c12
5 changed files with 63 additions and 26 deletions

View File

@@ -85,33 +85,42 @@ def _write_env_file(env_path: Path, updates: dict[str, str | None]) -> None:
"""Write key=value pairs to the .env file.
Values of ``None`` cause the key to be removed.
Uses ``_ENV_LOCK`` from ``api.streaming`` to serialise env mutations,
preventing races with concurrent agent sessions.
Holds ``_ENV_LOCK`` from ``api.streaming`` for the entire load → modify →
write cycle to prevent TOCTOU races between concurrent POST /api/providers
calls (each reading the same file baseline and overwriting the other's key).
Also serialises os.environ mutations with streaming sessions.
"""
from api.streaming import _ENV_LOCK
import stat as _stat
current = _load_env_file(env_path)
for key, value in updates.items():
if value is None:
current.pop(key, None)
with _ENV_LOCK:
with _ENV_LOCK:
current = _load_env_file(env_path)
for key, value in updates.items():
if value is None:
current.pop(key, None)
os.environ.pop(key, None)
continue
clean = str(value).strip()
if not clean:
continue
# Reject embedded newlines/carriage returns to prevent .env injection
if "\n" in clean or "\r" in clean:
raise ValueError("API key must not contain newline characters.")
current[key] = clean
with _ENV_LOCK:
continue
clean = str(value).strip()
if not clean:
continue
# Reject embedded newlines/carriage returns to prevent .env injection
if "\n" in clean or "\r" in clean:
raise ValueError("API key must not contain newline characters.")
current[key] = clean
os.environ[key] = clean
env_path.parent.mkdir(parents=True, exist_ok=True)
lines = [f"{key}={current[key]}" for key in sorted(current)]
env_path.write_text(
"\n".join(lines) + ("\n" if lines else ""), encoding="utf-8"
)
env_path.parent.mkdir(parents=True, exist_ok=True)
lines = [f"{key}={current[key]}" for key in sorted(current)]
# Create at owner-only mode from the first byte (O_CREAT honours the mode
# argument subject to umask). A trailing chmod guards pre-existing files.
_mode = _stat.S_IRUSR | _stat.S_IWUSR # 0o600
_fd = os.open(str(env_path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, _mode)
with os.fdopen(_fd, "w", encoding="utf-8") as _f:
_f.write("\n".join(lines) + ("\n" if lines else ""))
try:
env_path.chmod(_mode)
except OSError:
pass
def _provider_has_key(provider_id: str) -> bool: