Preserve writable OIDC fields on the loopback callback update.
The second attended attempt spawned the owner command, then failed closed. Stop posting the entire role read-back and requiring exact dict equality. Write a metadata receipt so the next failure has a class, not silence. Assistant: grok Assistant-Session: 01a0a23b-3bf0-7341-b4e5-9dc05f72573a
This commit is contained in:
parent
7496d9fab5
commit
041f6bdc51
4 changed files with 185 additions and 27 deletions
|
|
@ -29,14 +29,36 @@ def test_preserves_all_other_settings():
|
|||
assert len(writes) == 1
|
||||
|
||||
|
||||
def test_write_payload_omits_read_only_fields():
|
||||
current = dict(role(), request_id='synthetic', lease_duration=0)
|
||||
writes = []
|
||||
def write(value):
|
||||
writes.append(value)
|
||||
current.update(value)
|
||||
assert m.update(lambda: copy.deepcopy(current), write)
|
||||
assert 'request_id' not in writes[0]
|
||||
assert 'lease_duration' not in writes[0]
|
||||
assert writes[0]['allowed_redirect_uris'][-1] == m.CALLBACK
|
||||
|
||||
|
||||
def test_readback_allows_normalized_extra_fields():
|
||||
current = role()
|
||||
def read():
|
||||
return dict(current, lease_duration=0)
|
||||
def write(value):
|
||||
current.update(value)
|
||||
assert m.update(read, write)
|
||||
assert m.CALLBACK in current['allowed_redirect_uris']
|
||||
|
||||
|
||||
def test_observed_concurrent_change_prevents_write():
|
||||
reads = iter([role(), dict(role(), token_ttl=300)])
|
||||
with pytest.raises(ValueError):
|
||||
with pytest.raises(m.Refused, match='role_changed_before_write'):
|
||||
m.update(lambda: next(reads), lambda _: pytest.fail('must not write'))
|
||||
|
||||
|
||||
def test_failed_readback_is_not_success():
|
||||
with pytest.raises(ValueError):
|
||||
with pytest.raises(m.Refused, match='readback_callback_missing'):
|
||||
m.update(role, lambda _: None)
|
||||
|
||||
|
||||
|
|
@ -46,10 +68,25 @@ def test_unexpected_live_role_refused(monkeypatch, mutation):
|
|||
import subprocess
|
||||
payload = dict(role(), **mutation)
|
||||
monkeypatch.setattr(m.subprocess, 'run', lambda *a, **kw: subprocess.CompletedProcess(a, 0, json.dumps({'data': payload}).encode()))
|
||||
with pytest.raises(ValueError):
|
||||
with pytest.raises(m.Refused, match='unexpected_role'):
|
||||
m.read_role()
|
||||
|
||||
|
||||
def test_receipt_records_apply_and_failure(tmp_path, monkeypatch):
|
||||
import json
|
||||
receipt = tmp_path / 'loopback.json'
|
||||
monkeypatch.setattr(m, 'require_attended', lambda: None)
|
||||
monkeypatch.setattr(m, 'update', lambda: True)
|
||||
assert m.main(['--receipt', str(receipt)]) == 0
|
||||
assert json.loads(receipt.read_text())['status'] == 'applied'
|
||||
failed = tmp_path / 'failed.json'
|
||||
def boom():
|
||||
raise m.Refused('bao_write_failed')
|
||||
monkeypatch.setattr(m, 'update', boom)
|
||||
assert m.main(['--receipt', str(failed)]) == 1
|
||||
assert json.loads(failed.read_text())['status'] == 'bao_write_failed'
|
||||
|
||||
|
||||
def test_attended_wrapper_requires_absolute_existing_executable(tmp_path, monkeypatch):
|
||||
import os
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue