184 lines
13 KiB
JavaScript
184 lines
13 KiB
JavaScript
const test = require('node:test');
|
||
const assert = require('node:assert/strict');
|
||
const fs = require('node:fs/promises');
|
||
const path = require('node:path');
|
||
const { Pool } = require('pg');
|
||
const { ScheduleImportService } = require('../src/services/scheduleImportService');
|
||
const { getLocalDateTime } = require('../src/utils/etaTime');
|
||
const { parseSchedule, CATALOG_URL, HOME_URL } = require('../src/services/scheduleSource');
|
||
|
||
// Opt-in only. Use a disposable database; every run gets its own temporary schema.
|
||
test('PostgreSQL import lifecycle, concurrency, conflicts, rollback and news review', {
|
||
skip: !process.env.IMPORT_TEST_DATABASE_URL,
|
||
}, async () => {
|
||
const url = process.env.IMPORT_TEST_DATABASE_URL;
|
||
assert.match(new URL(url).pathname, /test/i, 'Use a dedicated test database');
|
||
const admin = new Pool({ connectionString: url });
|
||
const schema = `import_test_${Date.now()}`;
|
||
let pool;
|
||
try {
|
||
await admin.query('CREATE EXTENSION IF NOT EXISTS postgis');
|
||
await admin.query(`CREATE SCHEMA ${schema}`);
|
||
pool = new Pool({ connectionString: url, options: `-c search_path=${schema},public` });
|
||
for (const file of (await fs.readdir(path.join(__dirname, '../migrations'))).filter(f => f.endsWith('.sql')).sort()) {
|
||
await pool.query(await fs.readFile(path.join(__dirname, '../migrations', file), 'utf8'));
|
||
}
|
||
const fixture = name => fs.readFile(path.join(__dirname, 'fixtures/schedule-import', `${name}.html`), 'utf8');
|
||
let sourceHtml = await fixture('route1');
|
||
let newsHtml = await fixture('news');
|
||
const catalogHtml = await fixture('catalog');
|
||
const sourceUrl = 'https://ap2polotsk.of.by/ap2/m-1/';
|
||
const load = async url => ({ url, html: url === CATALOG_URL ? catalogHtml : url === HOME_URL ? newsHtml : sourceHtml });
|
||
const invalidations = [];
|
||
const service = new ScheduleImportService({ pool, load, invalidate: async id => invalidations.push(id) });
|
||
const source = parseSchedule(sourceHtml, sourceUrl);
|
||
sourceHtml = sourceHtml.replace('05:20', `05:20 через «${source.directions[0].stops[1]}»`)
|
||
.replace('06:30', '06:30 по пятницам').replace('06:50', '06:50 кроме пятницы')
|
||
.replace('07:30', '07:30 через «Нет в базе»');
|
||
const routeIds = [];
|
||
for (const direction of source.directions) {
|
||
const routeId = (await pool.query("INSERT INTO routes (route_number,name,type) VALUES ('1',$1,'bus') RETURNING id", [direction.stops[0] + ' → ' + direction.stops.at(-1)])).rows[0].id;
|
||
routeIds.push(routeId);
|
||
for (const [i, name] of direction.stops.entries()) {
|
||
const stopId = (await pool.query("INSERT INTO stops (name,location) VALUES ($1,ST_GeogFromText('POINT(28.77 55.48)')) RETURNING id", [name])).rows[0].id;
|
||
await pool.query('INSERT INTO route_stops (route_id,stop_id,sequence,time_offset_minutes) VALUES ($1,$2,$3,$4)', [routeId, stopId, i + 1, i * 2]);
|
||
}
|
||
}
|
||
const routeId = routeIds[0];
|
||
const validFrom = getLocalDateTime().date;
|
||
const prepare = async () => {
|
||
const inspection = await service.inspect({ routeIds, sourceUrl, userId: 1 });
|
||
return service.preview({ inspectionId: inspection.id, mappings: inspection.mappings, validFrom, userId: 1 });
|
||
};
|
||
|
||
const single = await service.inspect({ routeIds: [routeId], sourceUrl, userId: 1 });
|
||
assert.equal(single.directions.length, 1);
|
||
assert.equal(single.mappings[routeId], 'Швейная фабрика');
|
||
await assert.rejects(service.preview({ inspectionId: single.id, mappings: { ...single.mappings, [routeIds[1]]: 'Богатырская' }, validFrom, userId: 1 }), /каждое выбранное/);
|
||
await assert.rejects(service.inspect({ routeIds: [], sourceUrl, userId: 1 }), /направления/);
|
||
await assert.rejects(service.inspect({ routeIds: [routeId, routeId], sourceUrl, userId: 1 }), /направления/);
|
||
const other = (await pool.query("INSERT INTO routes (route_number,name,type) VALUES ('99','Другой маршрут','bus') RETURNING id")).rows[0].id;
|
||
await assert.rejects(service.inspect({ routeIds: [routeId, other], sourceUrl, userId: 1 }), /тот же номер/);
|
||
const first = await prepare();
|
||
assert.equal(first.hasChanges, true);
|
||
assert.deepEqual(first.plans.map(p => p.changes.map(c => c.after.length)), [[18, 14, 14], [18, 14, 14]]);
|
||
assert.deepEqual(first.plans.map(p => p.sourceStopName), ['Швейная фабрика', 'Богатырская']);
|
||
const concurrent = await Promise.all([service.apply(first.id), service.apply(first.id)]);
|
||
assert.equal(concurrent.filter(r => r.alreadyApplied).length, 1);
|
||
assert.equal(Number((await pool.query('SELECT count(*) FROM schedules')).rows[0].count), 6);
|
||
assert.deepEqual(invalidations.sort((a,b) => a-b), routeIds);
|
||
const savedTrips = (await pool.query("SELECT trips FROM schedules WHERE route_id=$1 AND day_type='weekday'", [routeId])).rows[0].trips;
|
||
assert.equal(savedTrips.find(t => t.departureTime === '05:20:00').viaStopIds.length, 1);
|
||
assert.deepEqual(savedTrips.find(t => t.departureTime === '06:30:00').daysOfWeek, [5]);
|
||
assert.deepEqual(savedTrips.find(t => t.departureTime === '06:50:00').exceptDaysOfWeek, [5]);
|
||
assert.equal(savedTrips.find(t => t.departureTime === '07:30:00').note, 'через «Нет в базе»');
|
||
assert.equal((await service.history()).length, 1);
|
||
assert.equal((await service.history())[0].directions.length, 2);
|
||
assert.equal((await prepare()).hasChanges, false);
|
||
assert.equal((await service.apply(first.id)).alreadyApplied, true);
|
||
assert.equal(Number((await pool.query("SELECT count(*) FROM sync_log WHERE table_name='schedules'")).rows[0].count), 6);
|
||
|
||
sourceHtml = sourceHtml.replace('05:20', '05:21');
|
||
const changed = await prepare();
|
||
sourceHtml = sourceHtml.replace('05:21', '05:22');
|
||
await assert.rejects(service.apply(changed.id), /на сайте изменилось/);
|
||
const stale = await prepare();
|
||
await pool.query("UPDATE schedules SET trips = jsonb_set(trips, '{0,note}', '\"Ручная правка\"'::jsonb) WHERE day_type = 'weekday'");
|
||
const beforeConflict = JSON.stringify((await pool.query('SELECT * FROM schedules ORDER BY id')).rows);
|
||
await assert.rejects(service.apply(stale.id), /после предпросмотра/);
|
||
assert.equal(JSON.stringify((await pool.query('SELECT * FROM schedules ORDER BY id')).rows), beforeConflict);
|
||
|
||
// Failure while writing the second direction must also roll back the first direction.
|
||
sourceHtml = sourceHtml.replace('05:40', '05:41');
|
||
const toFail = await prepare();
|
||
const beforeFailure = JSON.stringify((await pool.query('SELECT * FROM schedules ORDER BY id')).rows);
|
||
const failingPool = { query: (...args) => pool.query(...args), connect: async () => {
|
||
const client = await pool.connect();
|
||
let updates = 0;
|
||
return { release: () => client.release(), query: (sql, args) => {
|
||
if (sql.startsWith('UPDATE schedules SET departure_times') && ++updates === 2) throw new Error('Injected failure');
|
||
return client.query(sql, args);
|
||
} };
|
||
} };
|
||
await assert.rejects(new ScheduleImportService({ pool: failingPool, load }).apply(toFail.id), /Injected failure/);
|
||
assert.equal(JSON.stringify((await pool.query('SELECT * FROM schedules ORDER BY id')).rows), beforeFailure);
|
||
assert.equal((await service.getDraft(toFail.id)).status, 'ready');
|
||
await service.apply(toFail.id);
|
||
|
||
const reverseBefore = JSON.stringify((await pool.query('SELECT * FROM schedules WHERE route_id=$1 ORDER BY id', [routeIds[1]])).rows);
|
||
await pool.query("UPDATE schedules SET trips = jsonb_set(trips, '{0,note}', '\"Правка только прямого направления\"'::jsonb) WHERE day_type = 'weekday' AND route_id=$1", [routeId]);
|
||
const oneInspection = await service.inspect({ routeIds: [routeId], sourceUrl, userId: 1 });
|
||
const onePreview = await service.preview({ inspectionId: oneInspection.id, mappings: oneInspection.mappings, validFrom, userId: 1 });
|
||
assert.equal(onePreview.plans.length, 1);
|
||
assert.equal(onePreview.hasChanges, true);
|
||
await service.apply(onePreview.id);
|
||
assert.equal(JSON.stringify((await pool.query('SELECT * FROM schedules WHERE route_id=$1 ORDER BY id', [routeIds[1]])).rows), reverseBefore);
|
||
|
||
// A future import closes old versions on the preceding day, keeping today's data intact.
|
||
sourceHtml = sourceHtml.replace('05:22', '05:23');
|
||
const inspected = await service.inspect({ routeIds, sourceUrl, userId: 1 });
|
||
const tomorrow = new Date(`${validFrom}T12:00:00Z`); tomorrow.setUTCDate(tomorrow.getUTCDate() + 1);
|
||
const future = await service.preview({ inspectionId: inspected.id, mappings: inspected.mappings, validFrom: tomorrow.toISOString().slice(0, 10), userId: 1 });
|
||
await service.apply(future.id);
|
||
const versions = (await pool.query("SELECT to_char(valid_until,'YYYY-MM-DD') AS until FROM schedules WHERE day_type='weekday' AND route_id = $1 ORDER BY id", [routeId])).rows;
|
||
assert.equal(versions.length, 2);
|
||
assert.equal(versions[0].until, validFrom);
|
||
assert.equal(versions[1].until, null);
|
||
|
||
// Copying the selected version must not choose a newer future version or overwrite history.
|
||
await pool.query(`INSERT INTO schedules (route_id,day_type,departure_times,trips,valid_from,valid_until)
|
||
VALUES ($1,'holiday',ARRAY['04:00'::time],'[]','2020-01-01','2020-01-02')`, [routeId]);
|
||
const vm = require('node:vm');
|
||
const moduleStub = { exports: {} };
|
||
vm.runInNewContext(await fs.readFile(path.join(__dirname, '../src/controllers/schedulesController.js'), 'utf8'), {
|
||
module: moduleStub, require: name => {
|
||
if (name.endsWith('/database')) return pool;
|
||
if (name.endsWith('/redis')) return { del: async () => {} };
|
||
if (name.endsWith('/logger')) return { info() {}, error() {} };
|
||
throw new Error(`Unexpected dependency: ${name}`);
|
||
}
|
||
});
|
||
const selectedId = (await pool.query("SELECT id FROM schedules WHERE day_type='weekday' AND route_id = $1 ORDER BY id LIMIT 1", [routeId])).rows[0].id;
|
||
let responseBody;
|
||
const response = { status(code) { assert.ok(code < 400); return this; }, json(body) { responseBody = body; } };
|
||
await moduleStub.exports.copySchedule({ params: { routeId }, body: { fromDayType: 'weekday', toDayType: 'holiday', sourceScheduleId: selectedId } }, response);
|
||
assert.equal(responseBody.trips.some(t => t.departureTime === '05:22:00'), true);
|
||
assert.equal(responseBody.trips.some(t => t.departureTime === '05:23:00'), false);
|
||
assert.equal(Number((await pool.query("SELECT count(*) FROM schedules WHERE day_type='holiday'")).rows[0].count), 2);
|
||
|
||
// Route 24 uses separate Mon–Thu and Friday tables; persist their restrictions together.
|
||
sourceHtml = await fixture('route24');
|
||
const source24 = parseSchedule(sourceHtml, 'https://ap2polotsk.of.by/ap2/m-24/');
|
||
const ids24 = [];
|
||
for (const direction of source24.directions) {
|
||
const id = (await pool.query("INSERT INTO routes(route_number,name,type) VALUES('24',$1,'bus') RETURNING id", [direction.label])).rows[0].id;
|
||
ids24.push(id);
|
||
for (const [index, name] of [direction.stops[0], direction.stops.at(-1)].entries()) {
|
||
const stop = (await pool.query("INSERT INTO stops(name,location) VALUES($1,ST_GeogFromText('POINT(28.77 55.48)')) RETURNING id", [name])).rows[0].id;
|
||
await pool.query('INSERT INTO route_stops(route_id,stop_id,sequence) VALUES($1,$2,$3)', [id,stop,index+1]);
|
||
}
|
||
}
|
||
const inspection24 = await service.inspect({ routeIds: ids24, sourceUrl: 'https://ap2polotsk.of.by/ap2/m-24/', userId:1 });
|
||
const preview24 = await service.preview({ inspectionId:inspection24.id, mappings:inspection24.mappings, validFrom, userId:1 });
|
||
await service.apply(preview24.id);
|
||
const saved24 = (await pool.query("SELECT trips FROM schedules WHERE route_id=$1 AND day_type='weekday'", [ids24[0]])).rows[0].trips;
|
||
assert.equal(saved24.filter(t => t.departureTime === '06:00:00').length, 1);
|
||
assert.deepEqual(saved24.find(t => t.departureTime === '17:00:00').daysOfWeek, [1,2,3,4]);
|
||
assert.deepEqual(saved24.find(t => t.departureTime === '15:45:00').daysOfWeek, [5]);
|
||
assert.equal(Number((await pool.query('SELECT count(*) FROM schedules WHERE route_id=ANY($1::int[])', [ids24])).rows[0].count), 6);
|
||
|
||
let news = await service.news(true);
|
||
assert.equal(news.length, 3);
|
||
await service.reviewNews({ url: news[0].url, contentHash: news[0].content_hash });
|
||
assert.equal((await service.news(true))[0].reviewed, true);
|
||
newsHtml = newsHtml.replace('08:50', '08:51');
|
||
news = await service.news(true);
|
||
assert.equal(news[0].reviewed, false);
|
||
assert.equal(news.length, 3);
|
||
} finally {
|
||
if (pool) await pool.end();
|
||
await admin.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`);
|
||
await admin.end();
|
||
}
|
||
});
|