CHAT_OUTPUT_BEGIN
COMMAND_ID=SUPPORT-260915-INFRA-DR-BACKUPV2-PVE-LXC-VALIDATE-TOOLS-CONTRACT-RCA-424R1
STATUS=OK
RC=0
HOST=pve01
MODE=read-only
COMPONENT=infra-dr-backupv2-pve-lxc-validate-tools-contract-rca
REFERENCE_REGISTER_CHECK=OK
REFERENCE_SHA256=5a3d8e5154c41cb582a4e0aca68090be1f0138918bf82131a948df326f9d8d66
ERROR_REGISTER_CHECK=OK
ERROR_REGISTER_SHA256=3b09a553ec0f527ed3afeed4753f52a74ee3036045bbfb3c685e9f8af4ba7ba0
COMMAND_SHA256=0a62d1fafa30a57ee7569e92539141b0519987c9e68f21baf810196dc697a3b4
DUPLICATE_FAILED_COMMAND_BLOCKED=false
EXECUTION_STARTED=true
CHANGE_DECLARED=false
RESULT_CONTRACT_VALID=true
RESULT_CONTRACT_STATUS=NOT_APPLICABLE
RESULT_CONTRACT_ERROR=NONE
COMMAND_RC=0
CHANGES_MADE=false
ROLLBACK_STARTED=false
ROLLBACK_RESTORED=null
MUTATION_OUTCOME=NO_MUTATION
SANITIZED=yes
SECRETS_INCLUDED=no
PRIVATE_ADDRESSES_INCLUDED=no
RAW_EVIDENCE_SHA256=59dd1e9db7cc8348d95fb1899335ec05e3f7d515a00964d19c7d9b929a473076
SANITIZED_OUTPUT_SHA256=59dd1e9db7cc8348d95fb1899335ec05e3f7d515a00964d19c7d9b929a473076
OUTPUT_BEGIN
COMMAND_ID=SUPPORT-260915-INFRA-DR-BACKUPV2-PVE-LXC-VALIDATE-TOOLS-CONTRACT-RCA-424R1
MODE=READ_ONLY_PVE_LXC_VALIDATE_TOOLS_CONTRACT_RCA
NO_PVE_SUBPROCESS=YES
NO_BACKUP_EXECUTION=YES
NO_VZDUMP_EXECUTION=YES
NO_CLOUD_ACCESS=YES
NO_SYSTEMD_MUTATION=YES
NO_FILE_MUTATION=YES
ANCHOR_BINDINGS=PASS
PRODUCTION_AND_CANDIDATE_IDENTITY=PASS
PVE265_VALIDATE_TOOLS_CONTRACT={"comparisons":["set(tools) == {'vzdump', 'zstd'}","set(value) == {'path', 'sha256'}","t.sha(b) == value['sha256']","type(tools) is dict","type(value) is dict"],"need_calls":[{"code":"TOOLS_SCHEMA","line":2,"predicate":"type(tools) is dict and set(tools) == {'vzdump', 'zstd'}"},{"code":"TOOL_SCHEMA","line":4,"predicate":"type(value) is dict and set(value) == {'path', 'sha256'} and path_ok(value['path']) and core.valid_hash(value['sha256'])"},{"code":"TOOL_CHANGED_OR_NOT_EXECUTABLE","line":6,"predicate":"t.sha(b) == value['sha256'] and bool(identity[5] & 73)"}],"returns":[],"signature":"(tools)","source":"def validate_tools(tools):\n    need(type(tools) is dict and set(tools)=={'vzdump','zstd'},'TOOLS_SCHEMA')\n    for value in tools.values():\n        need(type(value) is dict and set(value)=={'path','sha256'} and path_ok(value['path']) and core.valid_hash(value['sha256']),'TOOL_SCHEMA')\n        b,identity=t.read_secure(Path(value['path']),128*1024**2)\n        need(t.sha(b)==value['sha256'] and bool(identity[5]&0o111),'TOOL_CHANGED_OR_NOT_EXECUTABLE')\n","source_sha256":"3b11c01ddd27d9c9e8591c33d788dafcba8989e1e41efb716dc268e8eb838b80","subscripts":["identity[5]","value['path']","value['sha256']"]}
VALIDATE_TOOLS_REFERENCED_GLOBALS={}
PVE265_PRODUCE_CONTRACT={"comparisons":["h == result['sha256']","job['type'] == 'lxc'","list(identity) == result['identity']","n == result['size']","os.listdir(parent) == []","rc == 0","v.f_frsize * v.f_bavail >= LOCAL_RESERVE"],"need_calls":[{"code":"GUARD_REQUIRED","line":7,"predicate":"callable(guard)"},{"code":"ARCHIVE_COMPRESSION_TEST_FAILED_RETAINED","line":19,"predicate":"rc == 0"},{"code":"PRODUCER_READBACK_MISMATCH","line":21,"predicate":"n == result['size'] and h == result['sha256'] and (list(identity) == result['identity'])"},{"code":"PRODUCER_WORKSPACE_NOT_EMPTY","line":9,"predicate":"os.listdir(parent) == []"},{"code":"STAGING_RESERVE_REACHED_PARTIAL_RETAINED","line":14,"predicate":"v.f_frsize * v.f_bavail >= LOCAL_RESERVE"}],"returns":[{"expr":"{**result, 'path': str(dest), 'vmid': job['vmid'], 'type': job['type'], 'node': job['node'], 'config_sha256': job['config_sha256'], 'compression_test_ok': True, 'restored': False, 'state': 'PRODUCED_COMPRESSION_CHECKED_NOT_CLOUD_SEALED', 'local_payload_retained': True}","line":23}],"signature":"(job, work, tools, guard, seconds=14400)","source":"def produce(job,work,tools,guard,seconds=14400):\n    \"\"\"Future explicit execution API. Caller must recheck full plan/coverage in guard.\n\n    Compression CRC and SHA are verified; guest bootability remains unproved.\n    This function has no cloud/seal/delete operations and returns a readonly artifact.\n    \"\"\"\n    need(callable(guard),'GUARD_REQUIRED');guard();validate_job(job,int(time.time()));validate_tools(tools)\n    work=Path(work);parent=stream.private_dir(work)\n    try:need(os.listdir(parent)==[],'PRODUCER_WORKSPACE_NOT_EMPTY')\n    finally:os.close(parent)\n    tmp=work/'tmp';t.make_dir(tmp,exclusive=True)\n    dest=work/('guest-%d.%s.zst'%(job['vmid'],'tar' if job['type']=='lxc' else 'vma'))\n    def free_guard():\n        v=os.statvfs(work);need(v.f_frsize*v.f_bavail>=LOCAL_RESERVE,'STAGING_RESERVE_REACHED_PARTIAL_RETAINED')\n    free_guard()\n    result=capture_archive(dump_argv(job,tools['vzdump']['path'],tmp,int(time.time())),dest,job['archive_cap_bytes'],seconds,free_guard)\n    guard();validate_tools(tools)\n    rc,_=t.bounded_process([tools['zstd']['path'],'--test','--quiet','--',str(dest)],dict(ENV),65536,min(seconds,3600))\n    need(rc==0,'ARCHIVE_COMPRESSION_TEST_FAILED_RETAINED')\n    n,h,identity=stream.hash_file(dest,job['archive_cap_bytes'])\n    need(n==result['size'] and h==result['sha256'] and list(identity)==result['identity'],'PRODUCER_READBACK_MISMATCH')\n    guard();validate_tools(tools)\n    return {**result,'path':str(dest),'vmid':job['vmid'],'type':job['type'],'node':job['node'],\n            'config_sha256':job['config_sha256'],'compression_test_ok':True,'restored':False,\n            'state':'PRODUCED_COMPRESSION_CHECKED_NOT_CLOUD_SEALED','local_payload_retained':True}\n","source_sha256":"d0659237df958ee02e24360721d2664177266e8a53bd5eb68d6655480e923f67","subscripts":["job['archive_cap_bytes']","job['config_sha256']","job['node']","job['type']","job['vmid']","result['identity']","result['sha256']","result['size']","tools['vzdump']","tools['vzdump']['path']","tools['zstd']","tools['zstd']['path']"]}
PVE270_PRODUCE_CONTRACT={"comparisons":["h == result['sha256']","job.get('config_sha256') == EXPECTED_CFG","job.get('type') == 'lxc'","job.get('vmid') == 110","list(identity) == result['identity']","n == result['size']","os.listdir(parent) == []","rc == 0","scratch is not None","v.f_frsize * v.f_bavail >= reserve"],"need_calls":[{"code":"GUARD_REQUIRED","line":2,"predicate":"callable(guard)"},{"code":"ONLY_ACCEPTED_CT110","line":3,"predicate":"job.get('vmid') == 110 and job.get('type') == 'lxc' and (job.get('config_sha256') == EXPECTED_CFG)"},{"code":"PRODUCER_WORKSPACE_NOT_EMPTY","line":6,"predicate":"os.listdir(parent) == []"},{"code":"ARCHIVE_COMPRESSION_TEST_FAILED_RETAINED","line":24,"predicate":"rc == 0"},{"code":"PRODUCER_READBACK_MISMATCH","line":26,"predicate":"n == result['size'] and h == result['sha256'] and (list(identity) == result['identity'])"},{"code":"reason","line":19,"predicate":"v.f_frsize * v.f_bavail >= reserve"}],"returns":[{"expr":"{**result, 'path': str(dest), 'vmid': 110, 'type': 'lxc', 'node': job['node'], 'config_sha256': job['config_sha256'], 'compression_test_ok': True, 'restored': False, 'state': 'PRODUCED_COMPRESSION_CHECKED_NOT_CLOUD_SEALED', 'local_payload_retained': True}","line":28}],"signature":"(job, work, tools, guard, *, mapping, seconds=14400, emit=<function <lambda> at 0x7c65069293a0>)","source":"def produce(job,work,tools,guard,*,mapping,seconds=14400,emit=lambda k,v:None):\n    need(callable(guard),'GUARD_REQUIRED');guard();old.validate_job(job,int(time.time()));old.validate_tools(tools)\n    need(job.get('vmid')==110 and job.get('type')=='lxc' and job.get('config_sha256')==EXPECTED_CFG,'ONLY_ACCEPTED_CT110')\n    uid,gid=validate_mapping(mapping)\n    work=Path(work);parent=stream.private_dir(work)\n    try:need(os.listdir(parent)==[],'PRODUCER_WORKSPACE_NOT_EMPTY')\n    finally:os.close(parent)\n    # Keep the old orchestration contract: this private placeholder stays empty.\n    t.make_dir(work/'tmp',exclusive=True)\n    scratch=None;complete=False\n    try:\n        scratch=MappedScratch(gid)\n        emit('PVE_TEMP_DIRECTORY',scratch.path);emit('PVE_TEMP_MODE','0710');emit('PVE_TEMP_GID',gid)\n        t.create(work/'PVE_TEMP.json',core.canonical_bytes({'path':str(scratch.path),'mode':'0710','uid':0,'gid':gid,'identity':list(scratch.identity)}))\n        scratch.probe(uid,work);emit('MAPPED_UID_ACCESS_AND_ISOLATION_PROBE','PASS')\n        def free_guard():\n            scratch.check()\n            for path,reserve,reason in ((work,old.LOCAL_RESERVE,'STAGING_RESERVE_REACHED_PARTIAL_RETAINED'),(scratch.path,256*1024**2,'PVE_TEMP_RESERVE_LOW')):\n                v=os.statvfs(path);need(v.f_frsize*v.f_bavail>=reserve,reason)\n        free_guard();guard();dest=work/'guest-110.tar.zst'\n        result=capture_archive(old.dump_argv(job,tools['vzdump']['path'],scratch.path,int(time.time())),dest,job['archive_cap_bytes'],seconds,free_guard)\n        guard();old.validate_tools(tools)\n        rc,_=t.bounded_process([tools['zstd']['path'],'--test','--quiet','--',str(dest)],dict(old.ENV),65536,min(seconds,3600))\n        need(rc==0,'ARCHIVE_COMPRESSION_TEST_FAILED_RETAINED')\n        n,h,identity=stream.hash_file(dest,job['archive_cap_bytes'])\n        need(n==result['size'] and h==result['sha256'] and list(identity)==result['identity'],'PRODUCER_READBACK_MISMATCH')\n        guard();old.validate_tools(tools);scratch.close(empty_only=True);complete=True;emit('PVE_TEMP_EMPTY_REMOVED','YES')\n        return {**result,'path':str(dest),'vmid':110,'type':'lxc','node':job['node'],'config_sha256':job['config_sha256'],\n                'compression_test_ok':True,'restored':False,'state':'PRODUCED_COMPRESSION_CHECKED_NOT_CLOUD_SEALED','local_payload_retained':True}\n    finally:\n        m=work/'PRODUCER_PROCESS.json'\n        if m.exists():\n            info=json.loads(t.read_secure(m,65536,private=True)[0])\n            emit('VZDUMP_PROCESS_STARTED','YES' if info.get('process_started') else 'NO')\n            emit('VZDUMP_PROCESS_RC',info.get('producer_rc'));emit('PRIVATE_STDERR_BYTES',info.get('stderr_bytes_retained'))\n        if scratch is not None and not complete:\n            emit('PVE_TEMP_RETAINED','YES');scratch.close(empty_only=False)\n","source_sha256":"8054e713b79b176819d990bb21c0430b909a11732850662638f60550fe574bdc","subscripts":["job['archive_cap_bytes']","job['config_sha256']","job['node']","result['identity']","result['sha256']","result['size']","t.read_secure(m, 65536, private=True)[0]","tools['vzdump']","tools['vzdump']['path']","tools['zstd']","tools['zstd']['path']"]}
PVE282_PRODUCE_CONTRACT={"comparisons":["h == result['sha256']","list(identity) == result['identity']","n == result['size']","os.listdir(parent) == []","rc == 0","scratch is not None","v.f_frsize * v.f_bavail >= reserve"],"need_calls":[{"code":"GUARD_REQUIRED","line":2,"predicate":"callable(guard)"},{"code":"PRODUCER_WORKSPACE_NOT_EMPTY","line":6,"predicate":"os.listdir(parent) == []"},{"code":"ARCHIVE_COMPRESSION_TEST_FAILED_RETAINED","line":24,"predicate":"rc == 0"},{"code":"PRODUCER_READBACK_MISMATCH","line":26,"predicate":"n == result['size'] and h == result['sha256'] and (list(identity) == result['identity'])"},{"code":"reason","line":19,"predicate":"v.f_frsize * v.f_bavail >= reserve"}],"returns":[{"expr":"{**result, 'path': str(dest), 'vmid': 112, 'type': 'lxc', 'node': job['node'], 'config_sha256': job['config_sha256'], 'compression_test_ok': True, 'restored': False, 'state': 'PRODUCED_COMPRESSION_CHECKED_NOT_CLOUD_SEALED', 'local_payload_retained': True}","line":28}],"signature":"(job, work, tools, guard, *, mapping, expected_config, seconds=14400, emit=<function <lambda> at 0x7c650692b880>)","source":"def produce(job,work,tools,guard,*,mapping,expected_config,seconds=14400,emit=lambda k,v:None):\n    need(callable(guard),'GUARD_REQUIRED');guard();old.validate_job(job,int(time.time()));old.validate_tools(tools)\n    require_target(job,expected_config)\n    uid,gid=validate_mapping(mapping)\n    work=Path(work);parent=stream.private_dir(work)\n    try:need(os.listdir(parent)==[],'PRODUCER_WORKSPACE_NOT_EMPTY')\n    finally:os.close(parent)\n    # Keep the old orchestration contract: this private placeholder stays empty.\n    t.make_dir(work/'tmp',exclusive=True)\n    scratch=None;complete=False\n    try:\n        scratch=MappedScratch(gid)\n        emit('PVE_TEMP_DIRECTORY',scratch.path);emit('PVE_TEMP_MODE','0710');emit('PVE_TEMP_GID',gid)\n        t.create(work/'PVE_TEMP.json',core.canonical_bytes({'path':str(scratch.path),'mode':'0710','uid':0,'gid':gid,'identity':list(scratch.identity)}))\n        scratch.probe(uid,work);emit('MAPPED_UID_ACCESS_AND_ISOLATION_PROBE','PASS')\n        def free_guard():\n            scratch.check()\n            for path,reserve,reason in ((work,old.LOCAL_RESERVE,'STAGING_RESERVE_REACHED_PARTIAL_RETAINED'),(scratch.path,256*1024**2,'PVE_TEMP_RESERVE_LOW')):\n                v=os.statvfs(path);need(v.f_frsize*v.f_bavail>=reserve,reason)\n        free_guard();guard();dest=work/'guest-112.tar.zst'\n        result=capture_archive(old.dump_argv(job,tools['vzdump']['path'],scratch.path,int(time.time())),dest,job['archive_cap_bytes'],seconds,free_guard)\n        guard();old.validate_tools(tools)\n        rc,_=t.bounded_process([tools['zstd']['path'],'--test','--quiet','--',str(dest)],dict(old.ENV),65536,min(seconds,3600))\n        need(rc==0,'ARCHIVE_COMPRESSION_TEST_FAILED_RETAINED')\n        n,h,identity=stream.hash_file(dest,job['archive_cap_bytes'])\n        need(n==result['size'] and h==result['sha256'] and list(identity)==result['identity'],'PRODUCER_READBACK_MISMATCH')\n        guard();old.validate_tools(tools);scratch.close(empty_only=True);complete=True;emit('PVE_TEMP_EMPTY_REMOVED','YES')\n        return {**result,'path':str(dest),'vmid':112,'type':'lxc','node':job['node'],'config_sha256':job['config_sha256'],\n                'compression_test_ok':True,'restored':False,'state':'PRODUCED_COMPRESSION_CHECKED_NOT_CLOUD_SEALED','local_payload_retained':True}\n    finally:\n        m=work/'PRODUCER_PROCESS.json'\n        if m.exists():\n            info=json.loads(t.read_secure(m,65536,private=True)[0])\n            emit('VZDUMP_PROCESS_STARTED','YES' if info.get('process_started') else 'NO')\n            emit('VZDUMP_PROCESS_RC',info.get('producer_rc'));emit('PRIVATE_STDERR_BYTES',info.get('stderr_bytes_retained'))\n        if scratch is not None and not complete:\n            emit('PVE_TEMP_RETAINED','YES');scratch.close(empty_only=False)\n","source_sha256":"a0d4189cc4f06d9ab9dbf34e619f93554f6386789c48d55b41d97294fa32f2e3","subscripts":["job['archive_cap_bytes']","job['config_sha256']","job['node']","result['identity']","result['sha256']","result['size']","t.read_secure(m, 65536, private=True)[0]","tools['vzdump']","tools['vzdump']['path']","tools['zstd']","tools['zstd']['path']"]}
TOOL_RELATED_TEST_SOURCE={"test_guest266.py":[{"kind":"ClassDef","line":61,"names":[],"source":"class Policy(unittest.TestCase):\n    def test_minimal_lxc_selected(self):self.assertEqual(g.choose(plan(),NOW)['vmid'],110)\n    def test_input_not_mutated(self):\n        z=plan();prior=copy.deepcopy(z);g.choose(z,NOW);self.assertEqual(z,prior)\n    def test_global_block(self):\n        z=plan();z['global_blockers']=['HOOK']\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_stale(self):\n        with self.assertRaises((t.Stop,p.PVEError)):g.choose(plan(),1301)\n    def test_future(self):\n        with self.assertRaises((t.Stop,p.PVEError)):g.choose(plan(),999)\n    def test_only_vm_not_selected(self):\n        z=plan();z['jobs']=z['jobs'][1:];z['local_guest_count']=1\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_blocked_not_selected(self):\n        z=plan();z['jobs'][0]['blockers']=['LOCK'];z['jobs'][0]['coverage_eligible']=False\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_compression_dependent_not_selected(self):\n        z=plan();z['jobs'][0]['compression_dependent']=True\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_external_mount_not_selected(self):\n        z=plan();z['jobs'][0]['disks'].append(dict(z['jobs'][0]['disks'][0],slot='mp0'))\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_unstable_plan(self):\n        z=plan();z['guest_membership_and_configs_stable']=False\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_duplicate_vmid_rejected(self):\n        z=plan();z['jobs'].append(copy.deepcopy(z['jobs'][0]));z['local_guest_count']=3\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_insufficient_storage_headroom(self):\n        z=plan();z['storages'][0]['free_bytes']=None\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_small_storage_headroom(self):\n        z=plan();z['storages'][0]['free_bytes']=7*G\n        with self.assertRaises(t.Stop):g.choose(z,NOW)\n    def test_changed_config_after_producer_rejected(self):\n        before=plan();after=copy.deepcopy(before);after['jobs'][0]['config_sha256']='0'*64\n        with self.assertRaises(t.Stop):g.same_state(before,after,110)\n    def test_changed_membership_rejected(self):\n        before=plan();after=copy.deepcopy(before);after['jobs']=after['jobs'][:1];after['local_guest_count']=1\n        with self.assertRaises(t.Stop):g.same_state(before,after,110)\n    def test_after_plan_timestamp_not_a_reason_to_fail(self):\n        before=plan();after=copy.deepcopy(before);after['observed_at']+=500;after['jobs'][0]['observed_at']+=500\n        g.same_state(before,after,110)\n    def test_changed_defaults_rejected(self):\n        before=plan();after=copy.deepcopy(before);after['global_defaults']['sha256']='1'*64\n        with self.assertRaises(t.Stop):g.same_state(before,after,110)\n    def test_running_task_blocks(self):\n        with self.assertRaises(t.Stop):g.no_active_tasks([{'status':'RUNNING','type':'vzdump','id':''}],110)\n    def test_finished_recent_task_accepted(self):g.no_active_tasks([{'status':'OK','type':'vzdump','endtime':999,'id':''}],110)\n    def test_console_is_ignored(self):g.no_active_tasks([{'status':'RUNNING','type':'vncshell','id':''}],110)\n    def test_unknown_task_on_target_blocks(self):\n        with self.assertRaises(t.Stop):g.no_active_tasks([{'status':'RUNNING','type':'new-operation','id':'110'}],110)\n    def test_truncated_task_page_rejected(self):\n        with self.assertRaises(t.Stop):g.no_active_tasks([{'status':'OK','endtime':1,'type':'x','id':''}]*1000,110)","source_sha256":"5671d58019db7f2b34b744ded6d888b087ea61bd7344918dd8a03c3a7e934d65"}],"test_producer282.py":[{"kind":"ClassDef","line":12,"names":[],"source":"class Gates(unittest.TestCase):\n def setUp(self): self.assertIsNotNone(f,'268R1 fix not implemented')\n def test_default_mapping(self):self.assertEqual(f.validate_mapping(MAP),(100000,100000))\n def test_root_mapping_refused(self):\n  d=copy.deepcopy(MAP);d['root_uid']=0\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def test_custom_mapping_refused(self):\n  d=copy.deepcopy(MAP);d['id_map'][0][2]=200000\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def test_bool_mapping_refused(self):\n  d=copy.deepcopy(MAP);d['root_gid']=True\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def test_extra_mapping_fields_refused(self):\n  d=copy.deepcopy(MAP);d['extra']='x'\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def gate(self,cfg=None,status=None,snaps=None):\n  return f.validate_ct(CFG if cfg is None else cfg, {'status':'running'} if status is None else status,\n                       [{'name':'current'}] if snaps is None else snaps, f.t.sha(f.core.canonical_bytes(CFG)))\n def test_clean_ct(self):self.assertTrue(self.gate())\n def test_stopped_ct(self):\n  with self.assertRaises(ValueError):self.gate(status={'status':'stopped'})\n def test_locked_ct(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'lock':'backup'})\n def test_custom_idmap(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'lxc.idmap':'u 0 100000 65536'})\n def test_old_vzdump_snapshot(self):\n  with self.assertRaises(ValueError):self.gate(snaps=[{'name':'current'},{'name':'vzdump'}])\n def test_bad_snapshot_schema(self):\n  with self.assertRaises(ValueError):self.gate(snaps=['vzdump'])\n def test_config_change(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'memory':999})\n def test_privileged_ct(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'unprivileged':0})","source_sha256":"66554808135edc1511e5ccf1d0ebdd3bbb2d1b052f70a5a05b23a088ec284b55"},{"kind":"ClassDef","line":133,"names":[],"source":"class ProducerFixture(unittest.TestCase):\n \"\"\"Actual subprocess + tar/zstd, synthetic data; never runs vzdump on a guest.\"\"\"\n def setUp(self):\n  self.assertIsNotNone(f,'268R1 fix not implemented');self.td=tempfile.TemporaryDirectory();self.root=Path(self.td.name)\n  self.work=self.root/'producer';self.work.mkdir(mode=0o700);self.events=[]\n def tearDown(self):\n  for k,v in self.events:\n   if k=='PVE_TEMP_DIRECTORY':\n    p=Path(v)\n    if p.exists():\n     for q in p.iterdir():\n      if q.is_dir():\n       import shutil;shutil.rmtree(q)\n      else:q.unlink()\n     p.rmdir()\n  self.td.cleanup()\n def fixture(self,fail=False,leftover=False):\n  import shutil\n  exe=self.root/'fake-vzdump'\n  code='''import os,sys,subprocess,shutil\nfrom pathlib import Path\nassert sys.argv[1]=='112'\np=Path(sys.argv[sys.argv.index('--tmpdir')+1])/'vzdumptmp_fixture_110'\np.mkdir();(p/'etc').mkdir();(p/'etc/vzdump').mkdir()\n(p/'etc/vzdump/pct.conf').write_bytes(b'NON_SECRET_SYNTHETIC_CT_CONFIG')\na=subprocess.run(['/usr/bin/tar','-cf','-','-C',str(p),'./etc/vzdump/pct.conf'],user=100000,group=100000,extra_groups=[],stdout=subprocess.PIPE,stderr=subprocess.PIPE)\nos.write(2,a.stderr)\nif a.returncode:raise SystemExit(a.returncode)\nz=subprocess.run(['/usr/bin/zstd','-q','-c'],input=a.stdout,stdout=subprocess.PIPE,stderr=subprocess.PIPE)\nos.write(1,z.stdout);os.write(2,b'fixture diagnostic\\\\n')\n'''\n  if not leftover:code+='shutil.rmtree(p)\\n'\n  code+='raise SystemExit(%s)\\n'%('23' if fail else 'z.returncode')\n  exe.write_text('#!'+sys.executable+'\\n'+code);exe.chmod(0o700)\n  tools={k:{'path':str(v),'sha256':hashlib.sha256(v.read_bytes()).hexdigest()} for k,v in {'vzdump':exe,'zstd':Path(shutil.which('zstd')).resolve()}.items()}\n  from test_pve265 import plan\n  job=plan(now=int(time.time()))['jobs'][0];job['vmid']=112;job['config_sha256']=f.t.sha(f.core.canonical_bytes(CFG))\n  return job,tools\n def produce(self,**kwargs):\n  job,tools=self.fixture(**kwargs)\n  return f.produce(job,self.work,tools,lambda:None,mapping=MAP,expected_config=job['config_sha256'],emit=lambda k,v:self.events.append((k,v)))\n def test_real_tar_sees_metadata_as_mapped_uid(self):\n  r=self.produce();self.assertEqual(r['producer_rc'],0);self.assertTrue(r['compression_test_ok']);self.assertEqual(list((self.work/'tmp').iterdir()),[])\n  self.assertIn(('MAPPED_UID_ACCESS_AND_ISOLATION_PROBE','PASS'),self.events)\n  self.assertIn(('PVE_TEMP_EMPTY_REMOVED','YES'),self.events)\n def test_producer_failed_archive_retained(self):\n  with self.assertRaisesRegex(ValueError,'PRODUCER_RC_FAILED'):self.produce(fail=True)\n  self.assertTrue((self.work/'guest-112.tar.zst').exists());self.assertIn(('VZDUMP_PROCESS_RC',23),self.events)\n  self.assertIn(('PVE_TEMP_RETAINED','YES'),self.events)\n def test_nonempty_pve_temp_blocks_success(self):\n  with self.assertRaisesRegex(ValueError,'TEMP_NONEMPTY_RETAINED'):self.produce(leftover=True)\n  self.assertTrue((self.work/'guest-112.tar.zst').exists())\n def test_fresh_archive_flows_through_existing_two_replica_engine(self):\n  import guest266 as guest\n  import test_guest266 as support\n  job,tools=self.fixture();work=self.root/'generation';work.mkdir(mode=0o700)\n  store=support.Store();now=int(time.time())\n  def quota(remote):return {'ok':True,'observed_at':now,'total':1000*1024**3,'used':0,'free':1000*1024**3-int(remote[1:])*1000}\n  store.quota=quota\n  def prod(j,p):return f.produce(j,p,tools,lambda:None,mapping=MAP,expected_config=job['config_sha256'],emit=lambda k,v:self.events.append((k,v)))\n  def crc(p):return subprocess.run(['/usr/bin/zstd','--test','--quiet',str(p)],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL).returncode==0\n  result=guest.run_flow(job,work,support.inventory(),store,support.Reader(store),prod,lambda:None,lambda:None,crc,'G268-fixture',lambda k,v:None,clock=lambda:now)\n  self.assertEqual(result['local_payload_files_remaining'],0);self.assertTrue(result['archive_cloud_restore_verified']);self.assertFalse(result['guest_restore_verified'])\n  self.assertTrue((work/'producer/producer.stderr.log').exists());self.assertFalse((work/'producer/guest-112.tar.zst').exists())","source_sha256":"a88818a2a77a33f99456fac73b8600d4a9a7c75767d6175b7a26c3ca375a8da6"}],"test_pve265.py":[{"kind":"ClassDef","line":24,"names":[],"source":"class Plans(unittest.TestCase):\n def job(self,kind='lxc',**extra):\n  c=cfg(kind);c.update(extra)\n  return plan(rows=[row(kind=kind)],configs={110:c})['jobs'][0]\n def test_rootfs_default_is_documented_inclusion(self):\n  j=self.job(); self.assertEqual(j['blockers'],[]);self.assertTrue(j['coverage_eligible']);self.assertEqual(j['disks'][0]['backup_setting'],'included_default_rootfs')\n def test_qemu_default_inclusion(self):self.assertEqual(self.job('qemu')['disks'][0]['backup_setting'],'included_default_qemu')\n def test_explicit_exclusion_blocks(self):self.assertIn('EXPLICITLY_EXCLUDED_DISK',self.job('qemu',scsi0='local-lvm:v,size=8G,backup=0')['blockers'])\n def test_mp_default_not_backed_up(self):self.assertIn('LXC_EXTRA_MOUNT_BACKUP_UNPROVEN',self.job(mp0='local-lvm:v,mp=/data,size=1G')['blockers'])\n def test_mp_explicit_inclusion(self):self.assertTrue(self.job(mp0='local-lvm:v,mp=/data,size=1G,backup=1')['coverage_eligible'])\n def test_bind_mount_blocks_even_backup1(self):self.assertIn('BIND_OR_DEVICE_MOUNT',self.job(mp0='/srv/data,mp=/data,backup=1')['blockers'])\n def test_missing_disk_size_blocks(self):self.assertIn('DISK_SIZE_UNKNOWN',self.job(rootfs='local-lvm:v')['blockers'])\n def test_no_disks_blocks(self):self.assertIn('NO_BACKUP_DISKS',plan(configs={110:{}})['jobs'][0]['blockers'])\n def test_hook_blocks(self):self.assertIn('GUEST_HOOK_PRESENT',self.job(hookscript='local:secret-hook')['blockers'])\n def test_lock_blocks(self):self.assertIn('GUEST_LOCK_PRESENT',self.job(lock='backup')['blockers'])\n def test_unused_disks_not_silently_lost(self):self.assertIn('UNUSED_VOLUMES_NOT_IN_DUMP',self.job(unused0='local-lvm:v')['blockers'])\n def test_missing_storage_blocks(self):self.assertIn('STORAGE_NOT_AVAILABLE',plan(storage=[])['jobs'][0]['blockers'])\n def test_lxc_unsupported_snapshot_blocks(self):\n  st=stores();st[0]['type']='dir';self.assertIn('LXC_SNAPSHOT_SUPPORT_UNPROVEN',plan(storage=st)['jobs'][0]['blockers'])\n def test_qemu_storage_need_not_snapshot(self):\n  st=stores();st[0]['type']='dir';self.assertTrue(plan(rows=[row(kind='qemu')],configs={110:cfg('qemu')},storage=st)['jobs'][0]['coverage_eligible'])\n def test_file_and_volume_prefixes(self):\n  self.assertTrue(self.job(rootfs='volume=local-lvm:v,size=8G')['coverage_eligible']);self.assertTrue(self.job('qemu',scsi0='file=local-lvm:v,size=8G')['coverage_eligible'])\n def test_hardware_passthrough_blocks(self):self.assertIn('EXTERNAL_DEVICE_OR_CUSTOM_CONFIG',self.job('qemu',hostpci0='0000:00:01')['blockers'])\n def test_qemu_custom_args_blocks(self):self.assertIn('EXTERNAL_DEVICE_OR_CUSTOM_CONFIG',self.job('qemu',args='-drive file=/secret')['blockers'])\n def test_lxc_raw_mount_blocks(self):self.assertIn('EXTERNAL_DEVICE_OR_CUSTOM_CONFIG',self.job(**{'lxc.mount.entry':'secret /data none bind 0 0'})['blockers'])\n def test_cdrom_explicitly_not_covered(self):\n  j=self.job('qemu',ide2='local:iso/foo.iso,media=cdrom');self.assertEqual(j['cdrom_count'],1);self.assertFalse(j['cdrom_media_included'])\n def test_snapshot_history_never_claimed(self):self.assertFalse(self.job()['snapshot_history_included'])\n def test_duplicate_vmid_fails(self):\n  with self.assertRaises(Exception):plan(rows=[row(),row()])\n def test_missing_guest_config_fails(self):\n  with self.assertRaises(p.PVEError):plan(configs={111:cfg()})\n def test_global_hook_blocks_no_value_export(self):\n  q=plan(defaults=b'script: /secret/hook\\n');self.assertIn('GLOBAL_DEFAULT_SCRIPT',q['global_blockers']);self.assertNotIn('/secret',str(q))\n def test_global_exclusions_block(self):self.assertIn('GLOBAL_DEFAULT_EXCLUDE_PATH',plan(defaults=b'exclude-path: /data\\n')['global_blockers'])\n def test_global_overridden_defaults_are_explicit(self):self.assertEqual(plan(defaults=b'mode: stop\\nremove: 1\\nstdexcludes: 1\\n')['global_blockers'],[])\n def test_unknown_global_default_blocks(self):self.assertIn('GLOBAL_DEFAULT_UNREVIEWED',plan(defaults=b'future-option: yes\\n')['global_blockers'])\n def test_malformed_default_blocks(self):self.assertIn('GLOBAL_DEFAULT_PARSE',plan(defaults=b'not valid\\n')['global_blockers'])\n def test_duplicate_default_blocks(self):self.assertIn('GLOBAL_DEFAULT_DUPLICATE',plan(defaults=b'mode: snapshot\\nmode: stop\\n')['global_blockers'])\n def test_capacity_never_uses_reported_maxdisk(self):\n  rr=row();rr['maxdisk']=1;j=plan(rows=[rr])['jobs'][0];self.assertEqual(j['configured_disk_bytes'],8*1024**3)\n def test_large_guest_capacity_dependency_is_reported(self):\n  j=plan(configs={110:{'rootfs':'local-lvm:v,size=128G'}},free=128*1024**3)['jobs'][0]\n  self.assertFalse(j['configured_full_cycle_fits']);self.assertTrue(j['compression_dependent']);self.assertLess(j['archive_cap_bytes'],j['configured_archive_bound_bytes'])\n def test_no_guest_is_dropped_if_blocked(self):\n  q=plan(rows=[row(),row(120,'qemu')],configs={110:cfg(),120:{'scsi0':'local-lvm:v,size=8G,backup=0'}})\n  self.assertEqual(len(q['jobs']),2);self.assertFalse(q['all_guest_coverage_eligible']);self.assertFalse(q['nodeset_backup_approved'])\n def test_stopped_and_template_not_auto_approved(self):\n  rr=row();rr['status']='stopped';self.assertIn('GUEST_NOT_RUNNING',plan(rows=[rr])['jobs'][0]['blockers'])\n  rr=row();rr['template']=True;self.assertIn('TEMPLATE_REQUIRES_SEPARATE_POLICY',plan(rows=[rr])['jobs'][0]['blockers'])\n def test_boolean_space_rejected(self):\n  with self.assertRaises(p.PVEError):plan(free=True)\n def test_plan_digest_binds_every_value(self):\n  q=plan();h=p.plan_hash(q);q['jobs'][0]['archive_cap_bytes']+=1;self.assertNotEqual(h,p.plan_hash(q))\n def test_argv_is_one_guest_snapshot_no_prune(self):\n  a=p.dump_argv(self.job(),'/usr/sbin/vzdump','/mnt/staging/tmp',1001)\n  self.assertEqual(a[:2],['/usr/sbin/vzdump','110']);self.assertEqual(a[a.index('--mode')+1],'snapshot');self.assertEqual(a[a.index('--remove')+1],'0')\n  self.assertEqual(a[a.index('--stdout')+1],'1');self.assertNotIn('--all',a);self.assertNotIn('--stop',a)\n def test_blocked_job_argv_rejected(self):\n  with self.assertRaises(p.PVEError):p.dump_argv(self.job(lock='backup'),'/usr/sbin/vzdump','/mnt/staging/tmp',1001)\n def test_stale_plan_rejected(self):\n  with self.assertRaises(p.PVEError):p.dump_argv(self.job(),'/usr/sbin/vzdump','/mnt/staging/tmp',1301)\n def test_future_plan_rejected(self):\n  with self.assertRaises(p.PVEError):p.dump_argv(self.job(),'/usr/sbin/vzdump','/mnt/staging/tmp',999)\n def test_path_injection_rejected(self):\n  with self.assertRaises(p.PVEError):p.dump_argv(self.job(),'/usr/sbin/vzdump','/mnt/../etc',1001)\n def test_bool_vmid_rejected(self):\n  j=self.job();j['vmid']=True\n  with self.assertRaises(p.PVEError):p.dump_argv(j,'/usr/sbin/vzdump','/mnt/staging/tmp',1001)","source_sha256":"a358f16a2761c4627d45036c24ae36f7d160988d844ceeda51e6faa9ee944b8d"},{"kind":"ClassDef","line":96,"names":[],"source":"class Processes(unittest.TestCase):\n def setUp(self):\n  self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name);self.root.chmod(0o700);self.dest=self.root/'payload'\n def tearDown(self):self.tmp.cleanup()\n def runp(self,code,cap=2*1024**2,seconds=2,check=lambda:None):\n  return p.capture_archive([sys.executable,'-c',code],self.dest,cap,seconds,check)\n def test_real_child_hash_and_readonly(self):\n  z=self.runp('import os;os.write(1,b\"abc\")');self.assertEqual(z['sha256'],t.sha(b'abc'));self.assertEqual(self.dest.stat().st_mode&0o777,0o400)\n def test_output_larger_than_chunk(self):\n  n=1024**2+17;z=self.runp('import sys;sys.stdout.buffer.write(b\"x\"*%d)'%n);self.assertEqual(z['size'],n);self.assertEqual(z['sha256'],t.sha(b'x'*n))\n def test_stderr_does_not_contaminate(self):\n  self.runp('import os;os.write(2,b\"secret\"*30000);os.write(1,b\"abc\")');self.assertEqual(self.dest.read_bytes(),b'abc')\n def test_nonzero_retains_partial(self):\n  with self.assertRaises(p.PVEError):self.runp('import os;os.write(1,b\"abc\");raise SystemExit(9)')\n  self.assertEqual(self.dest.read_bytes(),b'abc');self.assertEqual(self.dest.stat().st_mode&0o777,0o600)\n def test_empty_fails(self):\n  with self.assertRaises(p.PVEError):self.runp('pass')\n def test_size_cap_is_hard(self):\n  with self.assertRaises(p.PVEError):self.runp('print(\"abcdefghij\",end=\"\")',cap=3)\n  self.assertLessEqual(self.dest.stat().st_size,3)\n def test_timeout_retains(self):\n  with self.assertRaises(p.PVEError):self.runp('import time;time.sleep(1)',seconds=.05)\n  self.assertTrue(self.dest.exists())\n def test_no_overwrite(self):\n  self.dest.write_bytes(b'original')\n  with self.assertRaises(Exception):self.runp('print(\"x\")')\n  self.assertEqual(self.dest.read_bytes(),b'original')\n def test_symlink_refused(self):\n  target=self.root/'target';target.write_bytes(b'original');self.dest.symlink_to(target)\n  with self.assertRaises(Exception):self.runp('print(\"x\")')\n  self.assertEqual(target.read_bytes(),b'original')\n def test_guard_refusal_before_launch(self):\n  def no():raise p.PVEError('GUARD_TEST')\n  with self.assertRaises(p.PVEError):self.runp('print(\"x\")',check=no)\n  self.assertFalse(self.dest.exists())\n def test_guard_runs_during_stream(self):\n  calls=[];self.runp('print(\"abc\",end=\"\")',check=lambda:calls.append(1));self.assertGreaterEqual(len(calls),2)\n def test_producer_with_real_fake_executables(self):\n  exe=self.root/'vzdump';exe.write_text('#!'+sys.executable+'\\nimport os\\nos.write(1,b\"synthetic-compressed-payload\")\\n');exe.chmod(0o700)\n  val=self.root/'zstd';val.write_text('#!'+sys.executable+'\\nraise SystemExit(0)\\n');val.chmod(0o700)\n  j=plan(now=int(time.time()))['jobs'][0]\n  tools={k:{'path':str(v),'sha256':t.sha(v.read_bytes())} for k,v in [('vzdump',exe),('zstd',val)]}\n  work=self.root/'job';t.make_dir(work)\n  result=p.produce(j,work,tools,lambda:None,seconds=3)\n  self.assertTrue(result['compression_test_ok']);self.assertEqual(result['producer_rc'],0);self.assertFalse(result['restored']);self.assertEqual(stream.hash_file(Path(result['path']))[1],result['sha256'])\n def test_compression_reject_retains_archive(self):\n  exe=self.root/'vzdump';exe.write_text('#!'+sys.executable+'\\nprint(\"broken\",end=\"\")\\n');exe.chmod(0o700)\n  val=self.root/'zstd';val.write_text('#!'+sys.executable+'\\nraise SystemExit(1)\\n');val.chmod(0o700)\n  tools={k:{'path':str(v),'sha256':t.sha(v.read_bytes())} for k,v in [('vzdump',exe),('zstd',val)]}\n  work=self.root/'job';t.make_dir(work)\n  with self.assertRaises(p.PVEError):p.produce(plan(now=int(time.time()))['jobs'][0],work,tools,lambda:None,seconds=3)\n  self.assertTrue((work/'guest-110.tar.zst').exists())\n def test_changed_tool_rejected(self):\n  work=self.root/'job';t.make_dir(work);exe=self.root/'x';exe.write_bytes(b'X');exe.chmod(0o700)\n  tools={x:{'path':str(exe),'sha256':'0'*64} for x in ('vzdump','zstd')}\n  with self.assertRaises(p.PVEError):p.produce(plan(now=int(time.time()))['jobs'][0],work,tools,lambda:None,seconds=3)\n  self.assertFalse((work/'guest-110.tar.zst').exists())","source_sha256":"0f1f977dcdabb70d5c7330972848c0d7aae4a2a8199f29824fe1208f800b931e"}],"test_pve270.py":[{"kind":"ClassDef","line":12,"names":[],"source":"class Gates(unittest.TestCase):\n def setUp(self): self.assertIsNotNone(f,'268R1 fix not implemented')\n def test_default_mapping(self):self.assertEqual(f.validate_mapping(MAP),(100000,100000))\n def test_root_mapping_refused(self):\n  d=copy.deepcopy(MAP);d['root_uid']=0\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def test_custom_mapping_refused(self):\n  d=copy.deepcopy(MAP);d['id_map'][0][2]=200000\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def test_bool_mapping_refused(self):\n  d=copy.deepcopy(MAP);d['root_gid']=True\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def test_extra_mapping_fields_refused(self):\n  d=copy.deepcopy(MAP);d['extra']='x'\n  with self.assertRaises(ValueError):f.validate_mapping(d)\n def gate(self,cfg=None,status=None,snaps=None):\n  return f.validate_ct(CFG if cfg is None else cfg, {'status':'running'} if status is None else status,\n                       [{'name':'current'}] if snaps is None else snaps, f.t.sha(f.core.canonical_bytes(CFG)))\n def test_clean_ct(self):self.assertTrue(self.gate())\n def test_stopped_ct(self):\n  with self.assertRaises(ValueError):self.gate(status={'status':'stopped'})\n def test_locked_ct(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'lock':'backup'})\n def test_custom_idmap(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'lxc.idmap':'u 0 100000 65536'})\n def test_old_vzdump_snapshot(self):\n  with self.assertRaises(ValueError):self.gate(snaps=[{'name':'current'},{'name':'vzdump'}])\n def test_bad_snapshot_schema(self):\n  with self.assertRaises(ValueError):self.gate(snaps=['vzdump'])\n def test_config_change(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'memory':999})\n def test_privileged_ct(self):\n  with self.assertRaises(ValueError):self.gate(cfg={**CFG,'unprivileged':0})","source_sha256":"66554808135edc1511e5ccf1d0ebdd3bbb2d1b052f70a5a05b23a088ec284b55"},{"kind":"ClassDef","line":133,"names":[],"source":"class ProducerFixture(unittest.TestCase):\n \"\"\"Actual subprocess + tar/zstd, synthetic data; never runs vzdump on a guest.\"\"\"\n def setUp(self):\n  self.assertIsNotNone(f,'268R1 fix not implemented');self.td=tempfile.TemporaryDirectory();self.root=Path(self.td.name)\n  self.work=self.root/'producer';self.work.mkdir(mode=0o700);self.events=[]\n def tearDown(self):\n  for k,v in self.events:\n   if k=='PVE_TEMP_DIRECTORY':\n    p=Path(v)\n    if p.exists():\n     for q in p.iterdir():\n      if q.is_dir():\n       import shutil;shutil.rmtree(q)\n      else:q.unlink()\n     p.rmdir()\n  self.td.cleanup()\n def fixture(self,fail=False,leftover=False):\n  import shutil\n  exe=self.root/'fake-vzdump'\n  code='''import os,sys,subprocess,shutil\nfrom pathlib import Path\np=Path(sys.argv[sys.argv.index('--tmpdir')+1])/'vzdumptmp_fixture_110'\np.mkdir();(p/'etc').mkdir();(p/'etc/vzdump').mkdir()\n(p/'etc/vzdump/pct.conf').write_bytes(b'NON_SECRET_SYNTHETIC_CT_CONFIG')\na=subprocess.run(['/usr/bin/tar','-cf','-','-C',str(p),'./etc/vzdump/pct.conf'],user=100000,group=100000,extra_groups=[],stdout=subprocess.PIPE,stderr=subprocess.PIPE)\nos.write(2,a.stderr)\nif a.returncode:raise SystemExit(a.returncode)\nz=subprocess.run(['/usr/bin/zstd','-q','-c'],input=a.stdout,stdout=subprocess.PIPE,stderr=subprocess.PIPE)\nos.write(1,z.stdout);os.write(2,b'fixture diagnostic\\\\n')\n'''\n  if not leftover:code+='shutil.rmtree(p)\\n'\n  code+='raise SystemExit(%s)\\n'%('23' if fail else 'z.returncode')\n  exe.write_text('#!'+sys.executable+'\\n'+code);exe.chmod(0o700)\n  tools={k:{'path':str(v),'sha256':hashlib.sha256(v.read_bytes()).hexdigest()} for k,v in {'vzdump':exe,'zstd':Path(shutil.which('zstd')).resolve()}.items()}\n  from test_pve265 import plan\n  job=plan(now=int(time.time()))['jobs'][0];job['config_sha256']=f.EXPECTED_CFG\n  return job,tools\n def produce(self,**kwargs):\n  job,tools=self.fixture(**kwargs)\n  return f.produce(job,self.work,tools,lambda:None,mapping=MAP,emit=lambda k,v:self.events.append((k,v)))\n def test_real_tar_sees_metadata_as_mapped_uid(self):\n  r=self.produce();self.assertEqual(r['producer_rc'],0);self.assertTrue(r['compression_test_ok']);self.assertEqual(list((self.work/'tmp').iterdir()),[])\n  self.assertIn(('MAPPED_UID_ACCESS_AND_ISOLATION_PROBE','PASS'),self.events)\n  self.assertIn(('PVE_TEMP_EMPTY_REMOVED','YES'),self.events)\n def test_producer_failed_archive_retained(self):\n  with self.assertRaisesRegex(ValueError,'PRODUCER_RC_FAILED'):self.produce(fail=True)\n  self.assertTrue((self.work/'guest-110.tar.zst').exists());self.assertIn(('VZDUMP_PROCESS_RC',23),self.events)\n  self.assertIn(('PVE_TEMP_RETAINED','YES'),self.events)\n def test_nonempty_pve_temp_blocks_success(self):\n  with self.assertRaisesRegex(ValueError,'TEMP_NONEMPTY_RETAINED'):self.produce(leftover=True)\n  self.assertTrue((self.work/'guest-110.tar.zst').exists())\n def test_fresh_archive_flows_through_existing_two_replica_engine(self):\n  import guest266 as guest\n  import test_guest266 as support\n  job,tools=self.fixture();work=self.root/'generation';work.mkdir(mode=0o700)\n  store=support.Store();now=int(time.time())\n  def quota(remote):return {'ok':True,'observed_at':now,'total':1000*1024**3,'used':0,'free':1000*1024**3-int(remote[1:])*1000}\n  store.quota=quota\n  def prod(j,p):return f.produce(j,p,tools,lambda:None,mapping=MAP,emit=lambda k,v:self.events.append((k,v)))\n  def crc(p):return subprocess.run(['/usr/bin/zstd','--test','--quiet',str(p)],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL).returncode==0\n  result=guest.run_flow(job,work,support.inventory(),store,support.Reader(store),prod,lambda:None,lambda:None,crc,'G268-fixture',lambda k,v:None,clock=lambda:now)\n  self.assertEqual(result['local_payload_files_remaining'],0);self.assertTrue(result['archive_cloud_restore_verified']);self.assertFalse(result['guest_restore_verified'])\n  self.assertTrue((work/'producer/producer.stderr.log').exists());self.assertFalse((work/'producer/guest-110.tar.zst').exists())","source_sha256":"95832e4a60f84e22938b30d8c3752ae22ff31b459a8b67da3c7320e9af0b1829"}]}
VALIDATE_TOOLS_AND_PRODUCE_CALL_SITES=[{"calls":[{"call":"validate_tools(tools)","line":205},{"call":"validate_tools(tools)","line":215},{"call":"validate_tools(tools)","line":220}],"file":"pve265.py","sha256":"cd2f634414a8e38a3789015aeca5a9b47821f77d7d1fa61dcdbdc866172963be"},{"calls":[{"call":"old.validate_tools(tools)","line":293},{"call":"old.validate_tools(tools)","line":313},{"call":"old.validate_tools(tools)","line":318}],"file":"pve270.py","sha256":"97015277a4f3f8129c0fd11eb70c06293026366f1f8232a97feeed513a2dbcef"},{"calls":[{"call":"old.validate_tools(tools)","line":66},{"call":"old.validate_tools(tools)","line":86},{"call":"old.validate_tools(tools)","line":91}],"file":"pve282.py","sha256":"73c6cb7773eb8e82d1e7ed5ab4733b4afb88f74ae69c09125d0378d6e480e012"},{"calls":[{"call":"p.produce(j, work, tools, lambda: None, seconds=3)","line":139},{"call":"p.produce(plan(now=int(time.time()))['jobs'][0], work, tools, lambda: None, seconds=3)","line":146},{"call":"p.produce(plan(now=int(time.time()))['jobs'][0], work, tools, lambda: None, seconds=3)","line":151}],"file":"test_pve265.py","sha256":"16a9407d01bf73db687ef38684c59205fc2e6a9223cd9cee6ec6e8dde9d38707"},{"calls":[{"call":"f.produce(job, self.work, tools, lambda: None, mapping=MAP, emit=lambda k, v: self.events.append((k, v)))","line":172},{"call":"self.produce()","line":174},{"call":"self.produce(fail=True)","line":178},{"call":"self.produce(leftover=True)","line":182},{"call":"f.produce(j, p, tools, lambda: None, mapping=MAP, emit=lambda k, v: self.events.append((k, v)))","line":191}],"file":"test_pve270.py","sha256":"9fccd20b490613fdd5b14b4105cf0d9fd744f204e1ed1d6893e54c9f69bf6fb4"},{"calls":[{"call":"f.produce(job, self.work, tools, lambda: None, mapping=MAP, expected_config=job['config_sha256'], emit=lambda k, v: self.events.append((k, v)))","line":173},{"call":"self.produce()","line":175},{"call":"self.produce(fail=True)","line":179},{"call":"self.produce(leftover=True)","line":183},{"call":"f.produce(j, p, tools, lambda: None, mapping=MAP, expected_config=job['config_sha256'], emit=lambda k, v: self.events.append((k, v)))","line":192}],"file":"test_producer282.py","sha256":"a67d927cc428935791f9eecc3acf571d089f80761162f4589b7d684fba6b8af7"},{"calls":[{"call":"self.produce(j, p)","line":153},{"call":"self.produce(j, p)","line":157},{"call":"self.produce(j, p)","line":161},{"call":"self.produce(j, p)","line":165}],"file":"test_guest266.py","sha256":"460b950924a475ce0a28dc759512e7dbe2323ae2cc42f7ef4c6181710a2a13da"}]
RCA_423_STAGE=PRE_MUTATION_TOOL_DESCRIPTOR_PREFLIGHT
RCA_423_HISTORICAL_LITERAL_VZDUMP_SCHEMAS=ZERO
RCA_423_VALIDATE_TOOLS_REJECTED_FALLBACK=TOOLS_SCHEMA
PRODUCTION_POST_RCA_UNCHANGED=PASS
CANDIDATE_POST_RCA_UNCHANGED=PASS
STATUS=OK
PHASE=COMPLETE
REASON=NONE
NEXT_GATE=CT112_REAL_PRODUCER_ARCHIVE_1_TOOLS_CONTRACT_FIX
CHANGES_MADE=false
MUTATION_OUTCOME=NO_MUTATION
BACKUP_EXECUTIONS=0
VZDUMP_EXECUTIONS=0
CLOUD_WRITES=0
CLOUD_DELETIONS=0
SYSTEMD_MUTATIONS=0
SCHEDULER_CUTOVER=NO
HOMELAB_RESULT_CONTRACT={"changes_made":false,"command_id":"SUPPORT-260915-INFRA-DR-BACKUPV2-PVE-LXC-VALIDATE-TOOLS-CONTRACT-RCA-424R1","rollback_restored":null,"rollback_started":false,"status":"OK","version":1}

OUTPUT_END
CHAT_OUTPUT_END
