forked from wangziqi/gongxue-base
300 lines
9.6 KiB
JavaScript
300 lines
9.6 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import pg from 'pg';
|
|
import { spawn } from 'node:child_process';
|
|
|
|
const databaseUrl = process.env.DATABASE_URL || 'postgresql://postgres:postgres@127.0.0.1:54322/postgres';
|
|
const tenantId = '00000000-0000-0000-0000-000000000001';
|
|
const adminUserId = '00000000-0000-0000-0000-000000000102';
|
|
const ids = {
|
|
region: '00000000-0000-0000-0000-000000000301',
|
|
subject: '00000000-0000-0000-0000-000000000501',
|
|
category: '00000000-0000-0000-0000-000000000601',
|
|
contentEntry: '00000000-0000-0000-0000-000000000611',
|
|
contentNodeSchoolTarget: '00000000-0000-0000-0000-000000000614',
|
|
questionCollection: '00000000-0000-0000-0000-000000000615',
|
|
};
|
|
|
|
function runWorkerOnce() {
|
|
const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'imports'], {
|
|
cwd: process.cwd(),
|
|
env: {
|
|
...process.env,
|
|
DATABASE_URL: databaseUrl,
|
|
WORKER_IMPORT_BATCH_SIZE: '5',
|
|
WORKER_IMPORT_ID: 'imports-integration-test',
|
|
},
|
|
stdio: ['ignore', 'pipe', 'pipe'],
|
|
windowsHide: true,
|
|
});
|
|
let output = '';
|
|
child.stdout.on('data', chunk => {
|
|
output += chunk.toString();
|
|
});
|
|
child.stderr.on('data', chunk => {
|
|
output += chunk.toString();
|
|
});
|
|
return new Promise((resolve, reject) => {
|
|
child.on('error', reject);
|
|
child.on('exit', code => {
|
|
try {
|
|
assert.equal(code, 0, `worker should exit 0\n${output}`);
|
|
assert.match(output, /imports batch processed=\d+/, 'worker output should include imports summary');
|
|
resolve(output);
|
|
} catch (error) {
|
|
reject(error);
|
|
}
|
|
});
|
|
});
|
|
}
|
|
|
|
async function cleanup(pool) {
|
|
await pool.query(
|
|
`
|
|
delete from public.question_collection_items
|
|
where tenant_id = $1
|
|
and question_id in (
|
|
select id from public.questions
|
|
where tenant_id = $1
|
|
and (
|
|
legacy_id like 'worker-import-question-%'
|
|
or legacy_id like 'integration-import-async-choice-%'
|
|
)
|
|
)
|
|
`,
|
|
[tenantId],
|
|
);
|
|
await pool.query(
|
|
`
|
|
delete from public.question_versions
|
|
where tenant_id = $1
|
|
and question_id in (
|
|
select id from public.questions
|
|
where tenant_id = $1
|
|
and (
|
|
legacy_id like 'worker-import-question-%'
|
|
or legacy_id like 'integration-import-async-choice-%'
|
|
)
|
|
)
|
|
`,
|
|
[tenantId],
|
|
);
|
|
await pool.query(
|
|
`
|
|
delete from public.questions
|
|
where tenant_id = $1
|
|
and (
|
|
legacy_id like 'worker-import-question-%'
|
|
or legacy_id like 'integration-import-async-choice-%'
|
|
)
|
|
`,
|
|
[tenantId],
|
|
);
|
|
await pool.query(
|
|
`
|
|
delete from public.audit_logs
|
|
where tenant_id = $1
|
|
and target_type = 'content_import_job'
|
|
and details::text like '%worker-import%'
|
|
`,
|
|
[tenantId],
|
|
);
|
|
await pool.query(
|
|
`
|
|
delete from public.content_import_jobs
|
|
where tenant_id = $1
|
|
and (
|
|
source_name like 'worker-import-%'
|
|
or source_name = 'async-question-import.json'
|
|
)
|
|
`,
|
|
[tenantId],
|
|
);
|
|
}
|
|
|
|
async function createQueuedQuestionImport(pool) {
|
|
const preview = await pool.query(
|
|
`
|
|
insert into public.content_import_jobs (
|
|
tenant_id, created_by, import_type, source_format, status,
|
|
source_name, source_hash, target_region_id, target_subject_id,
|
|
target_category_id, target_content_node_id, target_collection_id,
|
|
dry_run, total_count, valid_count, error_count, warning_count,
|
|
summary, raw_payload, normalized_payload, execution_mode, queued_at,
|
|
next_attempt_at, parser_metadata
|
|
)
|
|
values (
|
|
$1, $2, 'questions', 'json', 'pending',
|
|
'worker-import-questions.json', 'worker-import-source-hash',
|
|
$3::uuid, $4::uuid, $5::uuid, $6::uuid, $7::uuid,
|
|
false, 1, 1, 0, 0,
|
|
$8::jsonb, $9::jsonb, $10::jsonb, 'async', now(), now(), '{}'::jsonb
|
|
)
|
|
returning id
|
|
`,
|
|
[
|
|
tenantId,
|
|
adminUserId,
|
|
ids.region,
|
|
ids.subject,
|
|
ids.category,
|
|
ids.contentNodeSchoolTarget,
|
|
ids.questionCollection,
|
|
JSON.stringify({
|
|
target: {
|
|
regionId: ids.region,
|
|
subjectId: ids.subject,
|
|
categoryId: ids.category,
|
|
contentNodeId: ids.contentNodeSchoolTarget,
|
|
collectionId: ids.questionCollection,
|
|
},
|
|
importOptions: { allowPartial: false },
|
|
source: 'worker-import-integration',
|
|
}),
|
|
JSON.stringify([
|
|
{
|
|
legacyId: 'worker-import-question-001',
|
|
type: 'choice',
|
|
content: '异步导入题：worker 应该复用哪套导入规则？',
|
|
options: ['自己重写', '复用后端导入 executor', '前端直写数据库', '跳过校验'],
|
|
correctOptionIndices: [1],
|
|
explanation: 'worker 和 API 必须复用同一套后端导入规则。',
|
|
difficulty: 2,
|
|
tags: ['worker-import'],
|
|
},
|
|
]),
|
|
JSON.stringify([
|
|
{
|
|
legacyId: 'worker-import-question-001',
|
|
type: 'choice',
|
|
typeLabel: null,
|
|
content: '异步导入题：worker 应该复用哪套导入规则？',
|
|
options: ['自己重写', '复用后端导入 executor', '前端直写数据库', '跳过校验'],
|
|
correctOptionIndex: 1,
|
|
correctOptionIndices: [1],
|
|
answerText: null,
|
|
explanation: 'worker 和 API 必须复用同一套后端导入规则。',
|
|
difficulty: 2,
|
|
tags: ['worker-import'],
|
|
mediaUrl: null,
|
|
subQuestions: [],
|
|
codeLang: null,
|
|
codeTemplate: null,
|
|
examMarkers: {},
|
|
sourceHash: 'worker-import-question-hash-001',
|
|
},
|
|
]),
|
|
],
|
|
);
|
|
const jobId = preview.rows[0].id;
|
|
await pool.query(
|
|
`
|
|
insert into public.content_import_items (
|
|
tenant_id, job_id, row_no, external_id, status, target_type,
|
|
source_payload, normalized_payload, content_hash, issues_count
|
|
)
|
|
values ($1, $2, 1, 'worker-import-question-001', 'valid', 'question', $3::jsonb, $4::jsonb, 'worker-import-question-hash-001', 0)
|
|
`,
|
|
[
|
|
tenantId,
|
|
jobId,
|
|
JSON.stringify({
|
|
legacyId: 'worker-import-question-001',
|
|
content: '异步导入题：worker 应该复用哪套导入规则？',
|
|
}),
|
|
JSON.stringify({
|
|
legacyId: 'worker-import-question-001',
|
|
type: 'choice',
|
|
typeLabel: null,
|
|
content: '异步导入题：worker 应该复用哪套导入规则？',
|
|
options: ['自己重写', '复用后端导入 executor', '前端直写数据库', '跳过校验'],
|
|
correctOptionIndex: 1,
|
|
correctOptionIndices: [1],
|
|
answerText: null,
|
|
explanation: 'worker 和 API 必须复用同一套后端导入规则。',
|
|
difficulty: 2,
|
|
tags: ['worker-import'],
|
|
mediaUrl: null,
|
|
subQuestions: [],
|
|
codeLang: null,
|
|
codeTemplate: null,
|
|
examMarkers: {},
|
|
sourceHash: 'worker-import-question-hash-001',
|
|
}),
|
|
],
|
|
);
|
|
return jobId;
|
|
}
|
|
|
|
async function main() {
|
|
const pool = new pg.Pool({ connectionString: databaseUrl });
|
|
try {
|
|
await cleanup(pool);
|
|
const jobId = await createQueuedQuestionImport(pool);
|
|
|
|
const output = await runWorkerOnce();
|
|
assert.match(output, /completed=1/, 'worker should complete exactly the queued import job after cleanup');
|
|
|
|
const job = await pool.query(
|
|
`
|
|
select status, execution_mode, inserted_count, updated_count, skipped_count,
|
|
locked_at, locked_by, attempt_count, error_message
|
|
from public.content_import_jobs
|
|
where tenant_id = $1 and id = $2
|
|
`,
|
|
[tenantId, jobId],
|
|
);
|
|
assert.equal(job.rows[0]?.status, 'completed', 'queued import job should be completed');
|
|
assert.equal(job.rows[0]?.execution_mode, 'async', 'job should keep async execution mode');
|
|
assert.equal(Number(job.rows[0]?.inserted_count), 1, 'worker should insert one question');
|
|
assert.equal(job.rows[0]?.locked_at, null, 'completed job should release lock');
|
|
assert.equal(job.rows[0]?.locked_by, null, 'completed job should clear lock owner');
|
|
assert.equal(Number(job.rows[0]?.attempt_count), 1, 'worker should record one attempt');
|
|
assert.equal(job.rows[0]?.error_message, null, 'completed job should not retain error message');
|
|
|
|
const question = await pool.query(
|
|
`
|
|
select q.id, v.content
|
|
from public.questions q
|
|
join public.question_versions v on v.id = q.current_version_id
|
|
where q.tenant_id = $1 and q.legacy_id = 'worker-import-question-001'
|
|
limit 1
|
|
`,
|
|
[tenantId],
|
|
);
|
|
assert.equal(question.rows[0]?.content, '异步导入题：worker 应该复用哪套导入规则？', 'worker should import question content');
|
|
|
|
const collectionItem = await pool.query(
|
|
`
|
|
select 1
|
|
from public.question_collection_items
|
|
where tenant_id = $1 and collection_id = $2 and question_id = $3
|
|
limit 1
|
|
`,
|
|
[tenantId, ids.questionCollection, question.rows[0]?.id],
|
|
);
|
|
assert.equal(collectionItem.rowCount, 1, 'worker should bind imported question to collection');
|
|
|
|
const audit = await pool.query(
|
|
`
|
|
select action
|
|
from public.audit_logs
|
|
where tenant_id = $1 and target_type = 'content_import_job' and target_id = $2
|
|
order by created_at desc
|
|
limit 1
|
|
`,
|
|
[tenantId, jobId],
|
|
);
|
|
assert.equal(audit.rows[0]?.action, 'content.import.questions.completed', 'worker import should write completion audit');
|
|
|
|
console.log('Import worker integration test complete.');
|
|
} finally {
|
|
await cleanup(pool).catch(() => {});
|
|
await pool.end();
|
|
}
|
|
}
|
|
|
|
main().catch(error => {
|
|
console.error(error);
|
|
process.exit(1);
|
|
});
|