kangshifu-board/server/test/data-dictionary.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import {readProjectDataDictionary,readAuthorizedDataDictionary} from '../src/data-dictionary.mjs';
import {describeBusinessSchema} from '../../skills/ksf-workshop-runtime/scripts/describe-business-schema.mjs';
import {runWorkshop} from '../../skills/ksf-workshop-runtime/scripts/ksf-workshop.mjs';
const dictionary=JSON.parse(fs.readFileSync(new URL('../references/business-data-dictionary.json',import.meta.url),'utf8'));
const definitions=JSON.parse(fs.readFileSync(new URL('../references/runtime-data.json',import.meta.url),'utf8'));

test('完整字典逐表逐列覆盖实际维护定义，来源、系统字段与写入口分开',()=>{
 assert.deepEqual(Object.keys(dictionary.tables),Object.keys(definitions.tables));
 assert.equal(dictionary.fieldCount,1388);
 for(const [table,defs]of Object.entries(definitions.tables)){
  assert.deepEqual(Object.keys(dictionary.tables[table].fields),Object.keys(defs));
  for(const [name,def]of Object.entries(defs)){
   const field=dictionary.tables[table].fields[name];assert.ok(field.meaning);assert.equal(field.type,def.type);
   assert.equal(field.targetClass,def.targetClass);
  }
 }
 assert.equal(dictionary.tables.KsfServiceCredential.fields.tokenHash.allowedInGenericImport,false);
 assert.equal(dictionary.tables.KfProduct.fields.objectId.writeOwner,'service');
});
test('结构接口只查元信息，保留未知表列并拒绝SQL注入',async()=>{
 const queries=[];const pg={any:async(sql,args)=>{
  queries.push({sql,args});assert.ok(sql.includes('information_schema.'));assert.ok(!/SELECT\s+\*|INSERT|UPDATE|DELETE|ALTER/i.test(sql));
  return sql.includes('information_schema.columns')?[
   {table_name:'KfProduct',column_name:'name',data_type:'text',udt_name:'text',is_nullable:'YES',has_default:false},
   {table_name:'KfProduct',column_name:'newField',data_type:'text',is_nullable:'YES'},
   {table_name:'KfExtra',column_name:'newColumn',data_type:'text',is_nullable:'YES'},
   {table_name:'KsfServiceCredential',column_name:'tokenHash',data_type:'text',is_nullable:'YES'}]:[];
 }};
 const result=await readProjectDataDictionary(pg,dictionary);
 assert.equal(result.databaseWrites,0);assert.equal(result.rowValuesReturned,false);assert.equal(result.tableCount,3);
 assert.equal(result.semanticCoverageComplete,false);assert.deepEqual(result.unknownFields,['KfProduct.newField','KfExtra.newColumn']);
 assert.ok(!JSON.stringify(result).includes('secret-value'));
 await assert.rejects(()=>readProjectDataDictionary(pg,dictionary,{table:'KfProduct;DROP TABLE x'}),{code:'VALIDATION_FAILED'});
 assert.equal(queries.length,2);
 const compared=describeBusinessSchema(dictionary,{live:result});assert.equal(compared.live.readyForFieldMapping,false);
});
test('workshop-schema只发结构查询，不创建身份或业务记录',async()=>{
 const calls=[];const r=await runWorkshop('workshop-schema',{summary:true},{requestId:'schema-1',call:async(name,p)=>{
  calls.push({name,p});return{readOnly:true,databaseWrites:0,tables:[]};
 }});
 assert.equal(r.databaseWrites,0);assert.deepEqual(calls,[{name:'ksfGetDataDictionary',p:{summary:true,requestId:'schema-1'}}]);
});
test('匿名与缺data:read在读取实际表结构前停止，不因字典扩大行数据权限',async()=>{
 let reads=0;
 const pg={any:async()=>{reads++;return[];}};
 const error=code=>{throw Object.assign(Error(code),{code});};
 const app={authenticate:async()=>null,need:actor=>{if(!actor)error('AUTH_REQUIRED');if(!actor.scopes.includes('data:read'))error('FORBIDDEN');},fail:error};
 await assert.rejects(()=>readAuthorizedDataDictionary({app,pg,dictionary},{},{}),{code:'AUTH_REQUIRED'});
 app.authenticate=async()=>({scopes:['ingest:write']});
 await assert.rejects(()=>readAuthorizedDataDictionary({app,pg,dictionary},{},{}),{code:'FORBIDDEN'});
 assert.equal(reads,0);
});

test('真实列类型漂移可识别，摘要回执不能冒充逐字段验证',()=>{
 const live={schema:'ksf.live-business-data-dictionary.v1',readOnly:true,databaseWrites:0,
  semanticCoverageComplete:true,unknownFields:[],tableCount:1,fieldCount:1,
  tables:[{name:'KfProduct',fields:[{name:'name',logicalName:'name',physicalType:'boolean'}]}]};
 const r=describeBusinessSchema(dictionary,{live});
 assert.equal(r.live.readyForFieldMapping,false);
 assert.ok(r.live.differences.some(x=>x.reason==='TYPE_MISMATCH'&&x.field==='name'));
 assert.throws(()=>describeBusinessSchema(dictionary,{live:{...live,tables:[{name:'KfProduct',fieldCount:1}]}}),/LIVE_DETAILED_SCHEMA_RECEIPT_REQUIRED/);
});