MikrotikManager/tests/e2e/traffic-flow.spec.ts

import { installOpenLiveStream, resumeLiveStream, sendLiveHeartbeat } from "./live-stream-fixture"
import { test, expect, type Page } from '@playwright/test'

async function setup(page: Page) {
  const control = { hold: '', holdIface: '', fail: false, failList: false, invalidSample: undefined as unknown, mergeSample: false, metrics: {} as Record<string, unknown>, release: () => {} }
  const gate = new Promise<void>(resolve => { control.release = resolve })
  const card = { id: '1', name: 'Экспортёр тест', subtitle: '', site: 'TEST', country: 'RU', status: 'online', rxNow: 2, txNow: 3, sessions: 1, rxSeries: [1, 2], txSeries: [2, 3], bytes: 200 }
  const data = { bpsNow: 1000000, bytes: 200, packets: 2, conversations: 1, uniqueSrc: 1, uniqueDst: 1, topProto: 'TCP', rxSeries: [1], txSeries: [2], applications: [], protocols: [], sources: [], destinations: [], interfaces: [], conversationsList: [], ifaces: [], live: false }
  await page.route('**/health', r => r.fulfill({ json: { ok: true } }))
  await page.route('**/api/**', async route => {
    const url = new URL(route.request().url()), path = url.pathname, range = url.searchParams.get('range')
    if (path === '/api/traffic/flow/monthly') return route.fulfill({ json: { month: url.searchParams.get('month'), bytes: 1500, countries: [{ id: 'RU', label: 'Россия', bytes: 1500, packets: 1, bps: 0, percent: 100 }], services: [], asns: [] } })
    if (path === '/api/traffic/flow/analytics') {
      if (control.hold === range || (control.holdIface && url.searchParams.get('iface') === control.holdIface)) await gate
      if (control.fail) return route.fulfill({ status: 500, json: { error: 'Срез временно недоступен' } })
      return route.fulfill({ json: { ...data, ...control.metrics, topCategory: url.searchParams.get('serverId') === '2' ? 'Второй сервер' : 'Категория ' + range } })
    }
    if (path === '/api/traffic/flow/live') return route.fulfill({ contentType: 'text/event-stream', body: 'event: sample\ndata: ' + JSON.stringify({ ...data, live: true }) + '\n\n' + (control.invalidSample === undefined ? '' : 'event: sample\ndata: ' + JSON.stringify(control.mergeSample ? { ...data, ...(control.invalidSample as object) } : control.invalidSample) + '\n\n') })
    if (path === '/api/traffic/flow/exporters' && control.failList) return route.fulfill({ status: 403, json: { error: 'Нет доступа к экспортёрам' } })
    const json = path === '/api/auth/config' ? { required: false }
      : path === '/api/traffic/flow/exporters' ? { exporters: [card, { ...card, id: '2', name: 'Экспортёр второй' }] }
      : path === '/api/traffic/flow/clients' ? { clients: [{ ...card, id: 'client1', name: 'Клиент тест' }] }
      : path === '/api/traffic/flows' ? { exportersOnline: 1, bytesPerMin: 1, uniqueSrc: 1, uniqueDst: 1, topProto: 'TCP', talkers: [], listenerBound: true }
      : path === '/api/traffic/servers' ? { servers: [] }
      : path.endsWith('/interfaces') || path.endsWith('/bound-interfaces') ? { interfaces: [] }
      : path === '/api/traffic/users' ? { users: [] }
      : path === '/api/servers' ? [] : {}
    return route.fulfill({ json })
  })
  await page.goto('/traffic')
  await page.getByRole('button', { name: 'Потоки', exact: true }).click()
  await expect(page.getByLabel('Скорость потоков')).toBeVisible()
  return control
}

test('Flow: поздний ответ периода не заменяет текущую аналитику', async ({ page }) => {
  const control = await setup(page)
  control.hold = '24h'
  await page.getByRole('button', { name: '24ч', exact: true }).click()
  await expect(page.getByText('Загрузка аналитики…', { exact: true })).toBeVisible()
  await expect(page.getByLabel('Скорость потоков')).toHaveCount(0)
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(page.getByText('Категория 15m', { exact: true }).filter({ visible: true })).toBeVisible()
  const response = page.waitForResponse(r => r.url().includes('/flow/analytics?') && r.url().includes('range=24h'))
  control.release()
  await response
  await expect(page.getByText('Категория 15m', { exact: true }).filter({ visible: true })).toBeVisible()
  await expect(page.getByText('Категория 24h', { exact: true })).toHaveCount(0)
})

test('Flow: ошибка аналитики и повтор сохраняют период', async ({ page }) => {
  const control = await setup(page)
  control.fail = true
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(page.getByRole('alert').filter({ hasText: 'Аналитика недоступна' })).toContainText('Срез временно недоступен')
  await expect(page.getByLabel('Скорость потоков')).toHaveCount(0)
  await expect(page.getByLabel('Сводка трафика').getByText('—', { exact: true })).toHaveCount(3)
  control.fail = false
  await page.getByRole('button', { name: 'Повторить аналитику', exact: true }).click()
  await expect(page.getByText('Категория 15m', { exact: true }).filter({ visible: true })).toBeVisible()
})

test('Flow: ошибка списка не отображается как отсутствие экспортёров', async ({ page }) => {
  const control = await setup(page)
  control.failList = true
  await page.getByRole('button', { name: 'Обновить', exact: true }).click()
  await expect(page.getByRole('alert').filter({ hasText: 'Потоки недоступны' })).toContainText('Нет доступа')
  await expect(page.getByLabel('Сводка трафика')).toHaveCount(0)
  await expect(page.getByLabel('Получение аналитики потоков')).toHaveCount(0)
  await expect(page.getByLabel('Получение списка потоков')).toHaveCount(0)
  await expect(page.getByText('Нет экспортёров IPFIX.', { exact: false })).toHaveCount(0)
  control.failList = false
  await page.getByRole('button', { name: 'Повторить загрузку потоков', exact: true }).click()
  await expect(page.getByLabel('Скорость потоков')).toBeVisible()
})

test('Flow: завершённый SSE не помечает последние данные как live', async ({ page }) => {
  await setup(page)
  await page.getByRole('button', { name: '5м', exact: true }).click()
  await expect(page.getByText('Последние данные · live недоступен', { exact: true }).filter({ visible: true })).toHaveCount(2)
  await expect(page.getByText('live', { exact: true })).toHaveCount(0)
})


test('Flow: месяц показывает только доступные агрегаты и не применяется к клиенту', async ({ page }) => {
  await setup(page)
  const monthlyRequest = page.waitForRequest(r => r.url().includes('/flow/monthly?'))
  await page.getByRole('button', { name: 'Текущий месяц', exact: true }).click()
  const url = new URL((await monthlyRequest).url())
  expect(url.searchParams.get('serverId')).toBe('1')
  expect(url.searchParams.get('month')).toMatch(/^\d{4}-\d{2}$/)
  expect([...url.searchParams.keys()].sort()).toEqual(['month', 'serverId'])
  const report = page.getByRole('region', { name: 'Месячный отчёт' })
  await expect(report).toContainText('1.5 КБ')
  await expect(report.getByRole('columnheader').filter({ hasText: 'Скорость' })).toHaveCount(0)
  await expect(page.getByLabel('Скорость потоков')).toHaveCount(0)
  await expect(page.getByLabel('Сводка трафика')).toHaveCount(0)
  await expect(page.getByRole('switch', { name: 'Без дублей' })).toBeDisabled()
  await expect(page.getByRole('switch', { name: 'Без overlay' })).toBeDisabled()
  await expect(page.getByRole('switch', { name: 'Без mesh' })).toBeDisabled()
  const clientRequest = page.waitForRequest(r => r.url().includes('/flow/analytics?') && r.url().includes('userId=client1'))
  await page.getByRole('group', { name: 'Источник потоков' }).getByRole('button', { name: 'Клиенты', exact: true }).click()
  expect(new URL((await clientRequest).url()).searchParams.get('range')).toBe('1h')
  await expect(page.getByLabel('Скорость потоков')).toBeVisible()
  await expect(report).toHaveCount(0)
  await expect(page.getByRole('button', { name: 'Текущий месяц', exact: true })).toHaveCount(0)
  await expect(page.getByRole('switch', { name: 'Без дублей' })).toBeEnabled()
})

for (const width of [320, 390, 1440]) {
  test(`Flow: адаптивность ${width}px`, async ({ page }) => {
    await page.setViewportSize({ width, height: 900 })
    await setup(page)
    for (const monthly of [false, true]) {
      if (monthly) {
        await page.getByRole('button', { name: 'Текущий месяц', exact: true }).click()
        await expect(page.getByRole('region', { name: 'Месячный отчёт' })).toBeVisible()
      }
      const button = page.getByRole('button', { name: 'Текущий месяц', exact: true })
      const box = await button.boundingBox()
      expect(box).not.toBeNull()
      expect(box!.x).toBeGreaterThanOrEqual(0)
      expect(box!.x + box!.width).toBeLessThanOrEqual(width + 1)
      const overflow = await page.evaluate(() => document.documentElement.scrollWidth > window.innerWidth + 1)
      expect(overflow).toBe(false)
    }
    await page.screenshot({ path: `test-results/flow-monthly-${width}.png`, fullPage: true })
  })
}


for (const [name, invalid, merge] of [
  ['неполный объект', { uniqueSrc: 1, destinations: [] }, false],
  ['отрицательная скорость', { bpsNow: -1 }, true],
  ['null', null, false],
  ['неверная строка таблицы', { destinations: [null] }, true],
  ['отрицательная серия', { rxSeries: [-1] }, true],
] as const) {
  test('Flow: повреждённый sample — ' + name, async ({ page }) => {
    const control = await setup(page)
    await page.getByRole("button", { name: "15м", exact: true }).click()
    await expect(page.getByText("Категория 15m", { exact: true }).filter({ visible: true })).toBeVisible()
    control.invalidSample = invalid
    control.mergeSample = merge
    await page.getByRole('button', { name: '5м', exact: true }).click()
    await expect(page.getByText('Некорректные данные live-потока', { exact: true })).toBeVisible()
    await expect(page.getByText('Последние данные · live недоступен', { exact: true }).filter({ visible: true })).toHaveCount(2)
    await expect(page.getByLabel('Скорость потоков')).toContainText('1.00 Мбит/с')
    await expect(page.getByText('live', { exact: true })).toHaveCount(0)
  })
}


for (const [name, metrics, rate, bytes] of [
  ['отсутствующие', {}, 'Нет данных', 'Нет данных'],
  ['нулевые', { bpsOverlay: 0, bpsWire: 0, bytesMesh: 0, bytesOverlay: 0 }, '0 Кбит/с', '0 Б'],
  ['положительные', { bpsOverlay: 2000000, bpsWire: 2000000, bytesMesh: 1500, bytesOverlay: 1500 }, '2.00 Мбит/с', '1.5 КБ'],
  ['некорректные REST', { bpsOverlay: -1, bpsWire: null, bytesMesh: -1 }, 'Нет данных', 'Нет данных'],
] as const) {
  test('Flow: optional-метрики — ' + name, async ({ page }) => {
    const control = await setup(page)
    control.metrics = metrics
    await page.getByRole('button', { name: '15м', exact: true }).click()
    await page.getByRole('switch', { name: 'Без mesh', exact: true }).click()
    const grid = page.getByLabel('Скорость потоков')
    const card = (label: string) => grid.locator('[data-slot="frame-panel"]').filter({ has: page.getByText(label, { exact: true }) })
    await expect(card('JH↔EN overlay')).toContainText(rate)
    await expect(card('Wire GRE')).toContainText(rate)
    await expect(card('Mesh')).toContainText(bytes)
    if (name === 'нулевые') await expect(card('JH↔EN overlay')).toContainText('0 Б')
  })
}


test('Flow: интерфейс прежнего сервера не передаётся новому; поздний ответ игнорируется', async ({ page }) => {
  const control = await setup(page)
  control.metrics = { ifaces: [{ name: 'ether1', index: '1', bps: 1 }] }
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(page.getByRole('button', { name: 'ether1', exact: true })).toBeVisible()
  control.holdIface = 'ether1'
  const oldRequest = page.waitForRequest(r => r.url().includes('/flow/analytics?') && r.url().includes('iface=ether1'))
  await page.getByRole('button', { name: 'ether1', exact: true }).click()
  await oldRequest
  const requests: URL[] = []
  page.on('request', r => { if (r.url().includes('/flow/analytics?')) requests.push(new URL(r.url())) })
  await page.getByRole('button').filter({ hasText: 'Экспортёр второй' }).click()
  await expect(page.getByText('Второй сервер', { exact: true }).filter({ visible: true })).toBeVisible()
  const newRequests = requests.filter(url => url.searchParams.get('serverId') === '2')
  expect(newRequests.length).toBeGreaterThan(0)
  expect(newRequests.every(url => !url.searchParams.has('iface'))).toBe(true)
  const late = page.waitForResponse(r => r.url().includes('/flow/analytics?') && r.url().includes('iface=ether1'))
  control.release()
  await late
  await expect(page.getByText('Второй сервер', { exact: true }).filter({ visible: true })).toBeVisible()
  await expect(page.getByText('Категория 15m', { exact: true })).toHaveCount(0)
})

test('Flow: смена сервера отменяет старую SSE-подписку', async ({ page }) => {
  await setup(page)
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(page.getByText('Категория 15m', { exact: true }).filter({ visible: true })).toBeVisible()
  let release = () => {}
  const gate = new Promise<void>(resolve => { release = resolve })
  let completed = () => {}
  const delivered = new Promise<void>(resolve => { completed = resolve })
  await page.route('**/api/traffic/flow/live?**', async route => {
    const old = new URL(route.request().url()).searchParams.get('serverId') === '1'
    if (old) await gate
    const sample = { bpsNow: old ? 99000000 : 7000000, bytes: 200, packets: 2, conversations: 1, uniqueSrc: 1, uniqueDst: 1, topProto: 'TCP', topCategory: old ? 'SSE прежнего сервера' : 'SSE текущего сервера', rxSeries: [1], txSeries: [2], applications: [], protocols: [], sources: [], destinations: [], interfaces: [], conversationsList: [], ifaces: [], live: true }
    try {
      await route.fulfill({ contentType: 'text/event-stream', body: 'event: sample\ndata: ' + JSON.stringify(sample) + '\n\n' })
    } finally {
      if (old) completed()
    }
  })
  const oldRequest = page.waitForRequest(r => r.url().includes('/flow/live?') && r.url().includes('serverId=1'))
  await page.getByRole('button', { name: '5м', exact: true }).click()
  await oldRequest
  const aborted = page.waitForEvent('requestfailed', r => r.url().includes('/flow/live?') && r.url().includes('serverId=1'))
  await page.getByRole('button').filter({ hasText: 'Экспортёр второй' }).click()
  await aborted
  await expect(page.getByText('SSE текущего сервера', { exact: true }).filter({ visible: true })).toBeVisible()
  release()
  await delivered
  await expect(page.getByText('SSE прежнего сервера', { exact: true })).toHaveCount(0)
  await expect(page.getByText('SSE текущего сервера', { exact: true }).filter({ visible: true })).toBeVisible()
  await expect(page.getByLabel('Скорость потоков')).toContainText('7.00 Мбит/с')
})

test('Flow: позднее чтение открытого SSE не заменяет данные нового сервера', async ({ page }) => {
  await setup(page)
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(page.getByText('Категория 15m', { exact: true }).filter({ visible: true })).toBeVisible()
  // Deliberately let the mock stream deliver a chunk after abort. This exercises
  // the post-read guard, independently of the browser transport's cancellation.
  await page.evaluate(() => {
    const originalFetch = window.fetch.bind(window)
    const state = { pending: false, aborted: false, delivered: false, release: () => {} }
    Object.assign(window, { delayedFlowStream: state })
    window.fetch = async (input, init) => {
      const url = new URL(input instanceof Request ? input.url : String(input), location.href)
      if (url.pathname !== '/api/traffic/flow/live') return originalFetch(input, init)
      const old = url.searchParams.get('serverId') === '1'
      const sample = (late = false) => ({
        bpsNow: old ? (late ? 99000000 : 1000000) : 7000000,
        bytes: 200, packets: 2, conversations: 1, uniqueSrc: 1, uniqueDst: 1,
        topProto: 'TCP', topCategory: old ? (late ? 'Поздний фрагмент' : 'Открытый поток первого') : 'Открытый поток второго',
        rxSeries: [1], txSeries: [2], applications: [], protocols: [], sources: [],
        destinations: [], interfaces: [], conversationsList: [], ifaces: [], live: true,
      })
      const encode = (late = false) => new TextEncoder().encode('event: sample\ndata: ' + JSON.stringify(sample(late)) + '\n\n')
      const stream = new ReadableStream<Uint8Array>({
        start(controller) {
          controller.enqueue(encode())
          if (old) state.release = () => controller.enqueue(encode(true))
        },
      })
      const response = new Response(stream, { headers: { 'Content-Type': 'text/event-stream' } })
      if (old) {
        init?.signal?.addEventListener('abort', () => { state.aborted = true })
        const reader = stream.getReader()
        const read = reader.read.bind(reader)
        let calls = 0
        reader.read = async () => {
          const delayed = ++calls === 2
          if (delayed) state.pending = true
          const result = await read()
          if (delayed) { state.pending = false; state.delivered = true }
          return result
        }
        stream.getReader = (() => reader) as typeof stream.getReader
      }
      return response
    }
  })
  await page.getByRole('button', { name: '5м', exact: true }).click()
  await expect(page.getByText('Открытый поток первого', { exact: true }).filter({ visible: true })).toBeVisible()
  const streamState = () => page.evaluate(() => {
    const state = (window as unknown as { delayedFlowStream: { pending: boolean; aborted: boolean; delivered: boolean } }).delayedFlowStream
    return { pending: state.pending, aborted: state.aborted, delivered: state.delivered }
  })
  await expect.poll(streamState).toEqual({ pending: true, aborted: false, delivered: false })
  await page.getByRole('button').filter({ hasText: 'Экспортёр второй' }).click()
  await expect(page.getByText('Открытый поток второго', { exact: true }).filter({ visible: true })).toBeVisible()
  await expect.poll(streamState).toEqual({ pending: true, aborted: true, delivered: false })
  await page.evaluate(() => (window as unknown as { delayedFlowStream: { release: () => void } }).delayedFlowStream.release())
  await expect.poll(streamState).toEqual({ pending: false, aborted: true, delivered: true })
  await expect(page.getByText('Поздний фрагмент', { exact: true })).toHaveCount(0)
  await expect(page.getByText('Открытый поток второго', { exact: true }).filter({ visible: true })).toBeVisible()
  await expect(page.getByLabel('Скорость потоков')).toContainText('7.00 Мбит/с')
})

test('Flow: время получения принадлежит текущему срезу и сохраняется после ошибки SSE', async ({ page }) => {
  const control = await setup(page)
  const receipt = page.getByLabel('Получение аналитики потоков')
  const first = new Date('2026-09-28T06:00:00Z')
  await page.clock.setFixedTime(first)
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(receipt.locator('time')).toHaveAttribute('datetime', first.toISOString())
  await expect(receipt).toContainText('Получено (API)')
  await expect(receipt).toContainText('время измерения не предоставлено')
  control.hold = '24h'
  const started = page.waitForRequest(r => r.url().includes('/flow/analytics?') && r.url().includes('range=24h'))
  await page.getByRole('button', { name: '24ч', exact: true }).click()
  await started
  await expect(receipt).toHaveCount(0)
  const second = new Date('2026-09-28T06:05:00Z')
  await page.clock.setFixedTime(second)
  control.release()
  await expect(receipt.locator('time')).toHaveAttribute('datetime', second.toISOString())
  control.invalidSample = null
  await page.getByRole('button', { name: '5м', exact: true }).click()
  await expect(page.getByText('Некорректные данные live-потока', { exact: true })).toBeVisible()
  await expect(receipt).toContainText('Получено (live-поток)')
  await expect(receipt.locator('time')).toHaveAttribute('datetime', second.toISOString())
  await page.clock.setFixedTime(new Date('2026-09-28T06:10:00Z'))
  await expect(receipt.locator('time')).toHaveAttribute('datetime', second.toISOString())
  control.fail = true
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(page.getByRole('alert').filter({ hasText: 'Аналитика недоступна' })).toBeVisible()
  await expect(receipt).toHaveCount(0)
})


test('Flow: молчащий SSE сохраняет данные и восстанавливает live после sample', async ({ page }) => {
  await installOpenLiveStream(page, '/api/traffic/flow/live', { bpsNow: 1000000, bytes: 200, packets: 2, conversations: 1, uniqueSrc: 1, uniqueDst: 1, topProto: 'TCP', rxSeries: [1], txSeries: [2], applications: [], protocols: [], sources: [], destinations: [], interfaces: [], conversationsList: [], ifaces: [], live: true })
  await setup(page)
  const receipt = page.getByLabel('Получение аналитики потоков').locator('time')
  await expect(page.getByText('live', { exact: true }).filter({ visible: true })).toHaveCount(3)
  const stamp = await receipt.getAttribute('datetime')
  await page.clock.fastForward(10_000)
      await sendLiveHeartbeat(page)
      await page.clock.fastForward(5_001)
  await expect(page.getByText('Нет обновлений live более 15 секунд', { exact: true })).toBeVisible()
  await expect(page.getByText('live', { exact: true })).toHaveCount(0)
  await expect(page.getByLabel('Скорость потоков')).toContainText('1.00 Мбит/с')
  await expect(receipt).toHaveAttribute('datetime', stamp!)
  await resumeLiveStream(page)
  await expect(page.getByText('live', { exact: true }).filter({ visible: true })).toHaveCount(3)
  await expect(page.getByText('Нет обновлений live более 15 секунд', { exact: true })).toHaveCount(0)
  await page.getByRole('button', { name: '15м', exact: true }).click()
  await expect(page.getByText('Категория 15m', { exact: true }).filter({ visible: true })).toBeVisible()
  await page.clock.fastForward(15_001)
  await expect(page.getByText('Нет обновлений live более 15 секунд', { exact: true })).toHaveCount(0)
})