fix: enforce drainage uniqueness and carrier-specific reporting
This commit is contained in:
hectorzhao
@@ -0,0 +1,393 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import { createRequire } from 'node:module';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import fs from 'node:fs';
|
||||
const require = createRequire(new URL('../../api/package.json', import.meta.url));
|
||||
const url = new URL(process.env.DRAINAGE_TEST_DATABASE_URL || '');
|
||||
assert(
|
||||
['localhost', '127.0.0.1'].includes(url.hostname) && url.pathname.startsWith('/cmpp_qa_'),
|
||||
'Dedicated loopback QA database required',
|
||||
);
|
||||
process.env.DATABASE_URL = url.toString();
|
||||
process.env.NODE_ENV = 'test';
|
||||
const { PrismaService } = require('./dist/prisma/prisma.service.js');
|
||||
const { SmsConfigService } = require('./dist/sms-config/sms-config.service.js');
|
||||
const { ChannelReportingService } = require('./dist/channels/channel-reporting.service.js');
|
||||
const { OpenApiService } = require('./dist/open-api/open-api.service.js');
|
||||
const { ReportBatchGenerationService } = require('./dist/report-materials/batch-generation.service.js');
|
||||
const { OperationsMessageQueries } = require('./dist/operations/queries/messages.queries.js');
|
||||
const { assessDrainage } = require('./dist/send-chain/drainage-authorization.js');
|
||||
const express = require('express');
|
||||
const db = new PrismaService();
|
||||
const sms = new SmsConfigService(db);
|
||||
const reporting = new ChannelReportingService(db);
|
||||
// No onModuleInit: transport, workers, reconciliation and authentication are outside this loopback service harness.
|
||||
const httpConfig = new OpenApiService(db, undefined);
|
||||
const batch = new ReportBatchGenerationService(db, undefined, sms, undefined, undefined);
|
||||
const messages = new OperationsMessageQueries(db);
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
app.set('json replacer', (_, value) => (typeof value === 'bigint' ? value.toString() : value));
|
||||
const wrap = (fn) => async (req, res) => {
|
||||
try {
|
||||
res.json(await fn(req));
|
||||
} catch (e) {
|
||||
res.status(e.getStatus?.() || 500).json({ message: e.message });
|
||||
}
|
||||
};
|
||||
app.get(
|
||||
'/api/admin/enterprise-signatures/:id',
|
||||
wrap((r) => sms.getSignature(r.params.id)),
|
||||
);
|
||||
app.post(
|
||||
'/api/admin/enterprise-signatures/:id/drainage-infos',
|
||||
wrap((r) => sms.createDrainageInfo(r.params.id, r.body, { initialAuditStatus: 'approved' })),
|
||||
);
|
||||
app.put(
|
||||
'/api/admin/drainage-infos/:id',
|
||||
wrap((r) => sms.updateDrainageInfo(r.params.id, r.body, { initialAuditStatus: 'approved' })),
|
||||
);
|
||||
app.get(
|
||||
'/api/admin/enterprise-applications/:id/report-fields',
|
||||
wrap((r) => sms.getApplicationReportFields(r.params.id)),
|
||||
);
|
||||
app.post(
|
||||
'/api/admin/drainage-infos/:id/status',
|
||||
wrap((r) => sms.changeDrainageInfoStatus(r.params.id, r.body)),
|
||||
);
|
||||
app.get(
|
||||
'/api/admin/drainage-infos/:id/report-targets',
|
||||
wrap((r) => sms.getDrainageReportTargets(r.params.id)),
|
||||
);
|
||||
app.post(
|
||||
'/api/admin/report-tasks/status-change',
|
||||
wrap((r) => reporting.changeReportTaskStatuses(r.body)),
|
||||
);
|
||||
app.get(
|
||||
'/api/admin/enterprise-applications/:id/http-api',
|
||||
wrap((r) => httpConfig.getConfig(r.params.id)),
|
||||
);
|
||||
app.put(
|
||||
'/api/admin/enterprise-applications/:id/http-api',
|
||||
wrap((r) => httpConfig.updateConfig(r.params.id, r.body)),
|
||||
);
|
||||
app.get(
|
||||
'/api/admin/messages/:id',
|
||||
wrap((r) => messages.getMessage(r.params.id)),
|
||||
);
|
||||
const port = Number(process.env.DRAINAGE_TEST_PORT || 16416);
|
||||
const server = await new Promise((resolve) => {
|
||||
const listener = app.listen(port, '127.0.0.1', () => resolve(listener));
|
||||
});
|
||||
const base = 'http://127.0.0.1:' + port;
|
||||
const checks = [];
|
||||
const check = (name, value) => {
|
||||
assert(value, name);
|
||||
checks.push(name);
|
||||
};
|
||||
const request = async (method, path, body) => {
|
||||
const r = await fetch(base + '/api/admin/' + path, {
|
||||
method,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: body === undefined ? undefined : JSON.stringify(body),
|
||||
});
|
||||
return { status: r.status, data: await r.json() };
|
||||
};
|
||||
try {
|
||||
const tag = randomUUID().slice(0, 8);
|
||||
const { Client } = require('pg');
|
||||
const migrationDb = new Client({ connectionString: url.toString() });
|
||||
await migrationDb.connect();
|
||||
try {
|
||||
await migrationDb.query('CREATE SCHEMA qa_migration_' + tag);
|
||||
await migrationDb.query('SET search_path TO qa_migration_' + tag);
|
||||
await migrationDb.query(
|
||||
'CREATE TABLE "ChannelSignatureReportTask" ("id" text, "signatureId" text, "drainageItemId" text, "channelId" text, "carrier" text, "reportType" text, "status" text)',
|
||||
);
|
||||
await migrationDb.query(
|
||||
`CREATE UNIQUE INDEX "ChannelSignatureReportTask_drainage_target_key" ON "ChannelSignatureReportTask" ("signatureId", "drainageItemId", "channelId") WHERE "reportType"='drainage' AND "drainageItemId" IS NOT NULL`,
|
||||
);
|
||||
await migrationDb.query(
|
||||
`INSERT INTO "ChannelSignatureReportTask" VALUES ('old','s','d','c',NULL,'drainage','approved')`,
|
||||
);
|
||||
const oldRow = (await migrationDb.query('SELECT * FROM "ChannelSignatureReportTask"')).rows;
|
||||
await migrationDb.query(
|
||||
fs.readFileSync(
|
||||
new URL('../../api/prisma/migrations/20260914093000_drainage_carrier_reports/migration.sql', import.meta.url),
|
||||
'utf8',
|
||||
),
|
||||
);
|
||||
assert.deepEqual((await migrationDb.query('SELECT * FROM "ChannelSignatureReportTask"')).rows, oldRow);
|
||||
check('migration preserves legacy approval exactly', true);
|
||||
for (const carrier of ['mobile', 'unicom', 'telecom'])
|
||||
await migrationDb.query(
|
||||
`INSERT INTO "ChannelSignatureReportTask" VALUES ($1,'s','d','c',$1,'drainage','pending')`,
|
||||
[carrier],
|
||||
);
|
||||
for (const carrier of ['mobile', null])
|
||||
await assert.rejects(
|
||||
migrationDb.query(
|
||||
`INSERT INTO "ChannelSignatureReportTask" VALUES ('dup','s','d','c',$1,'drainage','pending')`,
|
||||
[carrier],
|
||||
),
|
||||
{ code: '23505' },
|
||||
);
|
||||
check('migration permits independent carriers and rejects carrier and legacy duplicates', true);
|
||||
} finally {
|
||||
await migrationDb.end();
|
||||
}
|
||||
const tenant = await db.tenant.create({ data: { name: '隔离引流验收', code: 'QA-' + tag } });
|
||||
const application = await db.smsApplication.create({
|
||||
data: {
|
||||
tenantId: tenant.id,
|
||||
name: '隔离引流应用',
|
||||
cmppAccount: 'qa-' + tag,
|
||||
cmppEnterpriseCode: 'QA',
|
||||
secretHash: randomUUID(),
|
||||
interfaceEnabled: false,
|
||||
},
|
||||
});
|
||||
const signature = await db.smsSignature.create({
|
||||
data: { tenantId: tenant.id, applicationId: application.id, name: '【引流三网验收】', auditStatus: 'approved' },
|
||||
});
|
||||
const second = await db.smsSignature.create({
|
||||
data: { tenantId: tenant.id, applicationId: application.id, name: '【其他签名】', auditStatus: 'approved' },
|
||||
});
|
||||
const channel = await db.smsChannel.create({
|
||||
data: {
|
||||
code: 'QA-' + tag,
|
||||
name: '隔离三网通道',
|
||||
carriers: ['mobile', 'unicom', 'telecom'],
|
||||
gatewayHost: '127.0.0.1',
|
||||
gatewayPort: 1,
|
||||
account: 'unused',
|
||||
passwordCipher: 'unused',
|
||||
srcId: '1069',
|
||||
},
|
||||
});
|
||||
for (const carrier of ['mobile', 'unicom', 'telecom']) {
|
||||
const group = await db.smsChannelGroup.create({
|
||||
data: {
|
||||
code: 'QA-' + tag + '-' + carrier,
|
||||
name: carrier,
|
||||
carrier,
|
||||
items: { create: { channelId: channel.id, carrier } },
|
||||
},
|
||||
});
|
||||
await db.channelRouteRule.create({
|
||||
data: { tenantId: tenant.id, applicationId: application.id, groupId: group.id, carrier },
|
||||
});
|
||||
}
|
||||
const field = await db.drainageField.create({ data: { code: 'qa_' + tag, name: '引流信息', fieldType: 'string' } });
|
||||
await db.channelReportField.create({
|
||||
data: {
|
||||
channelId: channel.id,
|
||||
drainageFieldId: field.id,
|
||||
code: field.code,
|
||||
name: field.name,
|
||||
reportType: 'drainage',
|
||||
exportName: field.name,
|
||||
fieldType: 'string',
|
||||
required: false,
|
||||
},
|
||||
});
|
||||
const created = await request('POST', 'enterprise-signatures/' + signature.id + '/drainage-infos', {
|
||||
url: 'example.com',
|
||||
});
|
||||
if (created.status !== 200) console.error('create response', created);
|
||||
check('create', created.status === 200);
|
||||
const drainage = created.data;
|
||||
const duplicate = await request('POST', 'enterprise-signatures/' + signature.id + '/drainage-infos', {
|
||||
url: ' example.com ',
|
||||
});
|
||||
check('duplicate trimmed value returns 400', duplicate.status === 400);
|
||||
const other = await request('POST', 'enterprise-signatures/' + second.id + '/drainage-infos', { url: 'example.com' });
|
||||
check('same value in another signature allowed', other.status === 200);
|
||||
const parallel = await Promise.all(
|
||||
Array.from({ length: 5 }, () =>
|
||||
request('POST', 'enterprise-signatures/' + signature.id + '/drainage-infos', { url: 'parallel.example.com' }),
|
||||
),
|
||||
);
|
||||
check(
|
||||
'five concurrent creates commit once',
|
||||
parallel.filter((x) => x.status === 200).length === 1 && parallel.filter((x) => x.status === 400).length === 4,
|
||||
);
|
||||
const parallelItem = parallel.find((x) => x.status === 200).data;
|
||||
const conflicting = await request('PUT', 'drainage-infos/' + parallelItem.id, { url: 'example.com' });
|
||||
check('edit cannot collide', conflicting.status === 400);
|
||||
const editA = (
|
||||
await request('POST', 'enterprise-signatures/' + signature.id + '/drainage-infos', { url: 'a.example.net' })
|
||||
).data;
|
||||
const editB = (
|
||||
await request('POST', 'enterprise-signatures/' + signature.id + '/drainage-infos', { url: 'b.example.net' })
|
||||
).data;
|
||||
const racingEdits = await Promise.all(
|
||||
[editA, editB].map((x) => request('PUT', 'drainage-infos/' + x.id, { url: 'race.example.net' })),
|
||||
);
|
||||
check(
|
||||
'concurrent edits commit one target',
|
||||
racingEdits.filter((x) => x.status === 200).length === 1 &&
|
||||
racingEdits.filter((x) => x.status === 400).length === 1,
|
||||
);
|
||||
await db.smsDrainageInfo.update({ where: { id: parallelItem.id }, data: { auditStatus: 'deleted' } });
|
||||
check(
|
||||
'deleted target can be reused',
|
||||
(
|
||||
await request('POST', 'enterprise-signatures/' + signature.id + '/drainage-infos', {
|
||||
url: 'parallel.example.com',
|
||||
})
|
||||
).status === 200,
|
||||
);
|
||||
const unchanged = await request('PUT', 'drainage-infos/' + drainage.id, {
|
||||
url: 'example.com',
|
||||
remark: 'unchanged value',
|
||||
});
|
||||
check('edit self allowed', unchanged.status === 200);
|
||||
check(
|
||||
'restoration cannot bypass uniqueness',
|
||||
(await request('POST', `drainage-infos/${parallelItem.id}/status`, { status: 'approved' })).status === 400,
|
||||
);
|
||||
const targets = (await request('GET', 'drainage-infos/' + drainage.id + '/report-targets')).data;
|
||||
check('one channel has three targets', targets.length === 3 && new Set(targets.map((x) => x.carrier)).size === 3);
|
||||
const states = { mobile: 'approved', unicom: 'failed', telecom: 'pending' };
|
||||
const body = {
|
||||
items: targets.map((t) => ({
|
||||
signatureId: signature.id,
|
||||
drainageItemId: drainage.id,
|
||||
reportType: 'drainage',
|
||||
channelId: channel.id,
|
||||
carrier: t.carrier,
|
||||
status: states[t.carrier],
|
||||
})),
|
||||
sourceEntry: 'enterprise_signature',
|
||||
};
|
||||
check('save three carrier states', (await request('POST', 'report-tasks/status-change', body)).status === 200);
|
||||
const rows = await db.smsDrainageInfo.findMany({ where: { id: drainage.id }, include: { reportTasks: true } });
|
||||
const target = {
|
||||
key: 'url:example.com',
|
||||
category: 'url',
|
||||
value: 'example.com',
|
||||
text: 'example.com',
|
||||
start: 0,
|
||||
end: 11,
|
||||
};
|
||||
check('mobile route approved', assessDrainage([target], rows, 'mobile').allowedChannelIds.includes(channel.id));
|
||||
check('unicom route denied', assessDrainage([target], rows, 'unicom').allowedChannelIds.length === 0);
|
||||
check('telecom pending denied', assessDrainage([target], rows, 'telecom').allowedChannelIds.length === 0);
|
||||
await db.channelSignatureReportTask.create({
|
||||
data: {
|
||||
tenantId: tenant.id,
|
||||
signatureId: signature.id,
|
||||
drainageItemId: drainage.id,
|
||||
channelId: channel.id,
|
||||
reportType: 'drainage',
|
||||
carrier: null,
|
||||
status: 'approved',
|
||||
},
|
||||
});
|
||||
const legacyRows = await db.smsDrainageInfo.findMany({ where: { id: drainage.id }, include: { reportTasks: true } });
|
||||
check(
|
||||
'legacy approval cannot override explicit rejection',
|
||||
assessDrainage([target], legacyRows, 'unicom').allowedChannelIds.length === 0,
|
||||
);
|
||||
const view = (await request('GET', 'enterprise-signatures/' + signature.id)).data;
|
||||
check(
|
||||
'summary follows carrier',
|
||||
view.drainageCarrierReportSummary[drainage.id].mobile.approved === 1 &&
|
||||
view.drainageCarrierReportSummary[drainage.id].unicom.approved === 0,
|
||||
);
|
||||
const preview = await batch.inspectBatchItem({
|
||||
reportType: 'drainage',
|
||||
signatureId: signature.id,
|
||||
drainageItemId: drainage.id,
|
||||
});
|
||||
check(
|
||||
'batch targets preserve carriers',
|
||||
preview.targets.length === 3 && preview.targets.every((t) => t.carrier !== 'all'),
|
||||
);
|
||||
const details = await reporting.listReportDetailsPage({
|
||||
reportType: 'drainage',
|
||||
channelId: channel.id,
|
||||
signatureId: signature.id,
|
||||
pageSize: 100,
|
||||
});
|
||||
check(
|
||||
'report details unique carrier rows',
|
||||
details.items.filter((t) => t.drainageItemId === drainage.id).length === 3,
|
||||
);
|
||||
check(
|
||||
'material update',
|
||||
(await request('PUT', 'drainage-infos/' + drainage.id, { url: 'example.com', remark: 'invalidate reports' }))
|
||||
.status === 200,
|
||||
);
|
||||
const reset = await db.channelSignatureReportTask.findMany({ where: { drainageItemId: drainage.id } });
|
||||
check(
|
||||
'material update invalidates all approvals',
|
||||
reset.every((t) => t.status !== 'approved'),
|
||||
);
|
||||
check(
|
||||
'material update resets three carrier tasks',
|
||||
reset.filter((t) => t.carrier && t.status === 'pending').length === 3,
|
||||
);
|
||||
const whitelist = ['203.0.113.1', '203.0.113.0/24', '2001:db8::1'];
|
||||
check(
|
||||
'HTTP whitelist persists',
|
||||
(
|
||||
await request('PUT', 'enterprise-applications/' + application.id + '/http-api', {
|
||||
enabled: false,
|
||||
ipAllowlist: whitelist,
|
||||
})
|
||||
).status === 200,
|
||||
);
|
||||
const config = (await request('GET', 'enterprise-applications/' + application.id + '/http-api')).data;
|
||||
check(
|
||||
'HTTP whitelist readback',
|
||||
JSON.stringify([...config.ipAllowlist].sort()) === JSON.stringify([...whitelist].sort()),
|
||||
);
|
||||
const message = await db.smsMessageRecord.create({
|
||||
data: {
|
||||
messageId: 'QA-' + tag,
|
||||
tenantId: tenant.id,
|
||||
applicationId: application.id,
|
||||
signatureId: signature.id,
|
||||
phoneNumber: '13800001000',
|
||||
content: '【引流三网验收】只读展示',
|
||||
status: 'failed',
|
||||
},
|
||||
});
|
||||
await db.smsChannelSensitiveDecision.create({
|
||||
data: {
|
||||
messageRecordId: message.id,
|
||||
routeAttemptId: randomUUID(),
|
||||
snapshot: { hits: [], reason: null, candidateChannelIds: [], selectedChannelId: null },
|
||||
},
|
||||
});
|
||||
const fixture = {
|
||||
tag,
|
||||
tenantId: tenant.id,
|
||||
applicationId: application.id,
|
||||
signatureId: signature.id,
|
||||
drainageId: drainage.id,
|
||||
channelId: channel.id,
|
||||
messageId: message.id,
|
||||
base,
|
||||
checks,
|
||||
};
|
||||
if (process.env.DRAINAGE_TEST_EVIDENCE)
|
||||
fs.writeFileSync(process.env.DRAINAGE_TEST_EVIDENCE, JSON.stringify(fixture, null, 2));
|
||||
console.log(JSON.stringify({ passed: checks.length, checks, fixture }));
|
||||
if (process.env.DRAINAGE_TEST_KEEP_SERVER !== 'true') {
|
||||
await new Promise((r) => server.close(r));
|
||||
await db.$disconnect();
|
||||
}
|
||||
} catch (e) {
|
||||
console.error(e);
|
||||
await new Promise((r) => server.close(r));
|
||||
await db.$disconnect();
|
||||
process.exitCode = 1;
|
||||
}
|
||||
process.on('SIGINT', async () => {
|
||||
await new Promise((r) => server.close(r));
|
||||
await db.$disconnect();
|
||||
process.exit(0);
|
||||
});
|
||||
Reference in New Issue
Block a user