fix(test): align order dependency fault preflight with Nacos v3

This commit is contained in:
2026-10-03 06:21:57 +08:00
parent bde1cd532e
commit c55bd7f4b6
4 changed files with 57 additions and 14 deletions
@@ -13,6 +13,21 @@ function fixture() {
}
const env={CHONGCHONG_ENV:'test',JENKINS_URL:'https://offline.invalid/',JENKINS_USER:'offline',JENKINS_API_TOKEN:'never-print-this',FBA_CORE_JENKINS_AGENT:'test-agent'}
test('runtime classifies dependency HTTP stages without exposing tokens, URLs or response bodies',()=>{
const script=String.raw`
import importlib.util,sys,urllib.error
spec=importlib.util.spec_from_file_location('diagnostic_runtime',sys.argv[1]);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
for path,expected in [('/nacos/v3/auth/user/login','NACOS_LOGIN_404'),('/nacos/v3/admin/cs/config','NACOS_CONFIG_404'),('/internal/finance/wallet','DEPENDENCY_HTTP_404')]:
error=urllib.error.HTTPError('http://private.invalid'+path+'?accessToken=secret',404,'private response secret',{},None)
assert m.safe_error_code(error)==expected
assert m.safe_error_code(urllib.error.URLError('http://private.invalid?secret'))=='DEPENDENCY_TRANSPORT_FAILED'
assert m.safe_error_code(RuntimeError('KNOWN_FAILURE'))=='KNOWN_FAILURE'
assert m.safe_error_code(ValueError('secret in response'))=='ORDER_DEPENDENCY_ACTION_FAILED'
`
const result=spawnSync(process.env.PYTHON||'python',['-c',script,fileURLToPath(new URL('./order-submit-dependency-fault-runtime.py',import.meta.url))],{encoding:'utf8',env:{...process.env,PYTHONDONTWRITEBYTECODE:'1'}})
assert.equal(result.status,0,result.stderr||result.error?.message)
})
test('renderer binds TEST scope and emits no Jenkins credential; ambiguous remote mutation is never retried',async()=>{
const {plan}=fixture(),scope=scopeFor(plan)
assert.ok(!renderOrderFaultAction('status',scope,env.FBA_CORE_JENKINS_AGENT).includes(env.JENKINS_API_TOKEN))
@@ -167,7 +182,7 @@ class RuntimeTests(unittest.TestCase):
f,r,s,saved,controls=self.fixture()
f.current['Config']['Env']+=['NACOS_SERVER_ADDR=host.docker.internal:8848','NACOS_USERNAME=offline-user',
'NACOS_PASSWORD=offline-password','NACOS_GROUP=BAIYUE','NACOS_NAMESPACE=test-only']
routing={'nacos':{'baseUrl':'http://127.0.0.1:8848','loginPath':'/nacos/v1/auth/users/login','configPath':'/nacos/v1/cs/configs'}}
routing={'nacos':{'baseUrl':'http://127.0.0.1:8848','loginPath':'/nacos/v3/auth/user/login','configPath':'/nacos/v3/admin/cs/config'}}
calls=[]
class Response:
status=200
@@ -177,12 +192,19 @@ class RuntimeTests(unittest.TestCase):
def read(self):return self.value
def read(req,timeout):
calls.append(req)
return Response(b'{"accessToken":"offline-token"}' if not isinstance(req,str) else b'spring:\n test: only\n')
if req.data is not None:return Response(b'{"accessToken":"offline-token"}')
query=m.urllib.parse.parse_qs(m.urllib.parse.urlsplit(req.full_url).query)
self.assertEqual(req.get_method(),'GET');self.assertEqual(req.get_header('Authorization'),'Bearer offline-token')
self.assertNotIn('offline-token',req.full_url)
self.assertEqual(query['namespaceId'],['test-only']);self.assertEqual(query['groupName'],['BAIYUE'])
return Response(json.dumps({'code':0,'data':{'dataId':query['dataId'][0],'namespaceId':'test-only','groupName':'BAIYUE','content':'spring:\n test: only\n'}}).encode())
with patch.object(m,'http_open',side_effect=read):
value=m.Runtime.config_fingerprint(r,f.current,routing)
self.assertEqual(len(value),64);self.assertEqual(len(calls),3)
self.assertIn(b'password=offline-password',calls[0].data)
self.assertTrue(all('accessToken=offline-token' in url for url in calls[1:]))
self.assertTrue(all('/nacos/v3/admin/cs/config?' in req.full_url for req in calls[1:]))
with patch.object(m,'http_open',return_value=Response(b'{"accessToken":"offline-token","code":0,"data":{"content":"secret config","dataId":"foreign"}}')):
with self.assertRaisesRegex(RuntimeError,'NACOS_SOURCE_IDENTITY_MISMATCH'):m.Runtime.config_fingerprint(r,f.current,routing)
routing['nacos']['baseUrl']='http://other.invalid:8848'
with self.assertRaisesRegex(RuntimeError,'EXPLICIT_NACOS'):m.Runtime.config_fingerprint(r,f.current,routing)
self.assertEqual(len(calls),3)