import test from 'node:test'
import assert from 'node:assert/strict'
import { spawnSync } from 'node:child_process'
import { fileURLToPath } from 'node:url'

test('one container resource-confirm fault is scoped, one-shot, expiring and transparent (offline)', () => {
  const source=String.raw`
import hashlib,importlib.util,json,sys,unittest,concurrent.futures
spec=importlib.util.spec_from_file_location('fault',sys.argv[1]);p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p)
class Tests(unittest.TestCase):
 def setUp(self):
  self.now=[10.0];self.calls=[];self.logs=[]
  self.config={'environment':'TEST','containerBookingId':'7','warehouseId':'30','batchPrefixHash':hashlib.sha256(b'"7"').hexdigest(),
    'controlToken':'private-control','internalToken':'private-token','lastlegIps':['172.20.0.10'],'upstream':'http://warehouse.invalid:9305'}
  def forward(*args):self.calls.append(args);return 200,{'X-Upstream':'original'},b'original-response'
  self.router=p.Router(self.config,forward=forward,clock=lambda:self.now[0],record=self.logs.append)
  self.body={'warehouseId':30,'batchKey':self.router.prefix+'generation-1','commandKey':'original-command'}
  self.headers={'X-Internal-Service-Token':'private-token'}
 def send(self,body=None,path=p.CONFIRM_PATH,method='POST',headers=None,peer='172.20.0.10'):
  raw=json.dumps(self.body if body is None else body).encode()
  return self.router.dispatch(method,path,self.headers if headers is None else headers,raw,peer)
 def test_default_pass_and_original_payload_preserved(self):
  self.assertEqual(self.send(),(200,{'X-Upstream':'original'},b'original-response'))
  self.assertEqual(json.loads(self.calls[0][4]),self.body)
  self.assertEqual(self.calls[0][3],self.headers)
 def test_only_exact_confirm_body_caller_and_identity_can_consume_fault(self):
  self.router.set_mode('private-control','CONFIRM_ONCE',60)
  for kwargs in [dict(path=p.CONFIRM_PATH+'?other=1'),dict(path='/internal/v1/warehouse-pallet-batches/ensure-reserved'),
    dict(method='GET'),dict(body={**self.body,'warehouseId':31}),dict(body={**self.body,'batchKey':'another-container'}),
    dict(body=[]),dict(body={**self.body,'commandKey':''}),dict(peer='172.20.0.11'),dict(headers={}),
    dict(headers={'X-Internal-Service-Token':'wrong'})]:self.assertEqual(self.send(**kwargs)[0],200)
  self.assertEqual(self.router.status()['mode'],'CONFIRM_ONCE')
  self.assertEqual(self.send()[0],503);self.assertEqual(self.send()[0],200)
  self.assertEqual(self.router.status()['faulted'],1)
 def test_one_shot_is_atomic_under_concurrent_requests(self):
  self.router.set_mode('private-control','CONFIRM_ONCE',60)
  with concurrent.futures.ThreadPoolExecutor(max_workers=4) as e:codes=list(e.map(lambda _:self.send()[0],range(8)))
  self.assertEqual(codes.count(503),1);self.assertEqual(codes.count(200),7)
 def test_ttl_expiry_and_malformed_control_disarm(self):
  self.router.set_mode('private-control','CONFIRM_ONCE',2);self.now[0]=12
  self.assertEqual(self.send()[0],200)
  for mode,ttl in [('OTHER',1),('CONFIRM_ONCE',0),('CONFIRM_ONCE',61),('CONFIRM_ONCE',True)]:
   with self.assertRaises(ValueError):self.router.set_mode('private-control',mode,ttl)
   self.assertEqual(self.router.status()['mode'],'PASS')
  with self.assertRaises(PermissionError):self.router.set_mode('wrong','CONFIRM_ONCE',1)
 def test_production_or_mismatched_container_scope_rejected(self):
  for override in [{'environment':'PROD'},{'containerBookingId':'8'},{'warehouseId':'0'},{'lastlegIps':[]}]:
   with self.assertRaises(ValueError):p.Router({**self.config,**override})
 def test_malformed_json_is_forwarded_unchanged_and_never_consumes_fault(self):
  self.router.set_mode('private-control','CONFIRM_ONCE',60)
  raw=b'invalid-json'
  self.assertEqual(self.router.dispatch('POST',p.CONFIRM_PATH,self.headers,raw,'172.20.0.10')[0],200)
  self.assertEqual(self.calls[0][4],raw);self.assertEqual(self.router.status()['mode'],'CONFIRM_ONCE')
 def test_evidence_redacts_credentials_and_raw_body(self):
  self.router.set_mode('private-control','CONFIRM_ONCE',60);self.send()
  evidence=json.dumps(self.logs)
  for secret in ['private-control','private-token','original-command','generation-1']:self.assertNotIn(secret,evidence)
  self.assertTrue(self.logs[0]['internalTokenValidated']);self.assertFalse(self.logs[0]['forwarded'])
result=unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(Tests))
sys.exit(0 if result.wasSuccessful() else 1)
`
  const r=spawnSync(process.env.PYTHON||'python',['-c',source,fileURLToPath(new URL('./inbound-plan-resource-fault-proxy.py',import.meta.url))],
    {encoding:'utf8',env:{...process.env,PYTHONDONTWRITEBYTECODE:'1'}})
  assert.equal(r.status,0,r.stderr||r.error?.message)
})