GodopuandClaude Opus 5.5
71ea8095c5
PR #43 review (grok):
(a) A UTF-8 byte-order mark at the start of the file made the first line an
"invalid key", so its setting was dropped. The one parser (envfile.py,
used by the wrapper and mqtt_common) now reads with utf-8-sig.
(b) A key given twice keeps its first value (a later line does not replace
it). The behaviour is unchanged; .mam.env.example's FORMAT section and
envfile.py's docstring now say so.
(c) MESSAGING.md said a MAM_ENV_FILE naming no file logs an error and makes
connections fail closed. Both loaders skip it silently and refuse the
ambient search; a file that exists but cannot be read is the error
(EnvFileError, one ERROR line and rc 1). The text now says that.
Tests A7: BOM in both loaders, first-value-wins in both loaders plus the
FORMAT text, and a missing explicit file loading neither it nor the ambient
.mam.env without an error plus the MESSAGING text (all three fail at 0051fd6).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe
452 lines
21 KiB
Python
452 lines
21 KiB
Python
"""Issue #26 (A): one `.mam.env` parser for the wrapper and the Python loader.
|
|
|
|
- A1: the issue's file no longer kills the real wrapper, runs no shell, warns
|
|
per bad line without leaking values.
|
|
- A2: the real wrapper and the real Python loader read one table the same way.
|
|
- A3: an environment variable beats the file, in both loaders.
|
|
- A4: the parser's command line carries the path only.
|
|
- A5: the files this repo ships or generates parse without a warning.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
ENVFILE = SKILLS / "lib_py" / "delegation" / "envfile.py"
|
|
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
|
|
from lib_py.delegation import envfile # noqa: E402
|
|
|
|
ISSUE_FILE = (
|
|
"AGENT_SESSIONS_YAML= .mam/agent-sessions.yaml\n"
|
|
"X=$(touch pwned)\n"
|
|
"BAD KEY=1\n"
|
|
"noequals\n"
|
|
'MQTT_PASSWORD="s3cr3t-a1"\n'
|
|
)
|
|
CLEAN_FILE = (
|
|
"# comment\n"
|
|
"\n"
|
|
"AGENT_SESSIONS_YAML=.mam/agent-sessions.yaml\n"
|
|
'MQTT_PASSWORD="s3cr3t-a1"\n'
|
|
)
|
|
# key -> value both loaders must produce (A2)
|
|
TABLE_FILE = (
|
|
" SPACED = padded value \n"
|
|
"# ignored=1\n"
|
|
"\n"
|
|
"DQ=\"double quoted\"\n"
|
|
"SQ='single quoted'\n"
|
|
"HALF=\"unmatched\n"
|
|
"MIXED=\"a'\n"
|
|
"EQ=a=b=c\n"
|
|
"EMPTY=\n"
|
|
"LIT=$(echo no) ${HOME} `echo no`\n"
|
|
"export EXPORTED=1\n"
|
|
"1BAD=1\n"
|
|
)
|
|
TABLE = {
|
|
"SPACED": "padded value",
|
|
"DQ": "double quoted",
|
|
"SQ": "single quoted",
|
|
"HALF": '"unmatched',
|
|
"MIXED": "\"a'",
|
|
"EQ": "a=b=c",
|
|
"EMPTY": "",
|
|
"LIT": "$(echo no) ${HOME} `echo no`",
|
|
}
|
|
TABLE_SKIPPED = {"EXPORTED", "export EXPORTED", "1BAD"}
|
|
|
|
|
|
def _env(home: Path, env_file: Path, extra=None) -> dict:
|
|
env = {k: v for k, v in os.environ.items() if k not in TABLE and k not in ("MAM_JOB_TIMEOUT_S", "MQTT_PASSWORD",
|
|
"AGENT_SESSIONS_YAML", "X")}
|
|
env.update({"HOME": str(home), "WORKSPACE_ROOT": str(home), "MAM_ENV_FILE": str(env_file),
|
|
"DELEGATE_JOB_PYTHON": sys.executable, "AGENT_PYTHON_BIN": sys.executable,
|
|
"PYTHONPATH": str(SKILLS), "DELEGATE_JOB_LOGS_DIR": str(home / "logs")})
|
|
if extra:
|
|
env.update(extra)
|
|
return env
|
|
|
|
|
|
def _wrapper_values(env: dict, keys, cwd: Path) -> subprocess.CompletedProcess:
|
|
"""The values the real wrapper exported after loading its env file."""
|
|
probe = "; ".join(f'printf "%s=%s\\0" {k} "${{{k}-<unset>}}"' for k in keys)
|
|
return subprocess.run(["bash", "-c", f'source "{WRAPPER}"; {probe}'], capture_output=True, text=True,
|
|
env={**env, "MAM_DELEGATE_JOB_NO_MAIN": "1"}, cwd=str(cwd), timeout=60)
|
|
|
|
|
|
def _python_values(env: dict, keys) -> dict:
|
|
"""The values the real Python loader (mqtt_common) set from the same file."""
|
|
code = ("import json, os; from lib_py.delegation import mqtt_common; mqtt_common.ensure_env_loaded(); "
|
|
f"print(json.dumps({{k: os.environ.get(k, '<unset>') for k in {list(keys)!r}}}))")
|
|
res = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, env=env, timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
return json.loads(res.stdout.strip().splitlines()[-1])
|
|
|
|
|
|
def _split(out: str) -> dict:
|
|
return dict(item.split("=", 1) for item in out.split("\0") if item)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A1 - the issue's file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_a1_the_issue_file_does_not_kill_the_wrapper_nor_run_shell(tmp_path):
|
|
f = tmp_path / "issue.env"
|
|
f.write_text(ISSUE_FILE)
|
|
res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
|
|
env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
assert "submit" in res.stdout
|
|
assert not (tmp_path / "pwned").exists(), "a value must never be run as shell"
|
|
for n, reason in ((3, "invalid key"), (4, "no '='")):
|
|
assert f"{f}:{n}: skipped ({reason})" in res.stderr, res.stderr
|
|
assert ":1: skipped" not in res.stderr and ":2: skipped" not in res.stderr
|
|
assert "s3cr3t-a1" not in res.stdout + res.stderr
|
|
|
|
pairs, warnings = envfile.parse(str(f))
|
|
assert dict(pairs) == {"AGENT_SESSIONS_YAML": ".mam/agent-sessions.yaml", "X": "$(touch pwned)",
|
|
"MQTT_PASSWORD": "s3cr3t-a1"}
|
|
assert len(warnings) == 2 and not any("s3cr3t" in w for w in warnings)
|
|
|
|
|
|
def test_a1_a_clean_file_loads_silently(tmp_path):
|
|
f = tmp_path / "clean.env"
|
|
f.write_text(CLEAN_FILE)
|
|
res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
|
|
env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
assert "skipped" not in res.stderr and "WARNING" not in res.stderr, res.stderr
|
|
got = _wrapper_values(_env(tmp_path, f), ["AGENT_SESSIONS_YAML", "MQTT_PASSWORD"], tmp_path)
|
|
assert _split(got.stdout) == {"AGENT_SESSIONS_YAML": ".mam/agent-sessions.yaml", "MQTT_PASSWORD": "s3cr3t-a1"}
|
|
|
|
|
|
def test_a1_an_unreadable_parser_fails_loudly(tmp_path):
|
|
"""The wrapper stops when the parser cannot run instead of going on unconfigured."""
|
|
f = tmp_path / "clean.env"
|
|
f.write_text(CLEAN_FILE)
|
|
res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
|
|
env=_env(tmp_path, f, {"DELEGATE_JOB_PYTHON": str(tmp_path / "no-python")}),
|
|
cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 1
|
|
assert f"ERROR: cannot load {f}" in res.stderr
|
|
|
|
|
|
def _python_load(env: dict) -> subprocess.CompletedProcess:
|
|
code = "from lib_py.delegation import mqtt_common; mqtt_common.ensure_env_loaded(); print('LOADED')"
|
|
return subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, env=env, timeout=60)
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["permission", "utf8"])
|
|
def test_a1_a_file_that_cannot_be_read_stops_both_loaders(tmp_path, kind):
|
|
"""Review R1: a file that exists but cannot be read or decoded as a whole
|
|
is an error in the wrapper (rc 1, as before #26) and in the Python entry
|
|
points - never an empty configuration that runs on defaults."""
|
|
f = tmp_path / "broken.env"
|
|
if kind == "permission":
|
|
if os.geteuid() == 0:
|
|
pytest.skip("root reads a 000 file")
|
|
f.write_text("PROBE_A=one\n")
|
|
f.chmod(0)
|
|
cause = "PermissionError"
|
|
else:
|
|
f.write_bytes(b"PROBE_A=one\nPROBE_B=\xff\xfe\n")
|
|
cause = "UnicodeDecodeError"
|
|
try:
|
|
res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
|
|
env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert f"ERROR: {f}: cannot read ({cause})" in res.stderr, res.stderr
|
|
assert f"ERROR: cannot load {f}" in res.stderr, res.stderr
|
|
assert "one" not in res.stderr # no value in any message
|
|
py = _python_load(_env(tmp_path, f))
|
|
assert py.returncode != 0 and "LOADED" not in py.stdout, py.stdout + py.stderr
|
|
assert f"{f}: cannot read ({cause})" in py.stderr and "EnvFileError" in py.stderr, py.stderr[-500:]
|
|
finally:
|
|
f.chmod(0o600)
|
|
|
|
|
|
# Follow-up F2 (PR #43 claude S1): every main that loads the env file, with
|
|
# arguments its parser accepts when it parses before loading.
|
|
ENTRY_POINTS = {
|
|
"publish_event": ["--job", "abcd1234", "--event", "progress", "--registry-dir", "reg"],
|
|
"registry": [],
|
|
"job_subscriber": ["--job", "abcd1234", "--registry-dir", "reg"],
|
|
"cli": [],
|
|
"bench": [],
|
|
"heartbeat": ["--job", "abcd1234", "--interval", "30", "--registry-dir", "reg"],
|
|
"remote": ["run", "--session", "s", "--registry-dir", "reg"],
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["permission", "utf8"])
|
|
@pytest.mark.parametrize("module", sorted(ENTRY_POINTS))
|
|
def test_a1_python_entry_points_report_an_unreadable_file_in_one_line(tmp_path, module, kind):
|
|
"""The wrapper's shape at every Python entry point: one `ERROR:` line
|
|
naming the path and the exception type, rc 1, no traceback, no value."""
|
|
f = tmp_path / "broken.env"
|
|
if kind == "permission":
|
|
if os.geteuid() == 0:
|
|
pytest.skip("root reads a 000 file")
|
|
f.write_text("PROBE_A=one\n")
|
|
f.chmod(0)
|
|
cause = "PermissionError"
|
|
else:
|
|
f.write_bytes(b"PROBE_A=one\nPROBE_B=\xff\xfe\n")
|
|
cause = "UnicodeDecodeError"
|
|
try:
|
|
# main() itself: `-m` adds runpy's warning (the package __init__ imports these modules)
|
|
code = f"import sys; from lib_py.delegation import {module} as m; sys.exit(m.main(sys.argv[1:]))"
|
|
res = subprocess.run([sys.executable, "-c", code, *ENTRY_POINTS[module]],
|
|
capture_output=True, text=True, env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
|
|
finally:
|
|
f.chmod(0o600)
|
|
assert res.returncode == 1, (res.returncode, res.stderr[-500:])
|
|
assert res.stderr.strip().splitlines() == [f"ERROR: {f}: cannot read ({cause})"], res.stderr[-500:]
|
|
assert "one" not in res.stdout + res.stderr
|
|
assert not (tmp_path / "reg").exists(), "nothing ran past the env file"
|
|
|
|
|
|
def test_a1_a_nul_line_is_skipped_by_both_loaders(tmp_path):
|
|
"""Review R4: no environment can carry a NUL byte. Both loaders skip that
|
|
line with a warning naming it (not its value) and keep the lines around it."""
|
|
f = tmp_path / "nul.env"
|
|
f.write_bytes(b"GOOD=ok\nVALUE=left\x00right\nAFTER=yes\n")
|
|
env = _env(tmp_path, f)
|
|
got = _wrapper_values(env, ["GOOD", "VALUE", "AFTER"], tmp_path)
|
|
assert got.returncode == 0, got.stderr
|
|
assert _split(got.stdout) == {"GOOD": "ok", "VALUE": "<unset>", "AFTER": "yes"}
|
|
assert f"{f}:2: skipped (NUL byte)" in got.stderr and "left" not in got.stderr, got.stderr
|
|
assert _python_values(env, ["GOOD", "VALUE", "AFTER"]) == {"GOOD": "ok", "VALUE": "<unset>", "AFTER": "yes"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A2 - one table, two real loaders, same values
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_a2_the_wrapper_and_the_python_loader_read_the_same_values(tmp_path):
|
|
f = tmp_path / "table.env"
|
|
f.write_text(TABLE_FILE)
|
|
env = _env(tmp_path, f)
|
|
keys = list(TABLE) + ["EXPORTED"]
|
|
res = _wrapper_values(env, keys, tmp_path)
|
|
assert res.returncode == 0, res.stderr
|
|
wrapper = _split(res.stdout)
|
|
python = _python_values(env, keys)
|
|
assert wrapper == python == {**TABLE, "EXPORTED": "<unset>"}
|
|
for n in (11, 12):
|
|
assert f"{f}:{n}: skipped (invalid key)" in res.stderr, res.stderr
|
|
pairs, _ = envfile.parse(str(f))
|
|
assert not TABLE_SKIPPED & {k for k, _ in pairs}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A3 - the environment beats the file, in both loaders
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _dry_budget(env: dict, cwd: Path) -> int:
|
|
reg = cwd / "reg"
|
|
res = subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--dry-run",
|
|
"--registry-dir", str(reg), "--prompt", "p"],
|
|
capture_output=True, text=True, env=env, cwd=str(cwd), timeout=120)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
|
|
return json.loads((reg / f"{jid}.json").read_text())["timeout_sec"]
|
|
|
|
|
|
def test_a3_an_environment_value_beats_the_file(tmp_path):
|
|
f = tmp_path / "budget.env"
|
|
f.write_text("MAM_JOB_TIMEOUT_S=50\n")
|
|
(tmp_path / "file").mkdir()
|
|
(tmp_path / "both").mkdir()
|
|
assert _dry_budget(_env(tmp_path, f), tmp_path / "file") == 50
|
|
both = _env(tmp_path, f, {"MAM_JOB_TIMEOUT_S": "70"})
|
|
assert _dry_budget(both, tmp_path / "both") == 70
|
|
assert _python_values(both, ["MAM_JOB_TIMEOUT_S"]) == {"MAM_JOB_TIMEOUT_S": "70"}
|
|
assert _python_values(_env(tmp_path, f), ["MAM_JOB_TIMEOUT_S"]) == {"MAM_JOB_TIMEOUT_S": "50"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A4 - the parser sees the path only
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_a4_the_parser_command_line_carries_the_path_only():
|
|
calls = [l for l in WRAPPER.read_text().splitlines() if "envfile.py" in l and not l.lstrip().startswith("#")]
|
|
assert len(calls) == 1, calls
|
|
assert re.search(r'envfile\.py" --shell "\$TARGET_ENV"\)"', calls[0]), calls[0]
|
|
prefix = [l for l in WRAPPER.read_text().splitlines() if "_env_exports=" in l]
|
|
assert prefix == [' _env_exports="$(SCRIPT_DIR="$SCRIPT_DIR" REPO_ROOT="$REPO_ROOT" TARGET_ENV="$TARGET_ENV" \\'], prefix
|
|
src = WRAPPER.read_text()
|
|
assert not re.search(r'^\s*(source|\.)\s+"?\$TARGET_ENV', src, re.M), "the env file must never be sourced"
|
|
assert "set -a" not in src
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A5 - the files the repo ships or generates follow the format
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _install_seed(tmp_path: Path) -> Path:
|
|
"""The .mam.env install.sh writes, from its own heredoc run by bash."""
|
|
src = (REPO_ROOT / "deploy" / "install.sh").read_text()
|
|
m = re.search(r'cat <<EOF > "\$MAM_ENV"\n(.*?\nEOF)\n', src, re.S)
|
|
assert m, "install.sh no longer seeds .mam.env with a heredoc"
|
|
script = f'MAM_ENV="{tmp_path / "seed.env"}"; MAM_PORT=1883; MAM_CLIENT_PREFIX=hermes\ncat <<EOF > "$MAM_ENV"\n{m.group(1)}\n'
|
|
subprocess.run(["bash", "-c", script], check=True, env={**os.environ, "HOME": str(tmp_path)}, timeout=30)
|
|
return tmp_path / "seed.env"
|
|
|
|
|
|
def test_a5_shipped_and_generated_env_files_parse_without_warnings(tmp_path):
|
|
ws = tmp_path / "ws"
|
|
(ws / "deploy").mkdir(parents=True)
|
|
(ws / ".mam.env.example").write_text((REPO_ROOT / ".mam.env.example").read_text())
|
|
(ws / "deploy" / "generate-env.sh").write_text((REPO_ROOT / "deploy" / "generate-env.sh").read_text())
|
|
subprocess.run(["bash", str(ws / "deploy" / "generate-env.sh")], check=True, capture_output=True, timeout=30)
|
|
for f in (REPO_ROOT / ".mam.env.example", ws / ".mam.env", _install_seed(tmp_path)):
|
|
pairs, warnings = envfile.parse(str(f))
|
|
assert warnings == [], (f, warnings)
|
|
assert not any("$" in v or "`" in v for _, v in pairs), (f, pairs)
|
|
|
|
# The template's commented assignments are what a user uncomments: they must
|
|
# parse too, and hold no shell syntax the parser would keep literally.
|
|
commented = tmp_path / "uncommented.env"
|
|
lines = [l[2:] for l in (REPO_ROOT / ".mam.env.example").read_text().splitlines()
|
|
if re.match(r"# [A-Za-z_][A-Za-z0-9_]*=", l)]
|
|
assert len(lines) > 20
|
|
commented.write_text("\n".join(lines) + "\n")
|
|
pairs, warnings = envfile.parse(str(commented))
|
|
assert warnings == [] and len(pairs) == len(lines)
|
|
assert not [(k, v) for k, v in pairs if "$" in v or "`" in v]
|
|
|
|
|
|
@pytest.mark.parametrize("missing", [True, False])
|
|
def test_a5_shell_mode_exit_code_is_zero_for_absent_and_present_files(tmp_path, missing):
|
|
f = tmp_path / "x.env"
|
|
if not missing:
|
|
f.write_text("A=1\n")
|
|
res = subprocess.run([sys.executable, "-I", str(ENVFILE), "--shell", str(f)], capture_output=True, text=True,
|
|
env={k: v for k, v in os.environ.items() if k != "A"}, timeout=30)
|
|
assert res.returncode == 0
|
|
assert res.stdout == ("" if missing else "export A=1\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A6 - bash's own variables are never assigned from the file (review R1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# RANDOM/SECONDS/LINENO/OPTIND/SRANDOM are evaluated arithmetically on
|
|
# assignment, so even a shlex-quoted value runs its $(...) in the wrapper.
|
|
ARITH_SPECIALS = ["RANDOM", "SECONDS", "LINENO", "OPTIND", "SRANDOM"]
|
|
|
|
|
|
@pytest.mark.parametrize("key", ARITH_SPECIALS)
|
|
def test_a6_a_reserved_key_runs_nothing_in_the_real_wrapper(tmp_path, key):
|
|
f = tmp_path / "special.env"
|
|
f.write_text(f"{key}=BASH_VERSINFO[$(touch marker)0]\nSAFE=BASH_VERSINFO[$(touch marker2)0]\n")
|
|
env = _env(tmp_path, f)
|
|
env.pop(key, None)
|
|
res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True, env=env,
|
|
cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
assert not (tmp_path / "marker").exists(), f"{key}= from the file ran shell"
|
|
assert not (tmp_path / "marker2").exists()
|
|
assert f"{f}:1: skipped (reserved shell variable {key})" in res.stderr, res.stderr
|
|
assert "touch" not in res.stderr, "the warning must not echo the value"
|
|
# the same value under an ordinary key stays literal (positive case)
|
|
got = _wrapper_values(env, ["SAFE"], tmp_path)
|
|
assert _split(got.stdout) == {"SAFE": "BASH_VERSINFO[$(touch marker2)0]"}
|
|
assert not (tmp_path / "marker").exists() and not (tmp_path / "marker2").exists()
|
|
|
|
|
|
def test_a6_both_loaders_skip_reserved_keys(tmp_path):
|
|
f = tmp_path / "reserved.env"
|
|
keys = ["BASH_ENV", "PS4", "IFS", "SHELLOPTS", "COMP_WORDBREAKS", "READLINE_LINE", "RANDOM"]
|
|
f.write_text("".join(f"{k}=x\n" for k in keys) + "PLAIN=ok\n")
|
|
env = {k: v for k, v in _env(tmp_path, f).items() if k not in keys}
|
|
python = _python_values(env, keys + ["PLAIN"])
|
|
assert python == {**{k: "<unset>" for k in keys}, "PLAIN": "ok"}
|
|
pairs, warnings = envfile.parse(str(f))
|
|
assert pairs == [("PLAIN", "ok")]
|
|
assert [w.rsplit("(", 1)[1] for w in warnings] == [f"reserved shell variable {k})" for k in keys]
|
|
res = subprocess.run([sys.executable, "-I", str(ENVFILE), "--shell", str(f)], capture_output=True,
|
|
text=True, env=env, timeout=30)
|
|
assert res.stdout == "export PLAIN=ok\n"
|
|
|
|
|
|
def test_a6_the_reserved_set_covers_every_variable_a_clean_bash_defines():
|
|
res = subprocess.run(["env", "-i", "bash", "--norc", "--noprofile", "-c", "compgen -v"],
|
|
capture_output=True, text=True, check=True, timeout=30)
|
|
ordinary = {"PATH", "TERM"} # environment, not shell state; .mam.env may set them as before
|
|
uncovered = [v for v in res.stdout.split() if v not in ordinary and not envfile.is_reserved(v)]
|
|
assert not uncovered, f"bash defines these but envfile.RESERVED misses them: {uncovered}"
|
|
|
|
|
|
def test_a6_the_file_cannot_replace_the_wrappers_own_variables(tmp_path):
|
|
f = tmp_path / "own.env"
|
|
f.write_text(f"SCRIPT_DIR={tmp_path}/evil\nREPO_ROOT={tmp_path}/evil\nTARGET_ENV={tmp_path}/evil\n")
|
|
(tmp_path / "evil").mkdir()
|
|
(tmp_path / "evil" / "lib.sh").write_text("touch pwned-lib\n")
|
|
res = subprocess.run(["bash", str(WRAPPER), "--help"], capture_output=True, text=True,
|
|
env=_env(tmp_path, f), cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
assert not (tmp_path / "pwned-lib").exists()
|
|
got = _wrapper_values(_env(tmp_path, f), ["SCRIPT_DIR", "TARGET_ENV"], tmp_path)
|
|
assert _split(got.stdout) == {"SCRIPT_DIR": str(WRAPPER.parent), "TARGET_ENV": str(f)}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A7 - follow-ups F3 (PR #43 review grok)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_a7_a_leading_bom_is_not_part_of_the_first_key(tmp_path):
|
|
"""An editor's byte-order mark used to turn the first line into
|
|
"invalid key" and drop the first setting, in both loaders."""
|
|
f = tmp_path / "bom.env"
|
|
f.write_bytes(b"\xef\xbb\xbfFIRST=one\nSECOND=two\n")
|
|
env = _env(tmp_path, f)
|
|
res = _wrapper_values(env, ["FIRST", "SECOND"], tmp_path)
|
|
assert res.returncode == 0, res.stderr
|
|
assert _split(res.stdout) == {"FIRST": "one", "SECOND": "two"}
|
|
assert "skipped" not in res.stderr, res.stderr
|
|
assert _python_values(env, ["FIRST", "SECOND"]) == {"FIRST": "one", "SECOND": "two"}
|
|
assert envfile.parse(str(f)) == ([("FIRST", "one"), ("SECOND", "two")], [])
|
|
|
|
|
|
def test_a7_a_repeated_key_keeps_its_first_value_as_documented(tmp_path):
|
|
f = tmp_path / "dup.env"
|
|
f.write_text("DUP=first\nDUP=second\n")
|
|
env = _env(tmp_path, f)
|
|
res = _wrapper_values(env, ["DUP"], tmp_path)
|
|
assert res.returncode == 0, res.stderr
|
|
assert _split(res.stdout) == {"DUP": "first"} == _python_values(env, ["DUP"])
|
|
fmt = (REPO_ROOT / ".mam.env.example").read_text().split("# SECURITY:")[0]
|
|
assert "a key given twice keeps its FIRST value" in fmt, "document it in the FORMAT section"
|
|
|
|
|
|
def test_a7_a_missing_explicit_file_loads_nothing_and_says_nothing(tmp_path):
|
|
"""MAM_ENV_FILE naming no file: no error, and no fallback to the file the
|
|
ambient search would find - in both loaders, and so in MESSAGING.md."""
|
|
(tmp_path / ".mam.env").write_text("PROBE_AMBIENT=found\n")
|
|
absent = tmp_path / "absent.env"
|
|
env = _env(tmp_path, absent)
|
|
res = _wrapper_values(env, ["PROBE_AMBIENT"], tmp_path)
|
|
assert res.returncode == 0, res.stderr
|
|
assert _split(res.stdout) == {"PROBE_AMBIENT": "<unset>"}
|
|
assert "ERROR" not in res.stderr and str(absent) not in res.stderr, res.stderr
|
|
assert _python_values(env, ["PROBE_AMBIENT"]) == {"PROBE_AMBIENT": "<unset>"}
|
|
doc = next(l for l in (REPO_ROOT / "MESSAGING.md").read_text().splitlines() if "Explicit Override (`MAM_ENV_FILE`)" in l)
|
|
assert "does not exist, it is skipped silently" in doc and "an error is logged" not in doc, doc
|