MikrotikManager/backend/src/services/traffic-flow-ingest.test.ts

import assert from "node:assert/strict"
import {
  markIfaceRefreshAttempt,
  rememberServerIfaces,
  resetIfaceCacheForTests,
  shouldRefreshIfaces,
} from "./traffic-flow-ifindex.js"
import {
  applyHeartbeatForTests,
  flushPendingForTests,
  getFlowListenerState,
  getFlowRuntimeCounters,
  getFlowWorkerHealth,
  ingestParsedFlowsForServerForTests,
  lastFlushUsedTransactionForTests,
  maybeRefreshIfaces,
  peekPendingFlows,
  capFlowRowsPerServerBucket,
  resetFlowRingsForTests,
  setPendingCapForTests,
  setRefreshIfacesForTests,
  setWantListenForTests,
  simulateWorkerExitForTests,
} from "./traffic-flow-ingest.js"
import { configureEngine, droppedForTests, getEngineStats, isValidFlowInet, pendingSizeForTests } from "./traffic-flow-engine.js"
import { dbQuery } from "../db/index.js"
import { withPgOrSkip } from "../test/pg.js"

resetIfaceCacheForTests()
resetFlowRingsForTests()

let refreshCalls = 0
setRefreshIfacesForTests(async () => {
  refreshCalls += 1
})

rememberServerIfaces(1, [{ ".id": "*A", name: "wg-flow" }])
assert.equal(shouldRefreshIfaces(1), false)
assert.equal(maybeRefreshIfaces(1), false)
assert.equal(refreshCalls, 0)

resetIfaceCacheForTests()
assert.equal(shouldRefreshIfaces(2), true)
assert.equal(maybeRefreshIfaces(2), true)
assert.equal(refreshCalls, 1)

markIfaceRefreshAttempt(2)
assert.equal(shouldRefreshIfaces(2), false)
assert.equal(maybeRefreshIfaces(2), false)
assert.equal(refreshCalls, 1)

assert.equal(lastFlushUsedTransactionForTests(), false)

resetFlowRingsForTests()
setPendingCapForTests(3)
const many = Array.from({ length: 6 }, (_, i) => ({
  src: `10.1.1.${i + 1}`,
  dst: "8.8.8.8",
  proto: 6,
  srcPort: 50000 + i,
  dstPort: 443,
  bytes: 1000,
  packets: 1,
  inIface: "2",
  outIface: "",
}))
ingestParsedFlowsForServerForTests(9, many)
assert.equal(pendingSizeForTests(), 3)
assert.equal(droppedForTests(), 3)
assert.equal(peekPendingFlows().length, 3)
setPendingCapForTests(null)

assert.equal(isValidFlowInet("10.0.0.1"), true)
assert.equal(isValidFlowInet("8.8.8.8"), true)
assert.equal(isValidFlowInet("8.8.8.8/32"), true)
assert.equal(isValidFlowInet("0:0:0:0:0:0:0:1"), true)
assert.equal(isValidFlowInet("not-an-ip"), false)
assert.equal(isValidFlowInet("999.1.1.1"), false)

resetFlowRingsForTests()
ingestParsedFlowsForServerForTests(1, [{
  src: "not-an-ip",
  dst: "8.8.8.8",
  proto: 6,
  srcPort: 1,
  dstPort: 443,
  bytes: 10,
  packets: 1,
  inIface: "2",
  outIface: "",
}])
await flushPendingForTests()
assert.match(getEngineStats().lastError, /невалидн/)

resetFlowRingsForTests()
configureEngine({ topN: 20 })
const talkers = Array.from({ length: 25 }, (_, i) => ({
  src: `10.2.1.${i + 1}`,
  dst: "1.1.1.1",
  proto: 6,
  srcPort: 40000 + i,
  dstPort: 443,
  bytes: 1000 + i,
  packets: 1,
  inIface: "2",
  outIface: "",
}))
if (await withPgOrSkip()) {
  await dbQuery(`
    INSERT INTO servers (id, name, host) VALUES (91009, 'flow-ingest-test', '127.0.0.1')
    ON CONFLICT (id) DO NOTHING
  `)
  ingestParsedFlowsForServerForTests(91009, talkers)
  await flushPendingForTests()
  const stored = await dbQuery<{ n: number }>(
    `SELECT COUNT(*)::int AS n FROM flow_buckets WHERE server_id = 91009`,
  )
  assert.ok((stored.rows[0]?.n ?? 99) <= 20, `expected topN cap, got ${stored.rows[0]?.n}`)
  await dbQuery(`DELETE FROM flow_buckets WHERE server_id = 91009`)
  await dbQuery(`DELETE FROM flow_minute_stats WHERE server_id = 91009`)
  await dbQuery(`DELETE FROM flow_minute_dims WHERE server_id = 91009`)
  await dbQuery(`DELETE FROM flow_daily_dims WHERE server_id = 91009`)
  await dbQuery(`DELETE FROM servers WHERE id = 91009`)
}

applyHeartbeatForTests({
  bound: true,
  address: "127.0.0.1:4739",
  packetsReceived: 42,
  lastExporterIp: "10.255.254.3",
  lastError: "",
  lastDatagramAt: new Date().toISOString(),
  pendingSize: 1,
  dropped: 0,
  rowsStored: 1,
  workerAlive: true,
  rings: [],
})
assert.equal(getFlowListenerState().bound, true)
assert.equal((await getFlowRuntimeCounters()).packetsReceived, 42)
assert.equal(getFlowWorkerHealth().alive, false)
setWantListenForTests(true)
assert.equal(simulateWorkerExitForTests(), 1)
assert.equal(getFlowListenerState().bound, false)
setWantListenForTests(false)

resetFlowRingsForTests()
resetIfaceCacheForTests()
setRefreshIfacesForTests(null)

{
  const minute0 = "2026-01-01T00:00:00.000Z"
  const minute1 = "2026-01-01T00:01:00.000Z"
  const rows: Array<{ serverId: number; bucketAt: string; bytes: number; id: string }> = []
  for (let i = 1; i <= 25; i++) {
    rows.push({ serverId: 1, bucketAt: minute0, bytes: i, id: `a${i}` })
    rows.push({ serverId: 2, bucketAt: minute0, bytes: i, id: `b${i}` })
  }
  rows.push({ serverId: 1, bucketAt: minute1, bytes: 1, id: "a-min-other-minute" })
  const capped = capFlowRowsPerServerBucket(rows, 20)
  assert.equal(capped.filter((r) => r.serverId === 1 && r.bucketAt === minute0).length, 20)
  assert.equal(capped.filter((r) => r.serverId === 2).length, 20)
  assert.ok(capped.some((r) => r.id === "a-min-other-minute"), "другая минута не режется глобальным top-N")
  assert.ok(!capped.some((r) => r.id === "a1" || r.id === "b1"), "мелкие 5-tuple сервера выпадают только в своём bucket")
}

console.log("traffic-flow-ingest.test.ts: ok")