railiance-platform/tests/test_public_transport.py

import copy
import importlib.util
import json
from pathlib import Path
import subprocess
import unittest
from unittest.mock import patch
spec=importlib.util.spec_from_file_location('transport',Path(__file__).resolve().parents[1]/'scripts/argocd_public_transport.py')
t=importlib.util.module_from_spec(spec);spec.loader.exec_module(t)

class TransportTests(unittest.TestCase):
 def plan(self):return json.loads((Path(__file__).resolve().parents[1]/'docs/evidence/2026-10-11-issue-core-transport/transport-plan.json').read_text())
 def objects(self,plan):
  result={}
  for item in plan['resources']:
   obj={'metadata':{'uid':item['uid'],'resourceVersion':'17','labels':{'unrelated':'retain'}},'data':{},'spec':{}}
   t.replace(obj,item['path'],copy.deepcopy(item['before']));result[item['kind']]=obj
  return result
 def runner(self,objects,calls,modify_dry=False):
  def run(args,body=None):
   calls.append((args,body))
   if args[:2]==['get','nodes']:return json.dumps({'items':[{'status':{'addresses':[{'address':'92.205.62.239'}]}}]})
   kind=args[3]
   if args[2]=='get':return json.dumps(objects[kind])
   obj=copy.deepcopy(objects[kind])
   for op in json.loads(body):
    if op['op']=='test':self.assertEqual(t.value(obj,op['path']),op['value'])
    else:t.replace(obj,op['path'],op['value'])
   if '--dry-run=server' in args:
    if modify_dry:obj['metadata']['labels']['unrelated']='changed'
    return json.dumps(obj)
   objects[kind]=obj;return kind+'/'+args[4]
  return run
 def receipt(self):return {'resources':[],'mutation_attempted':False}
 def test_dry_run_changes_no_live_values(self):
  plan=self.plan();objects=self.objects(plan);before=copy.deepcopy(objects);calls=[]
  with patch.object(t,'run',self.runner(objects,calls)):r=t.perform(plan,'dry-run',self.receipt())
  self.assertEqual(objects,before);self.assertTrue(r['passed']);self.assertFalse(r['mutation_attempted']);self.assertEqual(sum(args[2]=='patch' for args,body in calls if len(args)>2),2)
 def test_execute_and_guarded_rollback_preserve_other_fields(self):
  plan=self.plan();objects=self.objects(plan);before=copy.deepcopy(objects);calls=[];checkpoints=[];receipt=self.receipt()
  with patch.object(t,'run',self.runner(objects,calls)):
   t.perform(plan,'execute',receipt,lambda:checkpoints.append(copy.deepcopy(receipt)))
   for item in plan['resources']:self.assertEqual(t.value(objects[item['kind']],item['path']),item['after'])
   self.assertTrue(any(x.get('pending_target') for x in checkpoints))
   t.perform(plan,'verify',self.receipt());t.perform(plan,'rollback',self.receipt())
  self.assertEqual(objects,before)
  for args,body in calls:
   if body:
    ops=json.loads(body);self.assertEqual([op['op'] for op in ops],['test','test','test','replace']);self.assertEqual(ops[0]['path'],'/metadata/uid');self.assertEqual(ops[1]['path'],'/metadata/resourceVersion')
 def test_other_target_preimage_drift_stops_before_first_patch(self):
  plan=self.plan();objects=self.objects(plan);objects['appproject']['spec']['sourceRepos'].append('ssh://unreviewed/repo');calls=[]
  with patch.object(t,'run',self.runner(objects,calls)):
   with self.assertRaisesRegex(t.Refused,'^target_preimage_changed$'):t.perform(plan,'execute',self.receipt())
  self.assertFalse(any(args[2]=='patch' for args,body in calls if len(args)>2))
 def test_partial_second_write_is_recorded_and_rollback_can_restore_first(self):
  plan=self.plan();objects=self.objects(plan);before=copy.deepcopy(objects);calls=[];runner=self.runner(objects,calls);receipt=self.receipt();snapshots=[]
  def failing(args,body=None):
   if len(args)>3 and args[2]=='patch' and args[3]=='appproject' and '--dry-run=server' not in args:raise t.Refused('command_outcome_unverified')
   return runner(args,body)
  with patch.object(t,'run',failing):
   with self.assertRaises(t.Refused):t.perform(plan,'execute',receipt,lambda:snapshots.append(copy.deepcopy(receipt)))
  self.assertEqual(receipt['pending_target'],'appproject/railiance-tenants');self.assertTrue(receipt['resources'][0]['changed_verified']);self.assertTrue(snapshots[-1]['mutation_attempted'])
  with patch.object(t,'run',runner):t.perform(plan,'rollback',self.receipt())
  self.assertEqual(objects,before)

 def test_target_uid_change_refused(self):
  plan=self.plan();objects=self.objects(plan);objects['configmap']['metadata']['uid']='foreign'
  with patch.object(t,'run',self.runner(objects,[])):
   with self.assertRaisesRegex(t.Refused,'^target_uid_changed$'):t.perform(plan,'execute',self.receipt())
 def test_server_admission_extra_change_refused(self):
  plan=self.plan();objects=self.objects(plan);before=copy.deepcopy(objects)
  with patch.object(t,'run',self.runner(objects,[],True)):
   with self.assertRaisesRegex(t.Refused,'^dry_run_modified_other_fields$'):t.perform(plan,'execute',self.receipt())
  self.assertEqual(objects,before)
 def test_broader_repo_or_host_replacement_refused(self):
  for kind in ('host','repo'):
   plan=self.plan()
   if kind=='host':plan['resources'][0]['after']='unreviewed replacement\n'
   else:plan['resources'][1]['after'].append('ssh://forgejo.coulomb.social/coulomb/*')
   with self.assertRaises(t.Refused):t.validate(plan)
 def test_wrong_estate_and_raw_canary_refused(self):
  with patch.object(t,'run',return_value=json.dumps({'items':[{'status':{'addresses':[{'address':'192.0.2.8'}]}}]})) as run:
   with self.assertRaisesRegex(t.Refused,'^wrong_estate$'):t.perform(self.plan(),'execute',self.receipt())
   self.assertEqual(run.call_count,1)
  with patch.object(t.subprocess,'run',return_value=subprocess.CompletedProcess([],1,'SECRET_CANARY','SECRET_CANARY')):
   with self.assertRaisesRegex(t.Refused,'^command_outcome_unverified$'):t.run(['get','nodes'])

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