import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' import { EventEmitter } from 'node:events' /** * Clear Exchanges Flow Integration Test * * Validates the full flow: * 1. Create exchanges for a device, verify they exist * 2. Call DELETE endpoint, verify 204 response * 3. Verify: exchange list is empty * 4. Verify: SSE stream remains active and continues receiving new messages * * Validates: Requirements 8.1, 8.2, 8.3, 8.4, 8.5 */ // --- Hoisted mocks --- const loggerMock = vi.hoisted(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn(), })) vi.mock('@adonisjs/core/services/logger', () => ({ default: loggerMock, })) vi.mock('#start/env', () => ({ default: { get: (_key: string, defaultValue?: any) => defaultValue, }, })) // In-memory store for exchanges const exchangeStore = vi.hoisted(() => { let exchanges: any[] = [] let seqCounter = 0 return { exchanges, getExchanges: () => exchanges, setExchanges: (ex: any[]) => { exchanges.splice(0, exchanges.length, ...ex) }, resetCounter: () => { seqCounter = 0 }, getCounter: () => seqCounter, incrementCounter: () => ++seqCounter, clear: () => { exchanges.splice(0, exchanges.length) seqCounter = 0 }, } }) vi.mock('#models/exchange', () => ({ default: { getCollection: () => ({ insertOne: vi.fn().mockResolvedValue({ insertedId: 'mock-id' }), }), findByDevice: vi.fn().mockImplementation((deviceId: string, options?: any) => { let results = exchangeStore.getExchanges().filter((e: any) => e.deviceId === deviceId) if (options?.after) { results = results.filter((e: any) => e.timestamp > options.after) } results.sort((a: any, b: any) => a.timestamp.localeCompare(b.timestamp)) const limit = options?.limit ?? 200 return Promise.resolve(results.slice(0, limit)) }), findAfterSequence: vi.fn().mockImplementation((deviceId: string, seqNumber: number) => { const results = exchangeStore.getExchanges() .filter((e: any) => e.deviceId === deviceId && e.sequenceNumber > seqNumber) .sort((a: any, b: any) => a.sequenceNumber - b.sequenceNumber) return Promise.resolve(results) }), deleteByDevice: vi.fn().mockImplementation((deviceId: string) => { const before = exchangeStore.getExchanges().length const remaining = exchangeStore.getExchanges().filter((e: any) => e.deviceId !== deviceId) exchangeStore.setExchanges(remaining) return Promise.resolve(before - remaining.length) }), countByDevice: vi.fn().mockImplementation((deviceId: string) => { return Promise.resolve(exchangeStore.getExchanges().filter((e: any) => e.deviceId === deviceId).length) }), deleteOldestBeyondCap: vi.fn().mockResolvedValue(undefined), hasExchanges: vi.fn().mockImplementation((deviceId: string) => { return Promise.resolve(exchangeStore.getExchanges().some((e: any) => e.deviceId === deviceId)) }), getMetadata: vi.fn().mockImplementation((deviceId: string) => { const deviceExchanges = exchangeStore.getExchanges().filter((e: any) => e.deviceId === deviceId) if (deviceExchanges.length === 0) { return Promise.resolve({ recordedFrom: null, recordedUntil: null, count: 0 }) } const sorted = [...deviceExchanges].sort((a, b) => a.timestamp.localeCompare(b.timestamp)) return Promise.resolve({ recordedFrom: sorted[0].timestamp, recordedUntil: sorted[sorted.length - 1].timestamp, count: deviceExchanges.length, }) }), }, })) vi.mock('#providers/mongodb_provider', () => ({ getMongoDb: () => ({ collection: () => ({ findOneAndUpdate: vi.fn().mockImplementation(() => { return Promise.resolve({ seq: exchangeStore.incrementCounter() }) }), }), }), })) vi.mock('#config/exchange_store', () => ({ default: { maxMessagesPerDevice: 5000, ttlSeconds: 86400, retry: { maxAttempts: 3, baseDelayMs: 10 }, featureFlags: { persistence: true, sseEnabled: true, apiSource: 'mongodb' }, sse: { inactivityTimeoutMs: 300_000, absoluteTimeoutMs: 7_200_000, maxConnectionsPerDevice: 5, retryIntervalMs: 3000, }, }, })) // Mock Tr069TraceService — device is always traced const tracedDevices = vi.hoisted(() => new Set()) vi.mock('#services/tr069_trace_service', () => ({ Tr069TraceService: { getInstance: () => ({ isTraced: (deviceId: string) => tracedDevices.has(deviceId), }), }, })) // --- Imports (after mocks) --- import ExchangeController from '../../app/controllers/exchange_controller.js' import { ExchangeStoreService } from '../../app/services/exchange_store_service.js' import type { CwmpExchange } from '../../app/models/exchange.js' describe('Clear Exchanges Flow (Integration)', () => { const DEVICE_ID = 'test-device-clear-001' let controller: ExchangeController let storeService: ExchangeStoreService beforeEach(() => { vi.clearAllMocks() exchangeStore.clear() tracedDevices.clear() tracedDevices.add(DEVICE_ID) // Reset singleton to get a fresh instance ;(ExchangeStoreService as any).instance = undefined storeService = ExchangeStoreService.getInstance() controller = new ExchangeController() }) afterEach(() => { tracedDevices.clear() }) /** * Helper: create mock exchanges in the in-memory store */ function createExchanges(deviceId: string, count: number): CwmpExchange[] { const exchanges: CwmpExchange[] = [] const baseTime = new Date('2025-01-15T10:00:00.000Z') for (let i = 0; i < count; i++) { const timestamp = new Date(baseTime.getTime() + i * 1000).toISOString() const exchange: CwmpExchange = { deviceId, direction: i % 2 === 0 ? 'CPE→ACS' : 'ACS→CPE', timestamp, sessionId: `10.0.0.1:${50000 + i}->192.168.1.1:8080`, cwmpMethod: i === 0 ? 'Inform' : i % 2 === 0 ? 'GetParameterValues' : 'GetParameterValuesResponse', rawXml: ``, sequenceNumber: i + 1, createdAt: new Date(baseTime.getTime() + i * 1000), } exchanges.push(exchange) } exchangeStore.setExchanges([...exchangeStore.getExchanges(), ...exchanges]) return exchanges } /** * Helper: create a mock HttpContext for testing */ function createMockContext(overrides: { params?: Record qs?: Record headers?: Record } = {}) { const responseData: { status?: number; body?: any } = {} return { params: overrides.params ?? { deviceId: DEVICE_ID }, request: { qs: () => overrides.qs ?? {}, header: (name: string) => (overrides.headers ?? {})[name] ?? undefined, request: new EventEmitter(), id: () => 'test-correlation-id', }, response: { status: (code: number) => ({ json: (data: any) => { responseData.status = code responseData.body = data return { status: code, body: data } }, send: (data: any) => { responseData.status = code responseData.body = data return { status: code, body: data } }, }), response: null as any, // for SSE — will be set per test }, responseData, } } // --- Test: Requirement 8.1 — Clear exchanges button (verified via DELETE endpoint) --- it('should have exchanges present before clearing', async () => { const created = createExchanges(DEVICE_ID, 5) expect(created).toHaveLength(5) const ctx = createMockContext() const result = await controller.index(ctx as any) expect(result.status).toBe(200) expect(result.body.exchanges).toHaveLength(5) expect(result.body.metadata.total).toBe(5) expect(result.body.metadata.traceStatus).toBe('ACTIVE') }) // --- Test: Requirement 8.4 — DELETE returns 204, empties exchange list --- it('should return 204 on DELETE and empty the exchange list', async () => { createExchanges(DEVICE_ID, 10) // Verify exchanges exist const getCtx = createMockContext() const beforeResult = await controller.index(getCtx as any) expect(beforeResult.status).toBe(200) expect(beforeResult.body.exchanges).toHaveLength(10) // Call DELETE endpoint const deleteCtx = createMockContext() const deleteResult = await controller.destroy(deleteCtx as any) expect(deleteResult.status).toBe(204) // Verify exchange list is now empty // After deletion, device has no exchanges but is still traced → 200 with empty array const afterCtx = createMockContext() const afterResult = await controller.index(afterCtx as any) expect(afterResult.status).toBe(200) expect(afterResult.body.exchanges).toHaveLength(0) expect(afterResult.body.metadata.total).toBe(0) }) // --- Test: Requirement 8.4 — SSE stream remains active after clear --- it('should keep SSE stream active after clearing exchanges and receive new messages', async () => { createExchanges(DEVICE_ID, 3) // Set up SSE stream (mock node response) const sseWritten: string[] = [] let sseEnded = false const mockNodeRes = { writeHead: vi.fn(), write: vi.fn().mockImplementation((chunk: string) => { sseWritten.push(chunk) }), end: vi.fn().mockImplementation(() => { sseEnded = true }), } const mockNodeReq = new EventEmitter() const sseCtx = { params: { deviceId: DEVICE_ID }, request: { qs: () => ({}), header: (_name: string) => undefined, request: mockNodeReq, id: () => 'test-correlation-id', }, response: { status: (code: number) => ({ json: (data: any) => ({ status: code, body: data }), }), response: mockNodeRes, }, } // Connect SSE stream await controller.stream(sseCtx as any) // Verify SSE headers were set expect(mockNodeRes.writeHead).toHaveBeenCalledWith(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no', }) // SSE connected — verify initial retry interval was sent expect(sseWritten.some((chunk) => chunk.includes('retry:'))).toBe(true) // Clear the SSE written log for clarity sseWritten.length = 0 // Now DELETE all exchanges const deleteCtx = createMockContext() await controller.destroy(deleteCtx as any) // Verify SSE stream is still active (not ended) expect(sseEnded).toBe(false) // Emit a new exchange event via ExchangeStoreService (simulating a new CWMP message arriving) const newExchange: CwmpExchange = { deviceId: DEVICE_ID, direction: 'CPE→ACS', timestamp: new Date('2025-01-15T11:00:00.000Z').toISOString(), sessionId: '10.0.0.1:60000->192.168.1.1:8080', cwmpMethod: 'Inform', rawXml: '', sequenceNumber: 100, createdAt: new Date('2025-01-15T11:00:00.000Z'), } // Emit event through the ExchangeStoreService emitter ;(storeService as any).emitter.emit(`exchange:${DEVICE_ID}`, newExchange) // Verify the SSE stream received the new message expect(sseWritten.length).toBeGreaterThan(0) const lastEvent = sseWritten[sseWritten.length - 1] expect(lastEvent).toContain('id: 100') expect(lastEvent).toContain('"cwmpMethod":"Inform"') expect(lastEvent).toContain('"deviceId":"test-device-clear-001"') // Clean up — simulate client disconnect mockNodeReq.emit('close') }) // --- Test: Requirement 8.3 — If user cancels, exchanges remain unchanged --- it('should preserve exchange log if DELETE is not called (cancel scenario)', async () => { createExchanges(DEVICE_ID, 7) // Verify exchanges exist const ctx = createMockContext() const result = await controller.index(ctx as any) expect(result.status).toBe(200) expect(result.body.exchanges).toHaveLength(7) // User "cancels" — no DELETE call made — exchanges remain const verifyCtx = createMockContext() const verifyResult = await controller.index(verifyCtx as any) expect(verifyResult.status).toBe(200) expect(verifyResult.body.exchanges).toHaveLength(7) }) // --- Test: Requirement 8.5 — Server failure to clear should not disrupt existing data --- it('should preserve exchange data if DELETE fails on the server', async () => { createExchanges(DEVICE_ID, 5) // Temporarily make deleteByDevice fail const ExchangeModel = (await import('../../app/models/exchange.js')).default const originalDelete = ExchangeModel.deleteByDevice ;(ExchangeModel.deleteByDevice as any) = vi.fn().mockRejectedValue(new Error('MongoDB connection lost')) // Call DELETE — it should throw const deleteCtx = createMockContext() await expect(controller.destroy(deleteCtx as any)).rejects.toThrow('MongoDB connection lost') // Restore ;(ExchangeModel.deleteByDevice as any) = originalDelete // Verify exchanges are still present const getCtx = createMockContext() const result = await controller.index(getCtx as any) expect(result.status).toBe(200) expect(result.body.exchanges).toHaveLength(5) }) // --- Test: Requirement 8.4 — After clear, new SSE messages still arrive --- it('should continue receiving multiple SSE messages after clearing exchanges', async () => { createExchanges(DEVICE_ID, 2) // Set up SSE stream const sseWritten: string[] = [] const mockNodeRes = { writeHead: vi.fn(), write: vi.fn().mockImplementation((chunk: string) => { sseWritten.push(chunk) }), end: vi.fn(), } const mockNodeReq = new EventEmitter() const sseCtx = { params: { deviceId: DEVICE_ID }, request: { qs: () => ({}), header: (_name: string) => undefined, request: mockNodeReq, id: () => 'test-correlation-id', }, response: { status: (code: number) => ({ json: (data: any) => ({ status: code, body: data }), }), response: mockNodeRes, }, } await controller.stream(sseCtx as any) sseWritten.length = 0 // DELETE exchanges const deleteCtx = createMockContext() await controller.destroy(deleteCtx as any) // Emit multiple new exchanges const newExchanges: CwmpExchange[] = [ { deviceId: DEVICE_ID, direction: 'CPE→ACS', timestamp: '2025-01-15T12:00:00.000Z', sessionId: '10.0.0.1:70000->192.168.1.1:8080', cwmpMethod: 'Inform', rawXml: '', sequenceNumber: 201, createdAt: new Date('2025-01-15T12:00:00.000Z'), }, { deviceId: DEVICE_ID, direction: 'ACS→CPE', timestamp: '2025-01-15T12:00:01.000Z', sessionId: '10.0.0.1:70000->192.168.1.1:8080', cwmpMethod: 'GetParameterValues', rawXml: '', sequenceNumber: 202, createdAt: new Date('2025-01-15T12:00:01.000Z'), }, { deviceId: DEVICE_ID, direction: 'CPE→ACS', timestamp: '2025-01-15T12:00:02.000Z', sessionId: '10.0.0.1:70000->192.168.1.1:8080', cwmpMethod: 'GetParameterValuesResponse', rawXml: '', sequenceNumber: 203, createdAt: new Date('2025-01-15T12:00:02.000Z'), }, ] for (const exchange of newExchanges) { ;(storeService as any).emitter.emit(`exchange:${DEVICE_ID}`, exchange) } // Verify all 3 new messages were received via SSE expect(sseWritten).toHaveLength(3) expect(sseWritten[0]).toContain('id: 201') expect(sseWritten[0]).toContain('"cwmpMethod":"Inform"') expect(sseWritten[1]).toContain('id: 202') expect(sseWritten[1]).toContain('"cwmpMethod":"GetParameterValues"') expect(sseWritten[2]).toContain('id: 203') expect(sseWritten[2]).toContain('"cwmpMethod":"GetParameterValuesResponse"') // Clean up mockNodeReq.emit('close') }) // --- Test: Requirement 8.2 / 8.4 — Confirm and clear with metadata reset --- it('should reset metadata after clearing exchanges', async () => { createExchanges(DEVICE_ID, 20) // Verify metadata before clear const beforeCtx = createMockContext() const beforeResult = await controller.index(beforeCtx as any) expect(beforeResult.body.metadata.total).toBe(20) expect(beforeResult.body.metadata.recordedFrom).not.toBeNull() expect(beforeResult.body.metadata.recordedUntil).not.toBeNull() // DELETE const deleteCtx = createMockContext() const deleteResult = await controller.destroy(deleteCtx as any) expect(deleteResult.status).toBe(204) // Verify metadata is reset (device still traced → 200 with empty + null metadata) const afterCtx = createMockContext() const afterResult = await controller.index(afterCtx as any) expect(afterResult.status).toBe(200) expect(afterResult.body.exchanges).toHaveLength(0) expect(afterResult.body.metadata.total).toBe(0) expect(afterResult.body.metadata.recordedFrom).toBeNull() expect(afterResult.body.metadata.recordedUntil).toBeNull() expect(afterResult.body.metadata.traceStatus).toBe('ACTIVE') }) })