railiance-platform/tests/test_t03_revocation_probe.py

import importlib.util,unittest,json
from pathlib import Path
from unittest.mock import patch
from contextlib import contextmanager
from types import SimpleNamespace
from urllib.error import HTTPError,URLError
spec=importlib.util.spec_from_file_location('native',Path(__file__).resolve().parents[1]/'scripts/t03-native-execution.py');m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
class TestProbe(unittest.TestCase):
 def check(self,status):
  scoped=SimpleNamespace(token='synthetic-revoked-token',token_capabilities=lambda *a,**k:['deny'])
  session=SimpleNamespace(client=scoped,revocation_succeeded=False)
  @contextmanager
  def session_context(role):
   yield session
   session.revocation_succeeded=True;scoped.token=''
  client=SimpleNamespace(addr='http://127.0.0.1:18200',approle_session=session_context)
  class Opener:
   def open(self,req,timeout):
    self_test.assertEqual(req.get_header('X-vault-token'),'synthetic-revoked-token')
    self_test.assertEqual(scoped.token,'')
    if status:raise HTTPError(req.full_url,status,'synthetic',{},None)
    return unittest.mock.MagicMock()
  self_test=self
  with patch('urllib.request.build_opener',return_value=Opener()):return m.verify_session_cleanup(client)
 def test_revoked_explicit_token_refused(self):self.assertEqual(self.check(403)['revoked_token_lookup_status'],403)
 def test_live_token_refused(self):
  with self.assertRaisesRegex(ValueError,'revoked_token_still_usable'):self.check(0)
 def test_server_failure_is_not_revocation_proof(self):
  with self.assertRaisesRegex(ValueError,'revocation_probe_not_definitive'):self.check(500)
 def test_reconciled_prior_phase(self):
  prior={'phase':'verify_attempt_started','failure_code':'revoked_token_still_usable','actions':[{'action':'apply','approval_id':m.PRIOR_IDS['apply'],'exit_code':0,'limits':{'token_ttl':900,'token_max_ttl':1800,'secret_id_ttl':900,'secret_id_num_uses':1,'token_num_uses':8}}]}
  with patch.object(Path,'read_text',return_value=json.dumps(prior)):
   self.assertEqual(m.resume_receipt(),prior)
  prior['actions'][0]['approval_id']='other'
  with patch.object(Path,'read_text',return_value=json.dumps(prior)):
   with self.assertRaisesRegex(ValueError,'prior_apply_not_verified'):m.resume_receipt()
 def test_fresh_apply_branch(self):
  hcl='path "auth/token/lookup-self" {\n  capabilities = ["read"]\n}\n'
  role=dict(m.LIMITS,token_policies=[m.ROLE])
  self.assertEqual(m.fresh_apply_plan(hcl,None,None),(tuple(m.IDS),None))
  actions,row=m.fresh_apply_plan(hcl,hcl+'\n',role)
  self.assertEqual(actions,('verify','exec'))
  self.assertNotIn('apply',actions)
  self.assertEqual(row,{'action':'apply','already_satisfied':True,'limits':dict(m.LIMITS)})
  self.assertNotIn('approval_id',row)
  for existing,live in ((hcl,None),(None,role),('other\n',role),(hcl,dict(role,token_ttl=1)),(hcl,dict(role,token_policies=[m.ROLE,'default'])),(hcl,dict(role,token_policies=m.ROLE))):
   with self.assertRaisesRegex(ValueError,'existing_native_objects_require_reconciliation'):m.fresh_apply_plan(hcl,existing,live)

if __name__=='__main__':unittest.main()