Fix attended login result auditing and reconcile blocked workplans
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6ef-4273-7fc2-8741-dc96b3e5fe0d
This commit is contained in:
parent
940e164c9b
commit
4c3a0f4d3a
11 changed files with 260 additions and 32 deletions
|
|
@ -330,6 +330,76 @@ def test_cli_login_lane_rejects_persistent_fetch(monkeypatch, tmp_path):
|
|||
assert "requires --exec" in r.output
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"phase,code,outcome",
|
||||
[
|
||||
("success", 0, "ok"),
|
||||
("missing_helper", 10, "login_failed"),
|
||||
("login_failed", 10, "login_failed"),
|
||||
("login_start", 10, "login_failed"),
|
||||
("child_start", 11, "child_failed"),
|
||||
("child_failed", 11, "child_failed"),
|
||||
("child_output", 12, "child_output"),
|
||||
("child_whitespace", 12, "child_output"),
|
||||
("revoke_failed", 13, "revoke_unconfirmed"),
|
||||
("cleanup_failed", 14, "cleanup_failed"),
|
||||
],
|
||||
)
|
||||
def test_cli_attended_result_and_both_audits(monkeypatch, tmp_path, phase, code, outcome):
|
||||
_proxy_env(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
calls = []
|
||||
sentinel = b"hvs.NONPRODUCTION_RESULT_SENTINEL"
|
||||
|
||||
def fake_run(argv, **kw):
|
||||
calls.append(argv)
|
||||
helper = Path(kw["env"]["HOME"]) / ".vault-token"
|
||||
if argv[:2] == ["bao", "login"]:
|
||||
if phase == "login_start":
|
||||
raise OSError("client unavailable")
|
||||
if phase != "missing_helper":
|
||||
helper.write_bytes(sentinel)
|
||||
return subprocess.CompletedProcess(
|
||||
argv, 1 if phase == "login_failed" else 0, sentinel, b""
|
||||
)
|
||||
if argv == ["reviewed-child"]:
|
||||
if phase == "child_start":
|
||||
raise OSError("child unavailable")
|
||||
output = {"child_output": sentinel, "child_whitespace": b" \n"}.get(phase, b"")
|
||||
return subprocess.CompletedProcess(
|
||||
argv, 9 if phase == "child_failed" else 0, b"", output
|
||||
)
|
||||
assert argv == ["bao", "token", "revoke", "-self"]
|
||||
return subprocess.CompletedProcess(
|
||||
argv, 1 if phase == "revoke_failed" else 0, b"", b""
|
||||
)
|
||||
|
||||
monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
|
||||
if phase == "cleanup_failed":
|
||||
def fail_cleanup(*args):
|
||||
raise OSError("cleanup unavailable")
|
||||
monkeypatch.setattr("warden.proxy.shutil.rmtree", fail_cleanup)
|
||||
result = runner.invoke(
|
||||
app, ["access", "login oidc", "--domain", "coulomb_social",
|
||||
"--exec", "--", "reviewed-child"]
|
||||
)
|
||||
assert result.exit_code == code, result.output
|
||||
assert sentinel.decode() not in result.output
|
||||
if code == 10:
|
||||
assert ["reviewed-child"] not in calls
|
||||
if phase not in ("login_start", "cleanup_failed"):
|
||||
assert calls[-1] == ["bao", "token", "revoke", "-self"]
|
||||
if phase != "cleanup_failed":
|
||||
assert not (tmp_path / ".warden-attended-login").exists()
|
||||
for name in ("access-audit.log", "audit.jsonl"):
|
||||
raw = (tmp_path / "state" / name).read_text()
|
||||
assert sentinel.decode() not in raw
|
||||
records = [json.loads(line) for line in raw.splitlines()]
|
||||
login = next(record for record in records if record["action"] == "login")
|
||||
assert login["exit_code"] == code
|
||||
assert login["outcome"] == outcome
|
||||
|
||||
|
||||
def test_attended_login_refuses_read_only_home_before_auth(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
monkeypatch.setattr(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue