222 lines
17 KiB
JavaScript
222 lines
17 KiB
JavaScript
import assert from 'node:assert/strict'
|
|
import test from 'node:test'
|
|
import { createHash } from 'node:crypto'
|
|
import { spawnSync } from 'node:child_process'
|
|
import { createTestFaultExecutor, createTestUnloadFaultExecutor, assertTestUnloadFaultProof, renderTestFaultAction } from './test-event-fault-executor.mjs'
|
|
|
|
const scope = { marker: 'TEST-ROOT', customerId: '7', loadingPlanId: '99', containerBookingId: '66' }
|
|
const environment = { CHONGCHONG_ENV: 'test', JENKINS_URL: 'https://jenkins.example.invalid/',
|
|
JENKINS_USER: 'offline', JENKINS_API_TOKEN: 'offline-secret', FBA_CORE_JENKINS_AGENT: 'java-test-agent' }
|
|
function requestedAction(options) {
|
|
const script = options.body.get('script')
|
|
const chunks = [...script.matchAll(/'([A-Za-z0-9+/=]+)'/g)].map(match => match[1]).join('')
|
|
const groovy = Buffer.from(chunks, 'base64').toString('utf8')
|
|
const argument = groovy.match(/'UTF-8'\),'([A-Za-z0-9+/=]+)'\]\)/)?.[1]
|
|
return JSON.parse(Buffer.from(argument, 'base64').toString('utf8'))
|
|
}
|
|
function transport(calls, failActivate = false) {
|
|
let recovered = false
|
|
return async (_url, options) => {
|
|
assert.equal(options.redirect, 'error')
|
|
const { action } = requestedAction(options)
|
|
calls.push(action)
|
|
if (action === 'recover') recovered = true
|
|
const data = action === 'proxy-ready' ? { ready: true } : action === 'health' ? { healthy: true }
|
|
: action === 'status' ? { producer: { publishStatus: recovered ? 'PUBLISHED' : 'DEAD', failedAttempts: 12 } }
|
|
: action === 'proof' ? { ...scope, restored: false } : {}
|
|
const result = action === 'activate' && failActivate ? { ok: false, error: 'DOCKER_409' } : { ok: true, data }
|
|
return { ok: true, text: async () => 'TEST_EVENT_JSON=' + JSON.stringify(result) }
|
|
}
|
|
}
|
|
test('scope and actions are bounded and renderer never embeds caller credentials', () => {
|
|
const rendered = renderTestFaultAction('status', scope, environment.FBA_CORE_JENKINS_AGENT)
|
|
assert.ok(!rendered.includes(environment.JENKINS_API_TOKEN))
|
|
for (const change of [{ ...scope, customerId: '7;DROP' }, { ...scope, marker: '../other' }]) {
|
|
assert.throws(() => renderTestFaultAction('status', change, environment.FBA_CORE_JENKINS_AGENT))
|
|
}
|
|
assert.throws(() => renderTestFaultAction('arbitrary', scope, 'java-test-agent'))
|
|
assert.throws(() => createTestFaultExecutor({ ...environment, CHONGCHONG_ENV: 'prod' }))
|
|
})
|
|
|
|
const unloadScope = { marker: 'TEST-ROOT', customerId: '7', containerBookingId: '66' }
|
|
const originalImageId = 'sha256:' + 'a'.repeat(64), originalFingerprint = 'b'.repeat(64)
|
|
function unloadProof(nanoseconds = false) {
|
|
const payload = { actualEmptyReportedAt: null, containerNo: 'OFFLINE66', sourceContainerBookingId: '66',
|
|
unloadingCompletedAt: nanoseconds ? '2026-10-02T01:00:00.136827515Z' : '2026-10-02T01:00:00Z', unloadingPhotoUploadedAt: '2026-10-02T00:59:00Z' }
|
|
const payloadHash = 'sha256:' + createHash('sha256').update(JSON.stringify(payload)).digest('hex')
|
|
const eventId = '00000000-0000-4000-8000-000000000001', entryId = '123'
|
|
const envelope = { eventId, eventType: 'OverseasContainerUnloaded', eventVersion: 1, tenantId: '0',
|
|
aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', aggregateVersion: 3, payloadHash, payload }
|
|
const row = { entryId, eventId, payloadHash, envelope, eventType: envelope.eventType, customerId: '0',
|
|
aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', eventKey: 'OverseasContainerUnloaded:66', publishStatus: 'PUBLISHED' }
|
|
const container = { id: '66', containerNo: 'OFFLINE66', version: '9', emptyReportSourceVersion: '3',
|
|
unloadingCompletedAt: nanoseconds ? '2026-10-02 01:00:00.137' : '2026-10-02 01:00:00', unloadingPhotoUploadedAt: '2026-10-02 00:59:00',
|
|
estimatedEmptyReportedAt: nanoseconds ? '2026-10-05 01:00:00.137' : '2026-10-05 01:00:00', actualEmptyReportedAt: null }
|
|
return JSON.parse(JSON.stringify({ ...unloadScope, readOnly: true, producer: row,
|
|
beforeRetry: { ...row, publishStatus: 'DEAD', failedAttempts: 12, container }, afterRetry: { ...row, container },
|
|
consumer: { count: 1, status: 'PROCESSED', eventId, payloadHash, eventType: envelope.eventType, aggregateId: '66', payload },
|
|
attempts: Array.from({ length: 13 }, (_, i) => ({ attemptId: `attempt-${i}`, entryId, eventId, serviceName: 'lastleg', outcome: i < 11 ? 'RETRY' : i === 11 ? 'DEAD' : 'DUPLICATE' })),
|
|
proxyRecords: Array.from({ length: 13 }, (_, i) => ({ eventId, payloadHash, attempt: i + 1, matched: true,
|
|
outcome: i ? 'DUPLICATE' : 'APPLIED', errorCode: i < 12 ? 'ACK_LOSS_INJECTED' : null })),
|
|
retryRequestId: '00000000-0000-4000-8000-000000000002', retryAuditCount: 1,
|
|
retryAudit: { requestId: '00000000-0000-4000-8000-000000000002', entryId, eventId, payloadHash, previousFailures: 12 } }))
|
|
}
|
|
function unloadTransport(calls, proof = unloadProof(), finish = {}) {
|
|
let recovered = false
|
|
return async (_url, options) => {
|
|
const config = requestedAction(options), { action } = config
|
|
assert.equal(config.direction, 'UNLOAD')
|
|
assert.deepEqual(config.scope, unloadScope)
|
|
calls.push(action)
|
|
if (action === 'recover') recovered = true
|
|
const data = action === 'prepare' ? { originalImageId, originalFingerprint }
|
|
: action === 'proxy-ready' ? { ready: true } : action === 'health' ? { healthy: true }
|
|
: action === 'status' ? { producer: { publishStatus: recovered ? 'PUBLISHED' : 'DEAD', failedAttempts: 12 } }
|
|
: action === 'proof' ? proof : action === 'finish' ? { restored: true, healthy: true, proxyStopped: true,
|
|
producerName: 'firstleg-test-lastleg', originalImageId, currentImageId: originalImageId,
|
|
originalFingerprint, currentFingerprint: originalFingerprint, ...finish } : {}
|
|
return { ok: true, text: async () => 'TEST_EVENT_JSON=' + JSON.stringify({ ok: true, data }) }
|
|
}
|
|
}
|
|
test('unload factory pins exactly three scope fields and verifies restoration before success', async () => {
|
|
const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls))
|
|
await assert.rejects(executor.prepare({ ...unloadScope, eventType: 'Other' }), /Invalid/)
|
|
await executor.prepare(unloadScope)
|
|
await assert.rejects(executor.prepare({ ...unloadScope, containerBookingId: '67' }), /scope drift/)
|
|
await assert.rejects(executor.readProof({ ...unloadScope, eventType: 'Other' }), /scope drift/)
|
|
const proof = await executor.readProof(unloadScope)
|
|
assert.equal(proof.restored, true)
|
|
assert.equal(proof.restoration.currentImageId, originalImageId)
|
|
await executor.cleanup()
|
|
assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover', 'status', 'proof', 'restore', 'health', 'finish'])
|
|
})
|
|
test('unload proof rejects changed original event, fabricated ACK, nonunique audit and changed business facts', () => {
|
|
assertTestUnloadFaultProof(unloadProof(), unloadScope)
|
|
for (const mutate of [
|
|
p => { p.producer.customerId = '7' }, p => { p.producer.eventKey = 'other' },
|
|
p => { p.producer.envelope.tenantId = '7' }, p => { p.producer.envelope.payload.containerNo = 'CHANGED' },
|
|
p => { p.beforeRetry.entryId = '124' }, p => { p.beforeRetry.envelope.aggregateVersion = 4 },
|
|
p => { p.beforeRetry.failedAttempts = 11 }, p => { p.afterRetry.eventId = 'OTHER' },
|
|
p => { p.consumer.count = 2 }, p => { p.consumer.status = 'RECEIVED' }, p => { p.consumer.payload = {} },
|
|
p => { p.attempts.pop() }, p => { p.attempts[0].serviceName = 'firstleg' },
|
|
p => { p.proxyRecords[0].outcome = 'DUPLICATE' }, p => { p.proxyRecords[0].errorCode = null },
|
|
p => { p.proxyRecords[12].attempt = 1 }, p => { p.retryAuditCount = 2 },
|
|
p => { p.retryAudit.requestId = 'OTHER' }, p => { p.afterRetry.container.version = '10' },
|
|
p => { p.beforeRetry.container.estimatedEmptyReportedAt = p.afterRetry.container.estimatedEmptyReportedAt = '2026-10-04 01:00:00' },
|
|
]) {
|
|
const proof = unloadProof(); mutate(proof)
|
|
assert.throws(() => assertTestUnloadFaultProof(proof, unloadScope), /Invalid unload fault proof/)
|
|
}
|
|
})
|
|
|
|
test('unload timestamps follow DATETIME(3) rounding rather than allowing time drift', () => {
|
|
const proof=unloadProof(true)
|
|
assertTestUnloadFaultProof(proof,unloadScope)
|
|
for(const row of [proof.beforeRetry,proof.afterRetry]) {
|
|
row.container.unloadingCompletedAt='2026-10-02 01:00:00.138'
|
|
row.container.estimatedEmptyReportedAt='2026-10-05 01:00:00.138'
|
|
}
|
|
assert.throws(()=>assertTestUnloadFaultProof(proof,unloadScope),/CONTAINER_FACTS/)
|
|
})
|
|
test('invalid unload proof still restores and incomplete restoration never becomes success', async () => {
|
|
const invalid = unloadProof(); invalid.consumer.count = 0
|
|
const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls, invalid))
|
|
await executor.prepare(unloadScope)
|
|
await assert.rejects(executor.readProof(unloadScope), /INBOX/)
|
|
assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish'])
|
|
for (const change of [{ restored: false }, { healthy: false }, { proxyStopped: false },
|
|
{ currentImageId: 'sha256:' + 'c'.repeat(64) }, { currentFingerprint: 'c'.repeat(64) }]) {
|
|
const failed = createTestUnloadFaultExecutor(environment, unloadTransport([], unloadProof(), change))
|
|
await failed.prepare(unloadScope)
|
|
await assert.rejects(failed.readProof(unloadScope), /restoration evidence mismatch/)
|
|
}
|
|
})
|
|
test('Python reverse SQL routing, scope isolation and full configuration recovery fail closed offline', () => {
|
|
const result = spawnSync('python', ['-c', String.raw`
|
|
import ast, copy, importlib.util, io, pathlib, sys
|
|
from unittest.mock import patch
|
|
sys.dont_write_bytecode = True
|
|
path = pathlib.Path('scripts/test-event-fault-runtime.py')
|
|
ast.parse(path.read_text(encoding='utf-8'))
|
|
spec = importlib.util.spec_from_file_location('fault_runtime', path)
|
|
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
|
|
scope = {'marker':'TEST-ROOT','customerId':'7','containerBookingId':'66'}
|
|
assert m.target_setting({'direction':'UNLOAD'}) == ('FIRSTLEG_SERVICE_URL','http://firstleg:9202')
|
|
assert m.scope_directory(scope, 'UNLOAD') != m.scope_directory(dict(scope, loadingPlanId='99'))
|
|
try: m.scope_directory(dict(scope, eventType='Other'), 'UNLOAD'); assert False
|
|
except RuntimeError as error: assert str(error) == 'INVALID_SCOPE'
|
|
queries=[]
|
|
with patch.object(m, 'inspect', side_effect=lambda name: name), patch.object(m, 'read_sql', side_effect=lambda name, sql: queries.append((name, sql)) or []):
|
|
assert m.producer(scope, 'UNLOAD') is None
|
|
assert queries[0][0] == 'firstleg-test-lastleg'
|
|
assert "customer_id='0'" in queries[0][1] and "event_key='OverseasContainerUnloaded:66'" in queries[0][1]
|
|
assert "JSON_EXTRACT(payload_json,'$')" in queries[0][1] and "publishStatus',status" in queries[0][1]
|
|
row={'entryId':'123','eventId':'00000000-0000-4000-8000-000000000001','payloadHash':'sha256:'+'a'*64,'publishStatus':'DEAD','failedAttempts':12,'envelope':{'eventType':'OverseasContainerUnloaded'}}
|
|
retry_state={'direction':'UNLOAD','operationsRequestId':'00000000-0000-4000-8000-000000000002'}
|
|
retry_container={'Config':{'Env':['EVENT_OPERATIONS_TOKEN=offline']},'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3'}}}}
|
|
with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value=row),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'private_write'),patch.object(m,'inspect',return_value=retry_container),patch.object(m.urllib.request,'urlopen',return_value=io.BytesIO(b'{"code":"200"}')) as http:
|
|
assert m.recover(scope,pathlib.Path('/offline'),retry_state)['retrySubmitted']
|
|
assert http.call_args.args[0].full_url == 'http://172.20.0.3:9303/internal/lastleg/event-operations/123/retry'
|
|
assert retry_state['beforeRetry']['container']=={'version':'9'}
|
|
assert retry_state['beforeRetry']['envelope']==row['envelope']
|
|
row['publishStatus']='PUBLISHED'
|
|
queries=[]
|
|
with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value={'records':[]}),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'inspect',side_effect=lambda name:name),patch.object(m,'read_sql',side_effect=lambda name,sql: queries.append((name,sql)) or []),patch.object(m,'private_write') as save:
|
|
proof=m.proof(scope,retry_state)
|
|
save.assert_called_once_with(m.scope_directory(scope,'UNLOAD')/'proof.json',proof)
|
|
assert proof['retryAuditCount']==0 and proof['retryRequestId']==retry_state['operationsRequestId']
|
|
assert [item[0] for item in queries]==['firstleg-test-lastleg','firstleg-test-lastleg','firstleg-test-business']
|
|
assert "WHERE entry_id='123' AND event_type='OverseasContainerUnloaded'" in queries[1][1]
|
|
assert 'fl_container_event_inbox' in queries[2][1]
|
|
container={'Name':'/firstleg-test-lastleg','Id':'container123456789','Image':'sha256:'+'a'*64,
|
|
'Config':{'Image':'lastleg:test','Env':['FIRSTLEG_SERVICE_URL=http://firstleg:9202','A=1'],'Labels':{'com.docker.compose.project':'firstleg-test'}},
|
|
'HostConfig':{'Binds':['source:/data:rw'],'PortBindings':{'9303/tcp':[{'HostPort':'9303'}]}},
|
|
'Mounts':[{'Type':'bind','Source':'/source','Destination':'/data','RW':True}],
|
|
'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3','Aliases':['lastleg-service','container123']}}},
|
|
'State':{'Health':{'Status':'healthy'}}}
|
|
fingerprint=m.fingerprint(container)
|
|
multi=copy.deepcopy(container)
|
|
multi['NetworkSettings']['Networks']['firstleg-test-network']['MacAddress']='02:42:ac:15:00:04'
|
|
multi['NetworkSettings']['Networks']['second-test-network']={'IPAddress':'172.22.0.7','MacAddress':'02:42:ac:16:00:07','Aliases':['lastleg']}
|
|
legacy=copy.deepcopy(multi); legacy['Config']['MacAddress']='02:42:ac:16:00:07'
|
|
assert m.fingerprint(legacy)==m.fingerprint(multi)
|
|
changed=copy.deepcopy(legacy); changed['Config']['MacAddress']='02:42:ac:99:00:01'
|
|
assert m.fingerprint(changed)!=m.fingerprint(multi)
|
|
changed=copy.deepcopy(legacy); changed['NetworkSettings']['Networks']['second-test-network']['MacAddress']='02:42:ac:16:00:08'
|
|
assert m.fingerprint(changed)!=m.fingerprint(multi)
|
|
state={'direction':'UNLOAD','restored':False,'image':container['Config']['Image'],'originalImageId':container['Image'],
|
|
'originalFingerprint':fingerprint,'originalBody':m.recreation(container,m.env_of(container)),'proxyPid':123,'controlToken':'offline'}
|
|
for mutate in [lambda c:c['HostConfig'].update(Binds=['other:/data:rw']),lambda c:c['NetworkSettings']['Networks']['firstleg-test-network'].update(IPAddress='172.20.0.4'),lambda c:c.update(Image='sha256:'+'b'*64),lambda c:c['State']['Health'].update(Status='unhealthy')]:
|
|
changed=copy.deepcopy(container); mutate(changed)
|
|
with patch.object(m,'inspect',return_value=changed), patch.object(m,'private_write') as write:
|
|
try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False
|
|
except RuntimeError as error: assert str(error).startswith('RESTORED_')
|
|
write.assert_not_called()
|
|
with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=False),patch.object(m,'private_write'):
|
|
restored=m.finish(pathlib.Path('/offline'),copy.deepcopy(state))
|
|
assert restored['proxyStopped'] and restored['currentFingerprint']==fingerprint and restored['originalImageId']==restored['currentImageId']
|
|
with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=True),patch.object(m.Path,'read_bytes',return_value=b'python3\x00'+str(pathlib.Path('/offline')/'proxy.py').encode()+b'\x00'),patch.object(m.os,'kill'),patch.object(m.time,'sleep'),patch.object(m,'private_write') as write:
|
|
try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False
|
|
except RuntimeError as error: assert str(error)=='PROXY_NOT_STOPPED'
|
|
write.assert_not_called()
|
|
`], { encoding: 'utf8', cwd: new URL('..', import.meta.url), timeout: 10000 })
|
|
assert.equal(result.status, 0, result.stderr || result.error?.message)
|
|
})
|
|
test('only original scoped event is recovered and restoration precedes a successful proof', async () => {
|
|
const calls = [], executor = createTestFaultExecutor(environment, transport(calls))
|
|
await executor.prepare(scope)
|
|
await executor.prepare({ ...scope, loadingPlanId: '100' })
|
|
const proof = await executor.readProof({ containerBookingId: '66', loadingPlanId: '99', customerId: '7', marker: 'TEST-ROOT' })
|
|
assert.equal(proof.restored, true)
|
|
await executor.cleanup()
|
|
assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover',
|
|
'status', 'proof', 'restore', 'health', 'finish'])
|
|
})
|
|
test('failed activation retains task ownership for explicit restoration, never retries activation', async () => {
|
|
const calls = [], executor = createTestFaultExecutor(environment, transport(calls, true))
|
|
await assert.rejects(executor.prepare(scope), /DOCKER_409/)
|
|
await executor.cleanup()
|
|
assert.equal(calls.filter(name => name === 'activate').length, 1)
|
|
assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish'])
|
|
})
|