- MongoDB storage: cwmp_exchanges collection with indexes (deviceId+timestamp, deviceId+sequenceNumber unique, TTL 24h) - ExchangeStoreService: CWMP method extraction, per-device queue for ordering, retry logic, message cap (5000/device) - Exchange API: GET/DELETE/SSE stream endpoints with Last-Event-ID reconnection - CwmpLoggerMiddleware: dual-write (MongoDB + legacy files) with fire-and-forget - Frontend: SequenceViewer (CPE↔ACS lifelines, arrows, auto-scroll, virtual scroll) + InspectorPanel (formatted XML, copy, syntax highlighting) - Tr069Trace page: integrated new viewer with col-6/col-6 layout - DeviceDetail page: collapsible exchange section when device is traced - 14 property-based tests, 40+ unit tests, 5 integration tests - Health check includes index verification - Config: feature flags, SSE timeouts, retry policy
700 lines
24 KiB
TypeScript
700 lines
24 KiB
TypeScript
/**
|
|
* Integration test: Full Exchange Lifecycle
|
|
*
|
|
* Tests the complete flow:
|
|
* SOAP capture → persist to MongoDB → SSE event emitted → data integrity verified
|
|
*
|
|
* This is a unit-level integration test using mocks (not a real MongoDB connection).
|
|
* We mock the MongoDB layer but verify the data flows through:
|
|
* ExchangeStoreService → ExchangeModel → EventEmitter correctly.
|
|
*
|
|
* Validates: Requirements 1.1, 1.7, 6.1, 6.2
|
|
*/
|
|
import { beforeEach, afterEach, describe, expect, it, vi } from 'vitest'
|
|
|
|
// ─── MongoDB Mocks ──────────────────────────────────────────────────────────────
|
|
|
|
let sequenceCounter = 0
|
|
const mockInsertOne = vi.fn().mockImplementation(async () => ({ insertedId: 'mock-id' }))
|
|
const mockCountDocuments = vi.fn().mockResolvedValue(0)
|
|
const mockDeleteMany = vi.fn().mockResolvedValue({ deletedCount: 0 })
|
|
const mockFind = vi.fn().mockReturnValue({
|
|
sort: vi.fn().mockReturnThis(),
|
|
limit: vi.fn().mockReturnThis(),
|
|
project: vi.fn().mockReturnThis(),
|
|
toArray: vi.fn().mockResolvedValue([]),
|
|
})
|
|
|
|
const mockFindOneAndUpdate = vi.fn().mockImplementation(async () => {
|
|
sequenceCounter++
|
|
return { seq: sequenceCounter }
|
|
})
|
|
|
|
vi.mock('#providers/mongodb_provider', () => ({
|
|
getMongoDb: () => ({
|
|
collection: (name: string) => {
|
|
if (name === 'cwmp_exchange_counters') {
|
|
return { findOneAndUpdate: mockFindOneAndUpdate }
|
|
}
|
|
return {
|
|
insertOne: mockInsertOne,
|
|
countDocuments: mockCountDocuments,
|
|
deleteMany: mockDeleteMany,
|
|
find: mockFind,
|
|
findOne: vi.fn().mockResolvedValue(null),
|
|
}
|
|
},
|
|
}),
|
|
}))
|
|
|
|
vi.mock('#models/exchange', () => ({
|
|
default: {
|
|
collectionName: 'cwmp_exchanges',
|
|
getCollection: () => ({
|
|
insertOne: mockInsertOne,
|
|
countDocuments: mockCountDocuments,
|
|
deleteMany: mockDeleteMany,
|
|
find: mockFind,
|
|
findOne: vi.fn().mockResolvedValue(null),
|
|
}),
|
|
countByDevice: vi.fn().mockResolvedValue(0),
|
|
deleteOldestBeyondCap: vi.fn().mockResolvedValue(undefined),
|
|
},
|
|
}))
|
|
|
|
vi.mock('#config/exchange_store', () => ({
|
|
default: {
|
|
maxMessagesPerDevice: 5000,
|
|
ttlSeconds: 86400,
|
|
sse: {
|
|
inactivityTimeoutMs: 300_000,
|
|
absoluteTimeoutMs: 7_200_000,
|
|
maxConnectionsPerDevice: 5,
|
|
retryIntervalMs: 3000,
|
|
},
|
|
retry: {
|
|
maxAttempts: 3,
|
|
baseDelayMs: 1, // 1ms for fast tests
|
|
},
|
|
featureFlags: {
|
|
persistence: true,
|
|
sseEnabled: true,
|
|
apiSource: 'mongodb',
|
|
},
|
|
},
|
|
}))
|
|
|
|
vi.mock('@adonisjs/core/services/logger', () => ({
|
|
default: {
|
|
warn: vi.fn(),
|
|
error: vi.fn(),
|
|
info: vi.fn(),
|
|
debug: vi.fn(),
|
|
},
|
|
}))
|
|
|
|
import { ExchangeStoreService } from '../../app/services/exchange_store_service.js'
|
|
import type { CwmpExchange } from '../../app/models/exchange.js'
|
|
|
|
// ─── Test Suite ─────────────────────────────────────────────────────────────────
|
|
|
|
describe('Exchange Lifecycle Integration', () => {
|
|
let service: ExchangeStoreService
|
|
|
|
beforeEach(() => {
|
|
vi.clearAllMocks()
|
|
sequenceCounter = 0
|
|
// Reset singleton for clean state
|
|
;(ExchangeStoreService as any).instance = undefined
|
|
service = ExchangeStoreService.getInstance()
|
|
})
|
|
|
|
afterEach(() => {
|
|
;(ExchangeStoreService as any).instance = undefined
|
|
})
|
|
|
|
describe('Full lifecycle: SOAP capture → persist → SSE event', () => {
|
|
it('persists a CPE→ACS Inform and emits SSE event with correct data', async () => {
|
|
const soapXml = `<?xml version="1.0" encoding="UTF-8"?>
|
|
<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
|
|
<soap-env:Body>
|
|
<cwmp:Inform>
|
|
<DeviceId>
|
|
<Manufacturer>Sagemcom</Manufacturer>
|
|
<OUI>F8B7FF</OUI>
|
|
<ProductClass>Livebox 4</ProductClass>
|
|
<SerialNumber>LB4-TEST-001</SerialNumber>
|
|
</DeviceId>
|
|
</cwmp:Inform>
|
|
</soap-env:Body>
|
|
</soap-env:Envelope>`
|
|
|
|
const deviceId = 'test-device-001'
|
|
const receivedEvents: CwmpExchange[] = []
|
|
|
|
// Subscribe to SSE events before persisting
|
|
const listener = (exchange: CwmpExchange) => {
|
|
receivedEvents.push(exchange)
|
|
}
|
|
service.subscribe(deviceId, listener)
|
|
|
|
// Simulate SOAP capture → persist (what CwmpLoggerMiddleware does)
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: '10.0.0.1:54321->192.168.1.1:8080',
|
|
rawXml: soapXml,
|
|
})
|
|
|
|
// Verify persistence result
|
|
expect(result).not.toBeNull()
|
|
expect(result!.deviceId).toBe(deviceId)
|
|
expect(result!.direction).toBe('CPE→ACS')
|
|
expect(result!.cwmpMethod).toBe('Inform')
|
|
expect(result!.rawXml).toBe(soapXml)
|
|
expect(result!.sequenceNumber).toBe(1)
|
|
expect(result!.sessionId).toBe('10.0.0.1:54321->192.168.1.1:8080')
|
|
expect(result!.timestamp).toBeDefined()
|
|
|
|
// Verify MongoDB insertOne was called with correct document
|
|
expect(mockInsertOne).toHaveBeenCalledTimes(1)
|
|
const insertedDoc = mockInsertOne.mock.calls[0][0]
|
|
expect(insertedDoc.deviceId).toBe(deviceId)
|
|
expect(insertedDoc.direction).toBe('CPE→ACS')
|
|
expect(insertedDoc.cwmpMethod).toBe('Inform')
|
|
expect(insertedDoc.rawXml).toBe(soapXml)
|
|
expect(insertedDoc.sequenceNumber).toBe(1)
|
|
|
|
// Verify SSE event was emitted with the same data
|
|
expect(receivedEvents).toHaveLength(1)
|
|
expect(receivedEvents[0].deviceId).toBe(deviceId)
|
|
expect(receivedEvents[0].direction).toBe('CPE→ACS')
|
|
expect(receivedEvents[0].cwmpMethod).toBe('Inform')
|
|
expect(receivedEvents[0].rawXml).toBe(soapXml)
|
|
expect(receivedEvents[0].sequenceNumber).toBe(1)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('persists an ACS→CPE GetParameterValuesResponse and emits SSE event', async () => {
|
|
const responseXml = `<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
|
|
<soap-env:Body>
|
|
<cwmp:GetParameterValuesResponse>
|
|
<ParameterList>
|
|
<ParameterValueStruct>
|
|
<Name>Device.DeviceInfo.SoftwareVersion</Name>
|
|
<Value>SR40_sip-fr-4.62.0</Value>
|
|
</ParameterValueStruct>
|
|
</ParameterList>
|
|
</cwmp:GetParameterValuesResponse>
|
|
</soap-env:Body>
|
|
</soap-env:Envelope>`
|
|
|
|
const deviceId = 'test-device-002'
|
|
const receivedEvents: CwmpExchange[] = []
|
|
const listener = (exchange: CwmpExchange) => receivedEvents.push(exchange)
|
|
|
|
service.subscribe(deviceId, listener)
|
|
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'ACS→CPE',
|
|
sessionId: '192.168.1.1:8080->10.0.0.1:54321',
|
|
rawXml: responseXml,
|
|
})
|
|
|
|
// Verify data integrity across all steps
|
|
expect(result).not.toBeNull()
|
|
expect(result!.deviceId).toBe(deviceId)
|
|
expect(result!.direction).toBe('ACS→CPE')
|
|
expect(result!.cwmpMethod).toBe('GetParameterValuesResponse')
|
|
expect(result!.rawXml).toBe(responseXml)
|
|
|
|
// Verify SSE event matches persisted document
|
|
expect(receivedEvents).toHaveLength(1)
|
|
expect(receivedEvents[0]).toEqual(result)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('handles a complete session: CPE→ACS Inform + ACS→CPE InformResponse', async () => {
|
|
const deviceId = 'test-device-session'
|
|
const receivedEvents: CwmpExchange[] = []
|
|
const listener = (exchange: CwmpExchange) => receivedEvents.push(exchange)
|
|
|
|
service.subscribe(deviceId, listener)
|
|
|
|
// Step 1: CPE sends Inform
|
|
const informXml = `<soap-env:Envelope xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
|
|
<soap-env:Body><cwmp:Inform><DeviceId/></cwmp:Inform></soap-env:Body>
|
|
</soap-env:Envelope>`
|
|
|
|
const inform = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: '10.0.0.1:5000->192.168.1.1:8080',
|
|
rawXml: informXml,
|
|
})
|
|
|
|
// Step 2: ACS responds with InformResponse
|
|
const informResponseXml = `<soap-env:Envelope xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
|
|
<soap-env:Body><cwmp:InformResponse><MaxEnvelopes>1</MaxEnvelopes></cwmp:InformResponse></soap-env:Body>
|
|
</soap-env:Envelope>`
|
|
|
|
const informResponse = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'ACS→CPE',
|
|
sessionId: '10.0.0.1:5000->192.168.1.1:8080',
|
|
rawXml: informResponseXml,
|
|
})
|
|
|
|
// Verify sequence numbers are monotonically increasing
|
|
expect(inform!.sequenceNumber).toBe(1)
|
|
expect(informResponse!.sequenceNumber).toBe(2)
|
|
|
|
// Verify all events were emitted in order
|
|
expect(receivedEvents).toHaveLength(2)
|
|
expect(receivedEvents[0].cwmpMethod).toBe('Inform')
|
|
expect(receivedEvents[0].direction).toBe('CPE→ACS')
|
|
expect(receivedEvents[0].sequenceNumber).toBe(1)
|
|
expect(receivedEvents[1].cwmpMethod).toBe('InformResponse')
|
|
expect(receivedEvents[1].direction).toBe('ACS→CPE')
|
|
expect(receivedEvents[1].sequenceNumber).toBe(2)
|
|
|
|
// Verify data integrity: rawXml preserved at each step
|
|
expect(receivedEvents[0].rawXml).toBe(informXml)
|
|
expect(receivedEvents[1].rawXml).toBe(informResponseXml)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
})
|
|
|
|
describe('Data integrity verification', () => {
|
|
it('preserves deviceId through the entire lifecycle', async () => {
|
|
const deviceId = 'unique-device-id-αβγ'
|
|
const receivedEvents: CwmpExchange[] = []
|
|
const listener = (exchange: CwmpExchange) => receivedEvents.push(exchange)
|
|
service.subscribe(deviceId, listener)
|
|
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 'sess-1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
// Verify deviceId preserved: service return
|
|
expect(result!.deviceId).toBe(deviceId)
|
|
// Verify deviceId preserved: MongoDB insert
|
|
expect(mockInsertOne.mock.calls[0][0].deviceId).toBe(deviceId)
|
|
// Verify deviceId preserved: SSE event
|
|
expect(receivedEvents[0].deviceId).toBe(deviceId)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('preserves direction through the entire lifecycle', async () => {
|
|
const deviceId = 'direction-test'
|
|
const receivedEvents: CwmpExchange[] = []
|
|
const listener = (exchange: CwmpExchange) => receivedEvents.push(exchange)
|
|
service.subscribe(deviceId, listener)
|
|
|
|
// Test CPE→ACS direction
|
|
await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
// Test ACS→CPE direction
|
|
await service.persistExchange({
|
|
deviceId,
|
|
direction: 'ACS→CPE',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:GetParameterValues/>',
|
|
})
|
|
|
|
expect(receivedEvents[0].direction).toBe('CPE→ACS')
|
|
expect(receivedEvents[1].direction).toBe('ACS→CPE')
|
|
|
|
// Verify in MongoDB writes
|
|
expect(mockInsertOne.mock.calls[0][0].direction).toBe('CPE→ACS')
|
|
expect(mockInsertOne.mock.calls[1][0].direction).toBe('ACS→CPE')
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('preserves cwmpMethod extraction through the entire lifecycle', async () => {
|
|
const deviceId = 'method-test'
|
|
const receivedEvents: CwmpExchange[] = []
|
|
const listener = (exchange: CwmpExchange) => receivedEvents.push(exchange)
|
|
service.subscribe(deviceId, listener)
|
|
|
|
const methods = [
|
|
{ xml: '<cwmp:Inform/>', expected: 'Inform' },
|
|
{ xml: '<cwmp:GetParameterValues/>', expected: 'GetParameterValues' },
|
|
{ xml: '<cwmp:SetParameterValuesResponse/>', expected: 'SetParameterValuesResponse' },
|
|
{ xml: '', expected: 'Empty' },
|
|
{ xml: '<no-cwmp-here/>', expected: 'Unknown' },
|
|
]
|
|
|
|
for (const { xml } of methods) {
|
|
await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 'sess',
|
|
rawXml: xml,
|
|
})
|
|
}
|
|
|
|
// Verify each extracted method in the SSE events
|
|
expect(receivedEvents[0].cwmpMethod).toBe('Inform')
|
|
expect(receivedEvents[1].cwmpMethod).toBe('GetParameterValues')
|
|
expect(receivedEvents[2].cwmpMethod).toBe('SetParameterValuesResponse')
|
|
expect(receivedEvents[3].cwmpMethod).toBe('Empty')
|
|
expect(receivedEvents[4].cwmpMethod).toBe('Unknown')
|
|
|
|
// Verify same values in MongoDB inserts
|
|
for (let i = 0; i < methods.length; i++) {
|
|
expect(mockInsertOne.mock.calls[i][0].cwmpMethod).toBe(methods[i].expected)
|
|
}
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('preserves rawXml without modification through the entire lifecycle', async () => {
|
|
const deviceId = 'rawxml-test'
|
|
const receivedEvents: CwmpExchange[] = []
|
|
const listener = (exchange: CwmpExchange) => receivedEvents.push(exchange)
|
|
service.subscribe(deviceId, listener)
|
|
|
|
const complexXml = `<?xml version="1.0" encoding="UTF-8"?>
|
|
<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap-enc="http://schemas.xmlsoap.org/soap/encoding/" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
|
|
<soap-env:Header>
|
|
<cwmp:ID soap-env:mustUnderstand="1">123456</cwmp:ID>
|
|
</soap-env:Header>
|
|
<soap-env:Body>
|
|
<cwmp:Inform>
|
|
<DeviceId>
|
|
<Manufacturer>Sagemcom</Manufacturer>
|
|
<OUI>F8B7FF</OUI>
|
|
<ProductClass>Livebox 4</ProductClass>
|
|
<SerialNumber>LB4-TEST-001</SerialNumber>
|
|
</DeviceId>
|
|
<Event>
|
|
<EventStruct>
|
|
<EventCode>0 BOOTSTRAP</EventCode>
|
|
<CommandKey></CommandKey>
|
|
</EventStruct>
|
|
</Event>
|
|
<ParameterList>
|
|
<ParameterValueStruct>
|
|
<Name>Device.DeviceInfo.SoftwareVersion</Name>
|
|
<Value xsi:type="xsd:string">4.62.0</Value>
|
|
</ParameterValueStruct>
|
|
</ParameterList>
|
|
</cwmp:Inform>
|
|
</soap-env:Body>
|
|
</soap-env:Envelope>`
|
|
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: '10.0.0.1:1234->192.168.1.1:8080',
|
|
rawXml: complexXml,
|
|
})
|
|
|
|
// Verify rawXml preserved exactly in service result
|
|
expect(result!.rawXml).toBe(complexXml)
|
|
// Verify rawXml preserved exactly in MongoDB insert
|
|
expect(mockInsertOne.mock.calls[0][0].rawXml).toBe(complexXml)
|
|
// Verify rawXml preserved exactly in SSE event
|
|
expect(receivedEvents[0].rawXml).toBe(complexXml)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
})
|
|
|
|
describe('SSE event emission and subscription', () => {
|
|
it('emits events only to subscribers of the correct device', async () => {
|
|
const eventsA: CwmpExchange[] = []
|
|
const eventsB: CwmpExchange[] = []
|
|
const listenerA = (e: CwmpExchange) => eventsA.push(e)
|
|
const listenerB = (e: CwmpExchange) => eventsB.push(e)
|
|
|
|
service.subscribe('device-A', listenerA)
|
|
service.subscribe('device-B', listenerB)
|
|
|
|
await service.persistExchange({
|
|
deviceId: 'device-A',
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
expect(eventsA).toHaveLength(1)
|
|
expect(eventsB).toHaveLength(0)
|
|
|
|
await service.persistExchange({
|
|
deviceId: 'device-B',
|
|
direction: 'ACS→CPE',
|
|
sessionId: 's2',
|
|
rawXml: '<cwmp:GetParameterValuesResponse/>',
|
|
})
|
|
|
|
expect(eventsA).toHaveLength(1)
|
|
expect(eventsB).toHaveLength(1)
|
|
|
|
service.unsubscribe('device-A', listenerA)
|
|
service.unsubscribe('device-B', listenerB)
|
|
})
|
|
|
|
it('stops emitting after unsubscribe', async () => {
|
|
const deviceId = 'unsub-test'
|
|
const events: CwmpExchange[] = []
|
|
const listener = (e: CwmpExchange) => events.push(e)
|
|
|
|
service.subscribe(deviceId, listener)
|
|
|
|
await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
expect(events).toHaveLength(1)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
|
|
await service.persistExchange({
|
|
deviceId,
|
|
direction: 'ACS→CPE',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:InformResponse/>',
|
|
})
|
|
expect(events).toHaveLength(1) // No new event after unsubscribe
|
|
})
|
|
|
|
it('supports multiple subscribers for the same device', async () => {
|
|
const deviceId = 'multi-sub'
|
|
const events1: CwmpExchange[] = []
|
|
const events2: CwmpExchange[] = []
|
|
|
|
const listener1 = (e: CwmpExchange) => events1.push(e)
|
|
const listener2 = (e: CwmpExchange) => events2.push(e)
|
|
|
|
service.subscribe(deviceId, listener1)
|
|
service.subscribe(deviceId, listener2)
|
|
|
|
await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
expect(events1).toHaveLength(1)
|
|
expect(events2).toHaveLength(1)
|
|
expect(events1[0]).toEqual(events2[0])
|
|
|
|
service.unsubscribe(deviceId, listener1)
|
|
service.unsubscribe(deviceId, listener2)
|
|
})
|
|
})
|
|
|
|
describe('Retry behavior for critical vs non-critical messages (Req 1.7)', () => {
|
|
it('retries for critical Inform message on MongoDB failure — never interrupts CWMP', async () => {
|
|
// Reset mock entirely and make all inserts fail
|
|
mockInsertOne.mockReset()
|
|
mockInsertOne.mockRejectedValue(new Error('MongoDB write failed'))
|
|
|
|
const deviceId = 'retry-critical-test'
|
|
const events: CwmpExchange[] = []
|
|
const listener = (e: CwmpExchange) => events.push(e)
|
|
service.subscribe(deviceId, listener)
|
|
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
// Service never throws and still returns the exchange (fire-and-forget persistence)
|
|
// The CWMP session must NOT be interrupted regardless of persistence outcome
|
|
expect(result).not.toBeNull()
|
|
expect(result!.cwmpMethod).toBe('Inform')
|
|
expect(result!.deviceId).toBe(deviceId)
|
|
|
|
// Verify retries happened: 1 initial attempt + 3 retries = 4 total calls
|
|
expect(mockInsertOne).toHaveBeenCalledTimes(4)
|
|
|
|
// SSE event is still emitted (subscribers get notified even on write failure)
|
|
expect(events).toHaveLength(1)
|
|
expect(events[0].cwmpMethod).toBe('Inform')
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('does NOT retry for non-critical Empty message on MongoDB failure', async () => {
|
|
// Reset mock entirely and make insert fail
|
|
mockInsertOne.mockReset()
|
|
mockInsertOne.mockRejectedValue(new Error('MongoDB write failed'))
|
|
|
|
const deviceId = 'retry-noncritical-test'
|
|
const events: CwmpExchange[] = []
|
|
const listener = (e: CwmpExchange) => events.push(e)
|
|
service.subscribe(deviceId, listener)
|
|
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '', // Empty → non-critical
|
|
})
|
|
|
|
// Service still returns exchange (fire-and-forget), CWMP never interrupted
|
|
expect(result).not.toBeNull()
|
|
expect(result!.cwmpMethod).toBe('Empty')
|
|
|
|
// Only 1 attempt — non-critical messages are discarded immediately on failure
|
|
expect(mockInsertOne).toHaveBeenCalledTimes(1)
|
|
|
|
// SSE event still emitted
|
|
expect(events).toHaveLength(1)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('critical messages get 3 retries while non-critical get 0 retries', async () => {
|
|
// This verifies the retry classification difference:
|
|
// Critical: 1 initial + 3 retries = 4 attempts
|
|
// Non-critical: 1 initial + 0 retries = 1 attempt
|
|
mockInsertOne.mockReset()
|
|
mockInsertOne.mockRejectedValue(new Error('MongoDB down'))
|
|
|
|
// Test critical method (Inform)
|
|
await service.persistExchange({
|
|
deviceId: 'retry-class-test',
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
const criticalAttempts = mockInsertOne.mock.calls.length
|
|
expect(criticalAttempts).toBe(4) // 1 initial + 3 retries
|
|
|
|
// Reset and test non-critical method (Unknown)
|
|
mockInsertOne.mockClear()
|
|
await service.persistExchange({
|
|
deviceId: 'retry-class-test',
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<no-cwmp-element/>',
|
|
})
|
|
const nonCriticalAttempts = mockInsertOne.mock.calls.length
|
|
expect(nonCriticalAttempts).toBe(1) // Only 1 attempt, no retries
|
|
})
|
|
})
|
|
|
|
describe('Sequence number assignment (Req 1.1)', () => {
|
|
it('assigns monotonically increasing sequence numbers across multiple messages', async () => {
|
|
const deviceId = 'seq-test'
|
|
const events: CwmpExchange[] = []
|
|
const listener = (e: CwmpExchange) => events.push(e)
|
|
service.subscribe(deviceId, listener)
|
|
|
|
for (let i = 0; i < 5; i++) {
|
|
await service.persistExchange({
|
|
deviceId,
|
|
direction: i % 2 === 0 ? 'CPE→ACS' : 'ACS→CPE',
|
|
sessionId: 'sess',
|
|
rawXml: `<cwmp:Inform/>`,
|
|
})
|
|
}
|
|
|
|
expect(events).toHaveLength(5)
|
|
for (let i = 0; i < 5; i++) {
|
|
expect(events[i].sequenceNumber).toBe(i + 1)
|
|
}
|
|
|
|
// Verify atomic increment was called for each message
|
|
expect(mockFindOneAndUpdate).toHaveBeenCalledTimes(5)
|
|
|
|
service.unsubscribe(deviceId, listener)
|
|
})
|
|
|
|
it('uses device-scoped sequence numbers (separate counters per device)', async () => {
|
|
const eventsA: CwmpExchange[] = []
|
|
const eventsB: CwmpExchange[] = []
|
|
const listenerA = (e: CwmpExchange) => eventsA.push(e)
|
|
const listenerB = (e: CwmpExchange) => eventsB.push(e)
|
|
service.subscribe('dev-X', listenerA)
|
|
service.subscribe('dev-Y', listenerB)
|
|
|
|
await service.persistExchange({
|
|
deviceId: 'dev-X',
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
await service.persistExchange({
|
|
deviceId: 'dev-Y',
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's2',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
// Verify the counter collection is called with device-scoped _id
|
|
expect(mockFindOneAndUpdate).toHaveBeenCalledWith(
|
|
{ _id: 'dev-X' },
|
|
{ $inc: { seq: 1 } },
|
|
{ upsert: true, returnDocument: 'after' }
|
|
)
|
|
expect(mockFindOneAndUpdate).toHaveBeenCalledWith(
|
|
{ _id: 'dev-Y' },
|
|
{ $inc: { seq: 1 } },
|
|
{ upsert: true, returnDocument: 'after' }
|
|
)
|
|
|
|
service.unsubscribe('dev-X', listenerA)
|
|
service.unsubscribe('dev-Y', listenerB)
|
|
})
|
|
})
|
|
|
|
describe('CWMP session non-blocking guarantee', () => {
|
|
it('never throws even if MongoDB sequence counter is unavailable', async () => {
|
|
mockFindOneAndUpdate.mockRejectedValue(new Error('MongoDB unavailable'))
|
|
|
|
const deviceId = 'error-test'
|
|
|
|
// Should not throw — CWMP session must never be interrupted
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<cwmp:Inform/>',
|
|
})
|
|
|
|
expect(result).toBeNull()
|
|
})
|
|
|
|
it('never throws even if MongoDB insertOne is unavailable', async () => {
|
|
mockInsertOne.mockRejectedValue(new Error('MongoDB insertOne failed'))
|
|
|
|
const deviceId = 'insert-error-test'
|
|
|
|
// Should not throw — returns null
|
|
const result = await service.persistExchange({
|
|
deviceId,
|
|
direction: 'CPE→ACS',
|
|
sessionId: 's1',
|
|
rawXml: '<no-cwmp/>', // Unknown → non-critical, no retries
|
|
})
|
|
|
|
expect(result).toBeNull()
|
|
})
|
|
})
|
|
})
|