atlas-seller-kit/apps/worker/test/platform-catalog-scheduler.test.mjs

import assert from 'node:assert/strict';
import { test } from 'node:test';
import { PlatformCatalogScheduler } from '../src/platform-catalog-scheduler.ts';

function schedulerDatabase(handler) {
    const statements = [];
    const query = async (sql, parameters = []) => {
        statements.push({ sql, parameters });
        return handler(sql, parameters);
    };
    return {
        statements,
        database: {
            query: async (sql, parameters = []) => {
                statements.push({ sql, parameters });
                if (sql.includes('SELECT source.platform_id')) {
                    return { rows: [{ platform_id: 'platform-1' }], rowCount: 1 };
                }
                return handler(sql, parameters);
            },
            transaction: async (work) => work({ query }),
        },
    };
}

test('没有管理员批次时只检查每周类目树，不自动创建属性或字典任务', async () => {
    const { database, statements } = schedulerDatabase(async (sql) => {
        if (sql.includes('SELECT revision FROM atlas.platform_catalog_sources')) {
            return { rows: [{ revision: 1 }], rowCount: 1 };
        }
        if (sql.includes("count(*)::text AS count FROM atlas.platform_catalog_sync_runs")) {
            return { rows: [{ count: '0' }], rowCount: 1 };
        }
        if (sql.includes('AS due')) return { rows: [{ due: false }], rowCount: 1 };
        if (sql.includes('SELECT item.batch_id')) return { rows: [], rowCount: 0 };
        return { rows: [], rowCount: 0 };
    });

    await new PlatformCatalogScheduler(database).run();

    assert.equal(
        statements.some(({ sql }) => sql.includes('INSERT INTO atlas.platform_catalog_sync_runs')),
        false,
    );
    assert.equal(
        statements.some(({ sql }) => sql.includes("scope = 'dictionary'")),
        false,
    );
});

test('到期类目树创建每周低优先级任务', async () => {
    const { database, statements } = schedulerDatabase(async (sql) => {
        if (sql.includes('SELECT revision FROM atlas.platform_catalog_sources')) {
            return { rows: [{ revision: 3 }], rowCount: 1 };
        }
        if (sql.includes("count(*)::text AS count FROM atlas.platform_catalog_sync_runs")) {
            return { rows: [{ count: '0' }], rowCount: 1 };
        }
        if (sql.includes('AS due')) return { rows: [{ due: true }], rowCount: 1 };
        if (sql.includes('SELECT id, priority FROM atlas.platform_catalog_sync_runs')) {
            return { rows: [], rowCount: 0 };
        }
        if (sql.includes('INSERT INTO atlas.platform_catalog_sync_runs')) {
            return { rows: [{ id: 'run-tree' }], rowCount: 1 };
        }
        if (sql.includes('SELECT item.batch_id')) return { rows: [], rowCount: 0 };
        return { rows: [], rowCount: 0 };
    });

    await new PlatformCatalogScheduler(database).run();

    const insert = statements.find(({ sql }) =>
        sql.includes('INSERT INTO atlas.platform_catalog_sync_runs'));
    assert.ok(insert);
    assert.deepEqual(insert.parameters, [
        'platform-1', 'tree', 'tree', null, 'weekly', 'low', 3,
    ]);
});

test('管理员批次按资源年龄创建高优先级类型属性任务', async () => {
    const { database, statements } = schedulerDatabase(async (sql) => {
        if (sql.includes('SELECT revision FROM atlas.platform_catalog_sources')) {
            return { rows: [{ revision: 4 }], rowCount: 1 };
        }
        if (sql.includes("count(*)::text AS count FROM atlas.platform_catalog_sync_runs")) {
            return { rows: [{ count: '0' }], rowCount: 1 };
        }
        if (sql.includes('AS due')) return { rows: [{ due: false }], rowCount: 1 };
        if (sql.includes('SELECT item.batch_id')) {
            return {
                rows: [{
                    batch_id: 'batch-1',
                    category_id: 'category-1',
                    source_revision: 4,
                    priority: 'high',
                }],
                rowCount: 1,
            };
        }
        if (sql.includes('SELECT id, priority FROM atlas.platform_catalog_sync_runs')) {
            return { rows: [], rowCount: 0 };
        }
        if (sql.includes('INSERT INTO atlas.platform_catalog_sync_runs')) {
            return { rows: [{ id: 'run-attribute' }], rowCount: 1 };
        }
        return { rows: [], rowCount: 0 };
    });

    await new PlatformCatalogScheduler(database).run();

    const insert = statements.find(({ sql }) =>
        sql.includes('INSERT INTO atlas.platform_catalog_sync_runs'));
    assert.ok(insert);
    assert.deepEqual(insert.parameters, [
        'platform-1', 'attributes', 'attributes:category-1', 'category-1',
        'admin_batch', 'high', 4,
    ]);
    assert.ok(statements.some(({ sql }) =>
        sql.includes('UPDATE atlas.platform_catalog_attribute_batch_items')));
});