Prepare exact informed-decision catalogue maintenance decision packet

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

tests/test_catalogue_bootstrap_transport.py

@@ -0,0 +1,65 @@
+"""Offline transport guards with explicit fixture receipts; never dispatch SSH.
+The mocked packet is not founder authority or a native execution test.
+"""
+import base64
+import contextlib
+import hashlib
+import importlib.util
+import io
+import json
+from pathlib import Path
+import re
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+transport=Path(__file__).resolve().parents[1]/'tools'/'run_approved_catalogue_bootstrap.py'
+if not transport.is_file():transport=Path('/home/worsch/user-engine/tools/run_approved_catalogue_bootstrap.py') # staged offline test fallback only
+spec=importlib.util.spec_from_file_location('catalogue_transport',transport)
+m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
+IMAGE='forgejo.coulomb.social/coulomb/user-engine@sha256:f239bbf332b8eb87320c86390fddca1dcb4f7c628a7af75e79dee8058d446a7b'
+UID='11111111-1111-4111-8111-111111111111'
+class Tests(unittest.TestCase):
+ def setUp(self):
+  self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name)
+  self.source=self.root/'executor.py';self.source.write_text('# explicit fixture source, not a native executor\n')
+  self.grant=self.root/'private.json';self.grant.write_text(json.dumps({'fixture':'not-authority'}));self.grant.chmod(0o600)
+  self.args=['tool','--executor',str(self.source),'--executor-sha256',hashlib.sha256(self.source.read_bytes()).hexdigest(),'--grant-file',str(self.grant),'--grant-sha256',hashlib.sha256(b'{"fixture":"not-authority"}').hexdigest(),'--expected-pod-uid',UID]
+ def tearDown(self):self.tmp.cleanup()
+ def call(self,args=None):
+  output=io.StringIO()
+  with patch('sys.argv',args or self.args),patch.object(m.subprocess,'run',return_value=SimpleNamespace(returncode=0,stdout='{"result":"created"}')) as dispatch,contextlib.redirect_stdout(output):
+   status=m.main()
+  return status,dispatch,json.loads(output.getvalue())
+ def test_wrong_executor_or_grant_hash_refuses_before_ssh(self):
+  for field in ['--executor-sha256','--grant-sha256']:
+   args=self.args.copy();args[args.index(field)+1]='0'*64
+   status,dispatch,out=self.call(args);self.assertEqual(status,2);dispatch.assert_not_called();self.assertEqual(out['result'],'refused-or-rolled-back')
+ def test_nonprivate_or_symlink_grant_refuses_before_ssh(self):
+  self.grant.chmod(0o644);status,dispatch,_=self.call();self.assertEqual(status,2);dispatch.assert_not_called()
+  self.grant.chmod(0o600);link=self.root/'link';link.symlink_to(self.grant);args=self.args.copy();args[args.index('--grant-file')+1]=str(link)
+  status,dispatch,_=self.call(args);self.assertEqual(status,2);dispatch.assert_not_called()
+ def test_invalid_uid_cannot_enter_remote_python_literal(self):
+  args=self.args.copy();args[args.index('--expected-pod-uid')+1]="invalid'); __import__('os').system('false'); #"
+  status,dispatch,_=self.call(args);self.assertEqual(status,2);dispatch.assert_not_called()
+ def test_transport_binds_source_grant_mode_to_stdin_without_credentials(self):
+  status,dispatch,_=self.call();self.assertEqual(status,0);self.assertEqual(dispatch.call_args.args[0],['ssh','railiance01','python3 -'])
+  remote=dispatch.call_args.kwargs['input'];self.assertIn(UID,remote);self.assertIn(IMAGE,remote)
+  native=base64.b64decode(re.search(r"base64.b64decode\('([^']+)'\)",remote).group(1)).decode()
+  packet=json.loads(base64.b64decode(re.search(r"base64.b64decode\('([^']+)'\)",native).group(1)))
+  self.assertEqual(base64.b64decode(packet['source']),self.source.read_bytes());self.assertEqual(packet['mode'],'dry-run');self.assertEqual(packet['grant'],{'fixture':'not-authority'})
+  self.assertIn("os.environ['USER_ENGINE_DATABASE_URL']",native);self.assertIn("os.environ['USER_ENGINE_FLEX_AUTH_TOKEN_FILE']",native)
+ def test_remote_rejects_wrong_uid_not_ready_and_changed_image_before_exec(self):
+  _,dispatch,_=self.call();remote=dispatch.call_args.kwargs['input']
+  for uid,ready,image in [('22222222-2222-4222-8222-222222222222',True,IMAGE),(UID,False,IMAGE),(UID,True,'fixture:changed')]:
+   pods={'items':[{'metadata':{'uid':uid,'name':'fixture-pod'},'status':{'phase':'Running','conditions':[{'type':'Ready','status':'True' if ready else 'False'}]},'spec':{'containers':[{'name':'portal','image':image}]}}]}
+   out=io.StringIO()
+   with patch('subprocess.run',return_value=SimpleNamespace(returncode=0,stdout=json.dumps(pods))) as native_dispatch,contextlib.redirect_stdout(out):exec(compile(remote,'<fixture-remote>','exec'),{'__name__':'fixture'})
+   self.assertEqual(native_dispatch.call_count,1);self.assertEqual(json.loads(out.getvalue())['result'],'refused-or-rolled-back')
+ def test_remote_exact_ready_pod_executes_only_fixed_portal_stdin(self):
+  _,dispatch,_=self.call();remote=dispatch.call_args.kwargs['input'];pods={'items':[{'metadata':{'uid':UID,'name':'fixture-pod'},'status':{'phase':'Running','conditions':[{'type':'Ready','status':'True'}]},'spec':{'containers':[{'name':'portal','image':IMAGE}]}}]}
+  out=io.StringIO()
+  with patch('subprocess.run',side_effect=[SimpleNamespace(returncode=0,stdout=json.dumps(pods)),SimpleNamespace(returncode=0,stdout=b'{"result":"created"}')]) as native_dispatch,contextlib.redirect_stdout(out):exec(compile(remote,'<fixture-remote>','exec'),{'__name__':'fixture'})
+  self.assertEqual(native_dispatch.call_count,2);self.assertEqual(native_dispatch.call_args.args[0],['kubectl','exec','-i','-n','user-engine','fixture-pod','-c','portal','--','python3','-']);self.assertIsInstance(native_dispatch.call_args.kwargs['input'],bytes)
+if __name__=='__main__':unittest.main()