2e2c31d237
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
33 lines
2.6 KiB
Python
33 lines
2.6 KiB
Python
import base64,importlib.util,json,time,unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
spec=importlib.util.spec_from_file_location("factor_renew",Path(__file__).resolve().parents[1]/"scripts/keycape_factor_renew.py")
|
|
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
|
class FactorRenewalTests(unittest.TestCase):
|
|
def drive(self,provider_failure=False,publish_failure=False,revoke_failure=False):
|
|
self.published=False;token="synthetic."+base64.urlsafe_b64encode(json.dumps({"exp":int(time.time())+3600}).encode()).decode()+".sig"
|
|
def http(url,body=None,headers=None,method=None,form=False):
|
|
if url.endswith('/auth/kubernetes/login'):return 200,{"auth":{"client_token":"synthetic-bao-session"}}
|
|
if url.endswith('/auth/token/revoke-self'):return (403 if revoke_failure else 204),{}
|
|
if url.endswith(m.ISSUER):return 200,{"data":{"data":{"REQUEST":m.PROVENANCE,"USERNAME":m.USER,"PASSWORD":"synthetic-service-password"}}}
|
|
if url.endswith('/auth'):
|
|
return (401,{}) if provider_failure else (200,{"result":{"status":True,"value":{"role":"admin","username":m.USER,"token":token}}})
|
|
if '/token/?' in url:return 200,{"result":{"status":True,"value":{"tokens":[{}],"count":1}}}
|
|
if url.endswith('/policy/'):return 403,{}
|
|
if url.endswith(m.TARGET):
|
|
if body:
|
|
if publish_failure:return 400,{}
|
|
self.assertEqual(1,body['options']['cas']);self.assertNotIn('PASSWORD',body['data']);self.published=True;return 200,{}
|
|
return 200,{"data":{"data":{"REQUEST":m.PROVENANCE,"TOKEN":token if self.published else "synthetic-old"},"metadata":{"version":2 if self.published else 1}}}
|
|
raise AssertionError('Unexpected endpoint')
|
|
with patch.object(m,'http',side_effect=http),patch.object(Path,'read_text',return_value='synthetic-kubernetes-jwt'):return m.run()
|
|
def test_renewal_publishes_verified_token_and_revokes_session(self):
|
|
r=self.drive();self.assertTrue(r['success']);self.assertEqual(2,r['kv_version']);self.assertTrue(r['session_revoked'])
|
|
self.assertNotIn('synthetic',json.dumps(r))
|
|
def test_rejected_provider_retains_current_token(self):
|
|
r=self.drive(provider_failure=True);self.assertFalse(r['success']);self.assertFalse(self.published);self.assertTrue(r['session_revoked'])
|
|
self.assertTrue(self.drive()['success'])
|
|
def test_cas_failure_is_not_success(self):
|
|
r=self.drive(publish_failure=True);self.assertFalse(r['success']);self.assertFalse(self.published);self.assertEqual('publish',r['phase'])
|
|
def test_cleanup_failure_is_reported(self):
|
|
r=self.drive(revoke_failure=True);self.assertFalse(r['success']);self.assertEqual('session_cleanup',r['phase'])
|