66 lines
4.6 KiB
JavaScript
66 lines
4.6 KiB
JavaScript
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)
|
|
})
|