pipe-ner-server/pipe-network-service/sql/tests/alarm_review.test.mjs

// PostgreSQL 兼容性回归，不替代 KingbaseES 最终验收；不连接实际数据库。
// node --test alarm_review.test.mjs；通过 PGLITE_MODULE 指向已有 PGlite ESM 模块。
import { readFile } from 'node:fs/promises'
import { pathToFileURL } from 'node:url'
import assert from 'node:assert/strict'
import { test, after } from 'node:test'

const { PGlite } = await import(process.env.PGLITE_MODULE ? pathToFileURL(process.env.PGLITE_MODULE).href : '@electric-sql/pglite')
const db = new PGlite()
const sql = await readFile(new URL('../migration_alarm_review.sql', import.meta.url), 'utf8')
await db.exec(`
 CREATE SCHEMA app_user;
 CREATE TABLE app_user.alarm_data(id varchar(64) PRIMARY KEY, alarm_status integer DEFAULT 0, alarm_level integer,
 warning_code text, warning_type text, remark text, actual_value numeric, min_value numeric, max_value numeric,
 create_time timestamp, work_order_id int8);
 CREATE TABLE app_user.work_order_log(log_id int8, work_order_id int8, oper_desc text, attach_url text,
 oper_user_id int8, oper_user_name text, oper_time timestamp);
 CREATE TABLE app_user.sys_menu(menu_id int8 PRIMARY KEY, menu_name text, parent_id int8, order_num integer,
 path text, component text, query text, is_frame integer, is_cache integer, menu_type text, visible text,
 status text, perms text, icon text, create_by text, create_time timestamp, update_time timestamp);
 CREATE TABLE app_user.sys_role_menu(role_id int8, menu_id int8, PRIMARY KEY(role_id, menu_id));
 INSERT INTO app_user.alarm_data(id,alarm_level) VALUES ('history',1);
 INSERT INTO app_user.sys_menu(menu_id,parent_id,path,component) VALUES
 (2118,100,'audit','water'),(2142,101,'bjsh','drainage'),(2097,102,'krqildjc','gas'),(2024,103,'alarmList','manhole'),(3000,104,'keep','keep');
 INSERT INTO app_user.sys_role_menu VALUES (9,2118),(9,2142),(9,2097),(9,2024);
`)
await db.exec(sql)
const rows = async query => (await db.query(query)).rows
after(() => db.close())

test('int8 primary key uses dedicated owned sequence, no auxiliary source key or history backfill', async () => {
  const [column] = await rows("SELECT udt_name,column_default FROM information_schema.columns WHERE table_schema='app_user' AND table_name='alarm_log' AND column_name='log_id'")
  assert.equal(column.udt_name, 'int8'); assert.match(column.column_default, /nextval\('app_user.alarm_log_id_seq'::regclass\)/)
  assert.equal((await rows("SELECT pg_get_serial_sequence('app_user.alarm_log','log_id') AS seq"))[0].seq, 'app_user.alarm_log_id_seq')
  assert.equal((await rows('SELECT * FROM app_user.alarm_log')).length, 0)
  assert.equal((await rows("SELECT column_name FROM information_schema.columns WHERE table_schema='app_user' AND column_name='source_key'")).length, 0)
  assert.equal((await rows("SELECT audit_status,alarm_level FROM app_user.alarm_data WHERE id='history'"))[0].alarm_level, 1)
})
test('four existing menu IDs get alarmAudit and exact independent scopes, unrelated menu untouched', async () => {
  const menus = await rows('SELECT * FROM app_user.sys_menu WHERE menu_id IN (2118,2142,2097,2024) ORDER BY menu_id')
  const scope = { 2118:{rootTypeId:1}, 2142:{rootTypeId:2}, 2097:{rootTypeId:3,warning_code:'WARN-GAS-CONCENTRATION'}, 2024:{rootTypeId:4} }
  for (const menu of menus) {
    assert.equal(menu.path, 'alarmAudit'); assert.equal(menu.component, 'subSystem/alarm/common/audit')
    assert.deepEqual(JSON.parse(menu.query), scope[menu.menu_id]); assert.equal(menu.perms,'alarm:review:list')
  }
  assert.equal((await rows('SELECT path FROM app_user.sys_menu WHERE menu_id=3000'))[0].path,'keep')
  assert.equal((await rows("SELECT * FROM app_user.sys_menu WHERE perms='alarm:review:audit'")).length,4)
})
test('repeated deployment preserves existing logs, IDs and intentional permission revocations', async () => {
  const oldLogs = await rows('SELECT log_id FROM app_user.alarm_log ORDER BY log_id')
  await db.exec("DELETE FROM app_user.sys_role_menu WHERE menu_id IN (SELECT menu_id FROM app_user.sys_menu WHERE parent_id=2118 AND perms='alarm:review:audit')")
  await db.exec(sql)
  assert.deepEqual(await rows('SELECT log_id FROM app_user.alarm_log ORDER BY log_id'), oldLogs)
  assert.equal((await rows("SELECT * FROM app_user.sys_menu WHERE perms='alarm:review:audit'")).length,4)
  assert.equal((await rows("SELECT * FROM app_user.sys_role_menu WHERE menu_id IN (SELECT menu_id FROM app_user.sys_menu WHERE parent_id=2118 AND perms='alarm:review:audit')")).length,0)
  await db.exec("INSERT INTO app_user.alarm_data(id) VALUES ('a2')")
  assert.equal((await rows("SELECT * FROM app_user.alarm_log WHERE alarm_id='a2'")).length,0)
})
test('transaction rollback also rolls back initialized logs', async () => {
  await db.exec("BEGIN; INSERT INTO app_user.alarm_data(id) VALUES ('rolled-back'); ROLLBACK;")
  assert.equal((await rows("SELECT * FROM app_user.alarm_log WHERE alarm_id='rolled-back'")).length,0)
  await assert.rejects(db.exec("UPDATE app_user.alarm_data SET audit_status=3 WHERE id='a2'"))
})
test('schema contains no stored lifecycle functions or triggers',async()=>{
  assert.equal((await rows("SELECT * FROM pg_trigger WHERE tgname IN ('trg_alarm_lifecycle_log','trg_work_order_alarm_log')")).length,0)
  assert.deepEqual(await rows("SELECT to_regprocedure('app_user.record_alarm_lifecycle()') AS lifecycle,to_regprocedure('app_user.record_alarm_work_order_log()') AS mirror"),[{lifecycle:null,mirror:null}])
})