rootandClaude Opus 4.7
85c0071fd8
来源：hgdjnet-server 9bfbafe10d417deea959bd53117bc88a10e60a83 项目身份 HKDJ / JHA-216，独立官网仓库，Jenkins 守卫键 hkdjnet 测试目录 /server/hkdj_net，App/Web API 30331/30332 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
55 lines
4.4 KiB
Python
55 lines
4.4 KiB
Python
import json,sys,os,shlex,hashlib,datetime
|
|
from pathlib import Path
|
|
from pymongo import MongoClient
|
|
from bson.int64 import Int64
|
|
import requests
|
|
p=json.load(sys.stdin); env=p['env']; plan=p['plan']
|
|
assert p['plan_hash']=='3f56441d0c496a9817c5d2f269084eab8450a2d73bcaec9f45a415cbd1751541'
|
|
assert plan['host']=='172.104.181.99' and {i['database'] for i in plan['mongodb']}=={'hkdj_net','hkdj_net_log'}
|
|
client=MongoClient(p['admin_uri'],serverSelectionTimeoutMS=10000)
|
|
roles=[{'role':'readWrite','db':n} for n in ('hkdj_net','hkdj_net_log')]
|
|
users=client.hkdj_net.command('usersInfo',env['HKDJNET_TEST_MONGODB_USERNAME'])['users']
|
|
if users:
|
|
assert sorted(users[0]['roles'],key=lambda x:x['db'])==sorted(roles,key=lambda x:x['db'])
|
|
else:
|
|
assert not client.hkdj_net.list_collection_names() and not client.hkdj_net_log.list_collection_names()
|
|
client.hkdj_net.command('createUser',env['HKDJNET_TEST_MONGODB_USERNAME'],pwd=env['HKDJNET_TEST_MONGODB_PASSWORD'],roles=roles)
|
|
for index in plan['mongodb']:
|
|
client[index['database']][index['collection']].create_index([tuple(k) for k in index['keys']],**index['options'])
|
|
for index in plan['mongodb']:
|
|
actual=list(client[index['database']][index['collection']].list_indexes())
|
|
def matches(x):
|
|
if list(x['key'].items())==[tuple(k) for k in index['keys']]: return True
|
|
# MongoDB stores text fields as _fts/_ftsx plus weights, not the requested keys.
|
|
text_fields={k for k,v in index['keys'] if v=='text'}
|
|
return bool(text_fields) and x.get('name')==index['options']['name'] and x['key'].get('_fts')=='text' and set(x.get('weights',{}))==text_fields and all(v==1 for v in x['weights'].values())
|
|
matching=[x for x in actual if matches(x)]
|
|
assert matching,(index['collection'],index['options']['name'])
|
|
for k,v in index['options'].items():
|
|
if k!='name': assert matching[0].get(k)==v,(index['collection'],k)
|
|
s=requests.Session();s.trust_env=False;s.auth=(env['HKDJNET_ES_USERNAME'],env['HKDJNET_ES_PASSWORD'])
|
|
def canonical_mapping(value):
|
|
if isinstance(value,list): return [canonical_mapping(v) for v in value]
|
|
if not isinstance(value,dict): return value
|
|
return {k:canonical_mapping(v) for k,v in value.items() if not (k=='search_analyzer' and v==value.get('analyzer'))}
|
|
for name,mapping in plan['elasticsearch'].items():
|
|
url=env['HKDJNET_ES_URL']+'/'+name; r=s.get(url,timeout=10)
|
|
if r.status_code==404:
|
|
r=s.put(url,json=mapping,timeout=30);r.raise_for_status()
|
|
r=s.get(url,timeout=10);r.raise_for_status()
|
|
assert canonical_mapping(r.json()[name]['mappings'])==canonical_mapping(mapping['mappings'])
|
|
assert r.json()[name]['settings']['index']['analysis']==mapping['settings']['analysis']
|
|
now=datetime.datetime.now(datetime.timezone.utc);db=client.hkdj_net
|
|
# Dedicated test identity, never copied from the source project.
|
|
db.mw_role.update_one({'_id':Int64(1)},{'$setOnInsert':{'name':'超级管理员','desc':'花魁NET专用测试角色','isEnb':True,'apiMenus':[],'router':[],'createdAt':now,'updatedAt':now}},upsert=True)
|
|
db.mw_acc.update_one({'_id':Int64(1)},{'$setOnInsert':{'acc':env['HKDJNET_TEST_ADMIN_USERNAME'],'pwd':hashlib.md5(env['HKDJNET_TEST_ADMIN_PASSWORD'].encode()).hexdigest(),'userName':'花魁NET测试','roleId':Int64(1),'roleName':'超级管理员','isEnb':True,'isBase32':True,'base32':env['HKDJNET_TEST_ADMIN_TOTP_SECRET'],'limitIP':['127.0.0.1'],'createdAt':now,'updatedAt':now}},upsert=True)
|
|
db.sys_cfg.update_one({'_id':Int64(1)},{'$setOnInsert':{'webTitle':'花魁短剧web站点','webSiteName':'花魁短剧web站点','webKeyworld':[],'webDescription':'花魁短剧web站点测试','webAddr':env['HKDJNET_TEST_H5_URL'],'foreverUrlAddr':env['HKDJNET_TEST_H5_URL']}},upsert=True)
|
|
for business in ('MwAcc','MwRole'):
|
|
db.generate_id.update_one({'businessId':business},{'$setOnInsert':{'maxId':Int64(1),'step':Int64(10),'desc':'测试初始化','createAt':now,'updateAt':now}},upsert=True)
|
|
root=Path('/server/hkdj_net'); root.mkdir(parents=True,exist_ok=True)
|
|
runtime={k:v for k,v in env.items() if k.startswith('HKDJNET_') and not k.startswith('HKDJNET_TEST_')}
|
|
text=''.join(k+'='+shlex.quote(v)+'\n' for k,v in sorted(runtime.items()))
|
|
tmp=root/'.env.pending';tmp.write_text(text);os.chmod(tmp,0o600);os.replace(tmp,root/'.env')
|
|
(root/'scripts').mkdir(exist_ok=True);(root/'scripts/probe-health.py').write_text(p['probe'])
|
|
print(json.dumps({'mongo_requests_verified':len(plan['mongodb']),'es_indices_verified':list(plan['elasticsearch']),'runtime_env_mode':'600','seeded_admin_id':1,'seeded_role_id':1}))
|