|
|
@@ -0,0 +1,188 @@
|
|
|
+import {test,before,after} from 'node:test';
|
|
|
+import assert from 'node:assert/strict';
|
|
|
+import {readFileSync} from 'node:fs';
|
|
|
+import {mkdtemp,rm,readdir,mkdir,writeFile} from 'node:fs/promises';
|
|
|
+import {tmpdir} from 'node:os';
|
|
|
+import {join} from 'node:path';
|
|
|
+import vm from 'node:vm';
|
|
|
+import {PGlite} from '@electric-sql/pglite';
|
|
|
+import {buildDirectBundle} from '../lib/legacy-direct-bundle.mjs';
|
|
|
+import {specs} from '../lib/legacy-direct-specs.mjs';
|
|
|
+import {legacyAddonFields} from '../lib/legacy-direct-projection.mjs';
|
|
|
+import {runDirectSync,changesFor,scheduledBusinessDate,pruneCompletedCaches} from '../run-nightly-direct-sync.mjs';
|
|
|
+const db=new PGlite();let context,runId;const owner='test-owner-0001';
|
|
|
+const sql=async(s,p=[])=>(await db.query(s,p)).rows;
|
|
|
+const Psql={query:sql,any:sql,one:async(s,p)=>(await sql(s,p))[0],oneOrNone:async(s,p)=>(await sql(s,p))[0]||null,none:async(s,p)=>{await sql(s,p);}};
|
|
|
+Psql.tx=async f=>{await db.exec('BEGIN');try{const r=await f(Psql);await db.exec('COMMIT');return r;}catch(e){await db.exec('ROLLBACK');throw e;}};
|
|
|
+async function call(params,role='super-admin',company='co'){
|
|
|
+ let code=200,body;const response={status:n=>{code=n;return response;},json:j=>{body=j;return j;}};
|
|
|
+ await context.handler({user:{id:'operator',fetch:async()=>{},get:k=>({adminRoleKey:role,company:{id:company}}[k])},body:{params:{runId,owner,...params}}},response);
|
|
|
+ return {code,...body};
|
|
|
+}
|
|
|
+before(async()=>{
|
|
|
+ await db.exec(readFileSync(new URL('../sql/legacy-direct-sync.sql',import.meta.url),'utf8'));
|
|
|
+ await db.exec('CREATE TABLE "CommonModel" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,"generalId" numeric,"itemId" numeric,"modelId" numeric,"createdAt" timestamptz,"updatedAt" timestamptz,status numeric,"teachingManaged" boolean,title text)');
|
|
|
+ for(const [m,s] of Object.entries(specs)){
|
|
|
+ const fields=legacyAddonFields[m]||(m==='52'?['sy','yb','lj','yp']:['content','synopsis','mp4','ico','author','source']);
|
|
|
+ await db.exec('CREATE TABLE "'+s.className+'" ("objectId" text PRIMARY KEY,company text,id numeric,"sourceKey" text,"createdAt" timestamptz,"updatedAt" timestamptz,"teachingManaged" boolean,'+fields.map(f=>'"'+f+'" text').join(',')+')');
|
|
|
+ }
|
|
|
+ await db.exec('ALTER TABLE "MemoryPracticeRecord" ADD "rewardTaskKey" text, ADD "rewardDuplicate" boolean');
|
|
|
+ const rewardSource=readFileSync(new URL('../sql/review-rewards.sql',import.meta.url),'utf8');await db.exec(rewardSource.match(/CREATE OR REPLACE FUNCTION xiaoshu_review_key[\s\S]*?\$\$;/)[0]);
|
|
|
+ await db.exec('CREATE TABLE "ReviewReward" ("objectId" text,company text,"taskId" text,amount numeric,status text)');
|
|
|
+ await db.exec('CREATE TABLE "LegacySyncConflict" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,dataset text,"legacyKey" text,"conflictType" text,message text,resolved boolean,resolution text,"createdAt" timestamptz,"updatedAt" timestamptz)');
|
|
|
+ const columns=await sql("SELECT table_name,column_name FROM information_schema.columns WHERE table_schema='public'");
|
|
|
+ context={Psql,Parse:{Config:{get:async()=>({get:k=>({legacyScheduleApiUrl:'https://legacy.invalid/api',legacyScheduleApiId:'test',legacyScheduleApiKey:'test'}[k])})}},fetch:async()=>({ok:true,json:async()=>({retcode:0,result:[{GeneralID:101,kcid:'5',xxcs:1}],page:{itemCount:1,pageCount:1}})}),AbortSignal,Date,console};
|
|
|
+ vm.createContext(context);vm.runInContext(buildDirectBundle(columns,[],'co'),context);
|
|
|
+ const r=await call({action:'claim',proposedRunId:'test-run-00001'});assert.equal(r.success,true,JSON.stringify(r));runId=r.data.runId;
|
|
|
+});
|
|
|
+after(()=>db.close());
|
|
|
+test('非超级管理员、其他帐套与另一执行者不能访问或写入',async()=>{
|
|
|
+ assert.equal((await call({action:'status'},'ops-manager')).code,403);
|
|
|
+ assert.equal((await call({action:'status'},'super-admin','other')).code,403);
|
|
|
+ assert.match((await call({action:'claim',owner:'another-owner-0001',proposedRunId:'another-run-00001'})).message,/another_sync_running/);
|
|
|
+ assert.match((await call({action:'read',modelId:53,ids:[],owner:'wrong-owner-0001'})).message,/lease/);
|
|
|
+});
|
|
|
+let original;
|
|
|
+test('主附表原子新增、批次重试复用结果，不允许同一批次编号换内容',async()=>{
|
|
|
+ const row={generalId:101,common:{objectId:'common101',status:99,title:'旧记录'},addon:{objectId:'addon101',id:101,kcid:'5',xxcs:'1'}};
|
|
|
+ const first=await call({action:'append',modelId:53,batchKey:'append-101',rows:[row]});assert.equal(first.success,true);assert.equal(first.data.createdCommon,1);original=first.data.postimages[0];
|
|
|
+ const again=await call({action:'append',modelId:53,batchKey:'append-101',rows:[row]});assert.deepEqual(JSON.parse(JSON.stringify(first.data)),JSON.parse(JSON.stringify(again.data)));
|
|
|
+ assert.equal((await sql('SELECT count(*)::int n FROM "PracticeRecord"'))[0].n,1);
|
|
|
+ assert.match((await call({action:'append',modelId:53,batchKey:'append-101',rows:[{...row,generalId:102}]})).message,/batch_key_reused/);
|
|
|
+ const bad={generalId:102,common:{objectId:'common102',unknown:'bad'},addon:{objectId:'addon102',id:102,kcid:'5'}};
|
|
|
+ assert.equal((await call({action:'append',modelId:53,batchKey:'bad-102',rows:[bad]})).success,false);
|
|
|
+ assert.equal((await sql('SELECT count(*)::int n FROM "PracticeRecord" WHERE id=102'))[0].n,0);
|
|
|
+});
|
|
|
+test('已保存投影可刷新，修改后的记录保留；微秒级版本变化也拒绝覆盖',async()=>{
|
|
|
+ const patch=()=>({generalId:101,commonId:original.common.objectId,addonId:original.addon.objectId,commonVersion:original.common.updatedAt,addonVersion:original.addon.updatedAt,values:{xxcs:'2'}});
|
|
|
+ let r=await call({action:'patch',modelId:53,batchKey:'patch-101',rows:[patch()]});assert.equal(r.data.updated,1);original=r.data.postimages[0];
|
|
|
+ await sql('UPDATE "PracticeRecord" SET xxcs=\'新系统修改\' WHERE id=101');
|
|
|
+ r=await call({action:'patch',modelId:53,batchKey:'preserve-101',rows:[patch()]});assert.equal(r.data.updated,0);assert.equal(r.data.conflicts.length,1);
|
|
|
+ await sql('UPDATE "PracticeRecord" SET xxcs=\'2\',"updatedAt"=\'2026-10-12T10:00:00.000002Z\' WHERE id=101');
|
|
|
+ const stale={...patch(),addonVersion:'2026-10-12T10:00:00.000001+00:00'};
|
|
|
+ r=await call({action:'patch',modelId:53,batchKey:'microsecond-101',rows:[stale]});assert.equal(r.data.updated,0);
|
|
|
+});
|
|
|
+test('服务端捕获真实源编号并重新核对投影，缺主附表时不能生成对账凭据',async()=>{
|
|
|
+ assert.equal((await call({action:'source',modelId:53,page:1})).success,true);
|
|
|
+ const evidence=await call({action:'evidence',modelId:53,conflicts:1});assert.equal(evidence.success,true);assert.equal(evidence.data.sourceChecksum,evidence.data.targetChecksum);
|
|
|
+ await sql('DELETE FROM "PracticeRecord" WHERE id=101');
|
|
|
+ assert.match((await call({action:'evidence',modelId:53})).message,/projection_reconciliation_failed/);
|
|
|
+ assert.match((await call({action:'finish',summary:{datasets:[]}})).message,/reconciliation_incomplete/);
|
|
|
+});
|
|
|
+test('冲突记录重复保存不增加重复行',async()=>{
|
|
|
+ for(let i=0;i<2;i++)assert.equal((await call({action:'conflicts',rows:[{modelId:53,generalId:101,fields:['xxcs']}]})).success,true);
|
|
|
+ assert.equal((await sql('SELECT count(*)::int n FROM "LegacySyncConflict"'))[0].n,1);
|
|
|
+});
|
|
|
+test('字段差异按业务类型规范化，旧空评分不抹掉新评分',()=>{
|
|
|
+ assert.deepEqual(changesFor(53,{xxcs:'1'},{xxcs:1},['xxcs']),{});
|
|
|
+ assert.deepEqual(changesFor(59,{pf:''},{pf:5},['pf']),{});
|
|
|
+ assert.deepEqual(changesFor(60,{FXZT:'1'},{fxzt:'0'},['fxzt']),{fxzt:'1'});
|
|
|
+});
|
|
|
+test('每日运行入口只使用直接接口、按依赖顺序并写私有报告，未对账不报成功',async()=>{
|
|
|
+ const root=await mkdtemp(join(tmpdir(),'direct-sync-test-')),realFetch=globalThis.fetch,order=[];
|
|
|
+ try{
|
|
|
+ globalThis.fetch=async(url,options)=>{assert.ok(url.endsWith('/xiaoshu/sync/direct-v1'));const p=JSON.parse(options.body).params;let data;
|
|
|
+ if(p.action==='status')data={source:'old-api/WXAPP',bridgeUsed:false};
|
|
|
+ else if(p.action==='claim')data={runId:'mock-run-000001',businessDate:'2026-10-12'};
|
|
|
+ else if(p.action==='supplement')data={accounts:{missing:0},nodes:{missing:0}};
|
|
|
+ else if(p.action==='source'){order.push(p.modelId);data={items:[],page:{itemCount:0,pageCount:1}};}
|
|
|
+ else if(p.action==='read')data=[];
|
|
|
+ else if(p.action==='evidence')data={sourceChecksum:'d41d8cd98f00b204e9800998ecf8427e',targetChecksum:'d41d8cd98f00b204e9800998ecf8427e',reconciled:true};
|
|
|
+ else if(p.action==='finish')data=p.summary;
|
|
|
+ return {ok:true,json:async()=>({success:true,data})};
|
|
|
+ };
|
|
|
+ const r=await runDirectSync({XIAOSHU_SYNC_REPORT_DIR:root,XIAOSHU_SYNC_OPERATOR_TOKEN:'test'});assert.equal(r.status,'completed');assert.ok(order.indexOf(56)<order.indexOf(60));assert.ok(order.indexOf(54)<order.indexOf(59));
|
|
|
+ assert.ok((await readdir(root)).some(f=>f.endsWith('.json')));assert.ok(!(await readdir(root)).includes('direct-nightly.lock'));
|
|
|
+ }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
|
|
|
+});
|
|
|
+
|
|
|
+
|
|
|
+test('有新系统计薪凭据的复习不回退、不重算工资',async()=>{
|
|
|
+ const row={generalId:301,common:{objectId:'memory-common',status:99},addon:{objectId:'memory-addon',id:301,yhid:'20',xxjlid:'33',plid:'30',kywrq:'2026-10-12',fxzt:'0'}};
|
|
|
+ const inserted=await call({action:'append',modelId:60,batchKey:'memory-301',rows:[row]});assert.equal(inserted.success,true,inserted.message);const t=inserted.data.postimages[0];
|
|
|
+ await sql('INSERT INTO "ReviewReward" VALUES(\'earned-reward\',\'co\',\'memory-addon\',1,\'earned\')');
|
|
|
+ const r=await call({action:'patch',modelId:60,batchKey:'memory-protected',rows:[{generalId:301,commonId:t.common.objectId,addonId:t.addon.objectId,commonVersion:t.common.updatedAt,addonVersion:t.addon.updatedAt,values:{fxzt:'1'}}]});
|
|
|
+ assert.equal(r.success,true,r.message);assert.equal(r.data.updated,0);assert.equal(r.data.conflicts.length,1);
|
|
|
+ assert.equal((await sql('SELECT amount FROM "ReviewReward" WHERE "objectId"=\'earned-reward\''))[0].amount,'1');
|
|
|
+ assert.equal((await sql('SELECT fxzt FROM "MemoryPracticeRecord" WHERE id=301'))[0].fxzt,'0');
|
|
|
+});
|
|
|
+test('北京时间23点前补昨天、23点后执行今天，跨UTC日期仍正确',()=>{
|
|
|
+ assert.equal(scheduledBusinessDate(new Date('2026-10-12T14:59:00Z')),'2026-10-11');
|
|
|
+ assert.equal(scheduledBusinessDate(new Date('2026-10-12T15:00:00Z')),'2026-10-12');
|
|
|
+ assert.equal(scheduledBusinessDate(new Date('2026-10-12T16:01:00Z')),'2026-10-12');
|
|
|
+});
|
|
|
+
|
|
|
+
|
|
|
+test('新增账号与栏目幂等补齐，已有账户余额和目录名称保持不变',async()=>{
|
|
|
+ await db.exec('CREATE TABLE "_User" ("objectId" text PRIMARY KEY,company text,username text UNIQUE,"legacyUserId" numeric,"legacyUserData" jsonb);CREATE TABLE "Node" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,"nodeId" numeric,"parentId" numeric,"nodeName" text,"createdAt" timestamptz,"updatedAt" timestamptz)');
|
|
|
+ await sql('INSERT INTO "_User" VALUES(\'existing-user\',\'co\',\'existing\',20,\'{"Purse":99}\')');
|
|
|
+ await sql('INSERT INTO "Node" ("objectId",company,"nodeId","nodeName") VALUES(\'existing-node\',\'co\',8,\'新系统目录\')');
|
|
|
+ const previous=context.fetch;let passwords=0;
|
|
|
+ context.Parse.Object={createWithoutData:(className,id)=>({id,className})};
|
|
|
+ context.Parse.User=class{constructor(){this.fields={};}set(k,v){this.fields[k]=v;}setUsername(v){this.fields.username=v;}setPassword(v){assert.ok(v.length>60);passwords++;}async save(){const f=this.fields;assert.equal(f.identityType,'member');assert.equal(f.adminRoleKey,undefined);assert.equal(f.password,undefined);await sql('INSERT INTO "_User" VALUES($1,$2,$3,$4,$5::jsonb)',['new-user',f.company.id,f.username,f.legacyUserId,JSON.stringify(f.legacyUserData)]);}};
|
|
|
+ context.fetch=async url=>{const action=new URL(url).searchParams.get('action');return {ok:true,json:async()=>({retcode:0,page:{pageCount:1},result:action==='user_list'?[{UserID:20,UserName:'existing',GroupID:1,Purse:1},{UserID:4000,UserName:'new-member',GroupID:1,Purse:10,Password:'ignored-hash'}]:[{NodeID:8,NodeName:'旧目录'},{NodeID:9,ParentID:8,NodeName:'新目录'}]})};};
|
|
|
+ try{let r=await call({action:'supplement'});assert.equal(r.success,true,r.message);assert.equal(r.data.accounts.created,1);assert.equal(r.data.nodes.created,1);
|
|
|
+ r=await call({action:'supplement'});assert.equal(r.success,true,r.message);assert.equal(r.data.accounts.created,0);assert.equal(r.data.nodes.created,0);assert.equal(passwords,1);
|
|
|
+ assert.equal((await sql('SELECT "legacyUserData" FROM "_User" WHERE "legacyUserId"=20'))[0].legacyUserData.Purse,99);
|
|
|
+ assert.equal((await sql('SELECT "nodeName" FROM "Node" WHERE "nodeId"=8'))[0].nodeName,'新系统目录');
|
|
|
+ }finally{context.fetch=previous;}
|
|
|
+});
|
|
|
+
|
|
|
+
|
|
|
+test('未取得服务器对账证据时不完成运行；已退出进程的目录锁可恢复',async()=>{
|
|
|
+ const root=await mkdtemp(join(tmpdir(),'direct-sync-negative-')),realFetch=globalThis.fetch;let finished=false;
|
|
|
+ try{
|
|
|
+ await mkdir(join(root,'direct-nightly.lock'));
|
|
|
+ await writeFile(join(root,'direct-nightly.lock/owner.json'),JSON.stringify({pid:2147483647}));
|
|
|
+ globalThis.fetch=async(url,options)=>{const p=JSON.parse(options.body).params;let data;
|
|
|
+ if(p.action==='status')data={source:'old-api/WXAPP',bridgeUsed:false};
|
|
|
+ else if(p.action==='claim')data={runId:'failed-run-000001',businessDate:'2026-10-12'};
|
|
|
+ else if(p.action==='supplement')data={accounts:{missing:0},nodes:{missing:0}};
|
|
|
+ else if(p.action==='source')data={items:[],page:{itemCount:0,pageCount:1}};
|
|
|
+ else if(p.action==='read')data=[];
|
|
|
+ else if(p.action==='evidence')data={sourceChecksum:'d41d8cd98f00b204e9800998ecf8427e',reconciled:false};
|
|
|
+ else if(p.action==='finish'){finished=true;data=p.summary;}
|
|
|
+ return {ok:true,json:async()=>({success:true,data})};
|
|
|
+ };
|
|
|
+ const r=await runDirectSync({XIAOSHU_SYNC_REPORT_DIR:root,XIAOSHU_SYNC_OPERATOR_TOKEN:'test'});
|
|
|
+ assert.equal(r.status,'blocked');assert.match(r.reason,/checksum/);assert.equal(finished,false);
|
|
|
+ assert.ok(!(await readdir(root)).includes('direct-nightly.lock'));
|
|
|
+ }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
|
|
|
+});
|
|
|
+
|
|
|
+test('只有全部数据集对账后才能完成，统计来自写入回执，完成重试幂等',async()=>{
|
|
|
+ await db.exec('CREATE TABLE "SysLog" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,type1 text,type2 text,type3 text,detail text,"createdAt" timestamptz,"updatedAt" timestamptz)');
|
|
|
+ const previous=context.fetch;
|
|
|
+ context.fetch=async()=>({ok:true,json:async()=>({retcode:0,result:[],page:{itemCount:0,pageCount:1}})});
|
|
|
+ try{
|
|
|
+ const datasets=[];
|
|
|
+ for(const modelId of [2,52,58,56,53,60,54,59,61]){
|
|
|
+ assert.equal((await call({action:'source',modelId,page:1})).success,true);
|
|
|
+ const e=await call({action:'evidence',modelId});assert.equal(e.success,true,e.message);
|
|
|
+ datasets.push({modelId,created:999,updated:999,failed:0,reconciled:true});
|
|
|
+ }
|
|
|
+ const summary={status:'completed-with-preserved-conflicts',datasets,conflicts:1};
|
|
|
+ const r=await call({action:'finish',summary});assert.equal(r.success,true,r.message);
|
|
|
+ assert.equal(r.data.createdBusinessRecords,2);assert.equal(r.data.refreshedBusinessRecords,1);
|
|
|
+ assert.equal(r.data.bridgeUsed,false);assert.equal(r.data.settlementReady,false);
|
|
|
+ const again=await call({action:'finish',summary});assert.deepEqual(JSON.parse(JSON.stringify(again.data)),JSON.parse(JSON.stringify(r.data)));
|
|
|
+ assert.equal((await sql('SELECT count(*)::int n FROM "SysLog"'))[0].n,1);
|
|
|
+ assert.equal((await call({action:'claim',proposedRunId:'next-run-0000001'})).data.alreadyCompleted,true);
|
|
|
+ }finally{context.fetch=previous;}
|
|
|
+});
|
|
|
+
|
|
|
+
|
|
|
+test('只清理七天前已完成运行的原始缓存，保留失败断点及报告',async()=>{
|
|
|
+ const root=await mkdtemp(join(tmpdir(),'direct-sync-retention-'));
|
|
|
+ const ids=['11111111-1111-1111-1111-111111111111','22222222-2222-2222-2222-222222222222','33333333-3333-3333-3333-333333333333'];
|
|
|
+ try{
|
|
|
+ for(let i=0;i<ids.length;i++){
|
|
|
+ await mkdir(join(root,'direct-'+ids[i]));
|
|
|
+ await writeFile(join(root,'2026-10-01-direct-'+ids[i]+'.json'),JSON.stringify({runId:ids[i],status:i===1?'blocked':'completed',finishedAt:i===2?'2026-10-12T12:00:00Z':'2026-10-01T12:00:00Z'}));
|
|
|
+ }
|
|
|
+ await pruneCompletedCaches(root,Date.parse('2026-10-13T00:00:00Z'));
|
|
|
+ const files=await readdir(root);
|
|
|
+ assert.ok(!files.includes('direct-'+ids[0]));assert.ok(files.includes('direct-'+ids[1]));assert.ok(files.includes('direct-'+ids[2]));
|
|
|
+ assert.equal(files.filter(x=>x.endsWith('.json')).length,3);
|
|
|
+ }finally{await rm(root,{recursive:true,force:true});}
|
|
|
+});
|
彭峰