mirror of
https://github.com/TencentCloud/Octop.git
synced 2026-10-19 06:10:07 +08:00
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""tests/unit/test_env_file.py"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from octop.infra.utils.env_file import (
|
|
apply_env_file,
|
|
apply_env_file_replace,
|
|
format_env_file,
|
|
load_env_file,
|
|
parse_env_text,
|
|
save_env_file,
|
|
search_env_changed,
|
|
)
|
|
|
|
|
|
def test_parse_and_roundtrip() -> None:
|
|
text = '# comment\nFOO=bar\nexport BAZ="hello world"\nEMPTY=\n'
|
|
parsed = parse_env_text(text)
|
|
assert parsed == {"FOO": "bar", "BAZ": "hello world", "EMPTY": ""}
|
|
assert format_env_file(parsed).splitlines() == [
|
|
'BAZ="hello world"',
|
|
"EMPTY=",
|
|
"FOO=bar",
|
|
]
|
|
|
|
|
|
def test_save_and_load(tmp_path: Path, monkeypatch) -> None:
|
|
path = tmp_path / "env"
|
|
save_env_file(path, {"API_KEY": "secret", "COUNT": "1"})
|
|
assert load_env_file(path)["API_KEY"] == "secret"
|
|
monkeypatch.delenv("API_KEY", raising=False)
|
|
apply_env_file(path)
|
|
import os
|
|
|
|
assert os.environ["API_KEY"] == "secret"
|
|
|
|
|
|
def test_apply_env_file_replace_unsets_removed_keys(tmp_path: Path, monkeypatch) -> None:
|
|
import os
|
|
|
|
path = tmp_path / "env"
|
|
save_env_file(path, {"KEEP": "1", "GONE": "2"})
|
|
monkeypatch.setenv("GONE", "2")
|
|
monkeypatch.setenv("KEEP", "1")
|
|
monkeypatch.setenv("UNRELATED", "host")
|
|
save_env_file(path, {"KEEP": "1"})
|
|
apply_env_file_replace(path, previous={"KEEP": "1", "GONE": "2"})
|
|
assert os.environ["KEEP"] == "1"
|
|
assert "GONE" not in os.environ
|
|
assert os.environ["UNRELATED"] == "host"
|
|
|
|
|
|
def test_overlay_stdio_spec_env_skips_protected() -> None:
|
|
from octop.infra.utils.env_file import overlay_stdio_spec_env
|
|
|
|
spec = overlay_stdio_spec_env(
|
|
{"transport": "stdio", "command": "npx", "env": {"SPEC": "1"}},
|
|
{"TAVILY_API_KEY": "tvly", "OCTOP_HOME": "/nope", "HOME": "/hack"},
|
|
)
|
|
assert spec["env"]["TAVILY_API_KEY"] == "tvly"
|
|
assert spec["env"]["SPEC"] == "1"
|
|
assert "OCTOP_HOME" not in spec["env"]
|
|
assert "HOME" not in spec["env"]
|
|
|
|
|
|
def test_overlay_stdio_mcp_configs_applies_to_stdio_only() -> None:
|
|
from octop.infra.utils.env_file import overlay_stdio_mcp_configs
|
|
|
|
configs = overlay_stdio_mcp_configs(
|
|
{
|
|
"stdio_srv": {"transport": "stdio", "command": "npx"},
|
|
"http_srv": {"transport": "http", "url": "http://localhost"},
|
|
},
|
|
{"MY_KEY": "secret"},
|
|
)
|
|
assert configs["stdio_srv"]["env"]["MY_KEY"] == "secret"
|
|
assert "env" not in configs["http_srv"]
|
|
|
|
|
|
def test_search_env_changed_detects_tavily() -> None:
|
|
assert search_env_changed({}, {"TAVILY_API_KEY": "x"}) is True
|
|
assert search_env_changed({"FOO": "1"}, {"FOO": "2"}) is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
[
|
|
r"C:\\Users\\me\\AppData",
|
|
'say "hi"',
|
|
r"a\\b\\\"c",
|
|
"ends\\\\",
|
|
],
|
|
)
|
|
def test_backslash_and_quote_values_roundtrip(value: str) -> None:
|
|
"""The escaping format_env_file writes is undone by parsing, so values with
|
|
backslashes (e.g. Windows paths) or embedded quotes survive a save/load."""
|
|
text = format_env_file({"KEY": value})
|
|
assert parse_env_text(text)["KEY"] == value
|
|
|
|
|
|
def test_escape_roundtrip_is_stable_across_cycles() -> None:
|
|
"""Previously every save/load cycle doubled the backslashes in a value."""
|
|
value = r"C:\\Users\\me\\AppData"
|
|
loaded = parse_env_text(format_env_file({"KEY": value}))
|
|
assert loaded["KEY"] == value
|
|
assert parse_env_text(format_env_file(loaded))["KEY"] == value
|
|
|
|
|
|
def test_multiline_value_survives_roundtrip() -> None:
|
|
"""format_env_file quotes values containing whitespace but leaves line breaks raw."""
|
|
values = {
|
|
"PEM_KEY": "-----BEGIN KEY-----\nMIIB\n-----END KEY-----",
|
|
"CRLF": "line1\r\nline2",
|
|
}
|
|
|
|
assert parse_env_text(format_env_file(values)) == {
|
|
"PEM_KEY": "-----BEGIN KEY-----\nMIIB\n-----END KEY-----",
|
|
"CRLF": "line1\nline2",
|
|
}
|
|
|
|
|
|
def test_quoted_value_with_inner_quote_does_not_swallow_following_lines() -> None:
|
|
"""Only an unterminated quote continues onto the next line."""
|
|
text = 'MALFORMED="abc"def\nNEXT=1\n'
|
|
|
|
assert parse_env_text(text) == {"MALFORMED": '"abc"def', "NEXT": "1"}
|