Codex
114 lines
3.3 KiB
JavaScript
114 lines
3.3 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import { EventEmitter } from 'node:events';
|
|
import { loadConfig } from '../dist/config.js';
|
|
import { buildApp } from '../dist/app.js';
|
|
import {
|
|
DEVICE_UPLINK_TOPIC,
|
|
DEVICE_WILL_TOPIC,
|
|
MqttService
|
|
} from '../dist/mqtt/mqtt-service.js';
|
|
|
|
class FakeClient extends EventEmitter {
|
|
connected = false;
|
|
subscriptions = [];
|
|
publications = [];
|
|
ended = false;
|
|
|
|
subscribe(topics, options, callback) {
|
|
this.subscriptions.push({ topics, options });
|
|
callback();
|
|
}
|
|
|
|
publish(topic, payload, options, callback) {
|
|
this.publications.push({ topic, payload, options });
|
|
callback();
|
|
}
|
|
|
|
end(_force, _options, callback) {
|
|
this.ended = true;
|
|
callback();
|
|
}
|
|
}
|
|
|
|
const config = loadConfig({
|
|
NODE_ENV: 'test',
|
|
QIPAI_MQTT_URL: 'mqtt://127.0.0.1:1883',
|
|
QIPAI_MQTT_CLIENT_ID: 'qipai-backend-test',
|
|
QIPAI_MQTT_USERNAME: 'backend-test',
|
|
QIPAI_MQTT_PASSWORD: 'not-a-real-secret',
|
|
QIPAI_MQTT_MAX_MESSAGE_BYTES: '1024'
|
|
});
|
|
const fake = new FakeClient();
|
|
let connectOptions;
|
|
const service = new MqttService(config.mqtt, (_url, options) => {
|
|
connectOptions = options;
|
|
return fake;
|
|
});
|
|
|
|
service.start();
|
|
assert.equal(connectOptions.protocolVersion, 3);
|
|
assert.equal(connectOptions.clean, false);
|
|
assert.equal(connectOptions.reconnectPeriod, 3000);
|
|
assert.equal(connectOptions.resubscribe, false);
|
|
assert.equal(connectOptions.queueQoSZero, false);
|
|
|
|
fake.connected = true;
|
|
fake.emit('connect');
|
|
assert.deepEqual(fake.subscriptions, [{
|
|
topics: [DEVICE_UPLINK_TOPIC, DEVICE_WILL_TOPIC],
|
|
options: { qos: 1 }
|
|
}]);
|
|
assert.equal(service.health().subscriptionsReady, true);
|
|
|
|
await service.publishDeviceCommand('BOX_001', '{"command":"GetInfo"}');
|
|
assert.equal(fake.publications[0].topic, '/deviceaccept/BOX_001');
|
|
assert.equal(fake.publications[0].options.qos, 1);
|
|
assert.equal(fake.publications[0].options.retain, false);
|
|
|
|
fake.emit('message', '/devicesend/BOX_001', Buffer.from('{"result":"ok"}'));
|
|
assert.equal(service.health().receivedMessages, 1);
|
|
fake.emit('message', '/devicesend/BOX_001', Buffer.alloc(1025));
|
|
assert.equal(service.health().rejectedOversizeMessages, 1);
|
|
assert.match(service.health().lastError, /exceeded 1024 bytes/);
|
|
|
|
fake.emit('reconnect');
|
|
assert.equal(service.health().reconnectCount, 1);
|
|
assert.equal(service.health().subscriptionsReady, false);
|
|
fake.emit('connect');
|
|
assert.equal(fake.subscriptions.length, 2);
|
|
assert.equal(service.health().subscriptionsReady, true);
|
|
|
|
const app = await buildApp({ config, mqtt: service });
|
|
const ready = await app.inject({ method: 'GET', url: '/app-api/ready' });
|
|
assert.equal(ready.statusCode, 200);
|
|
assert.deepEqual(ready.json().checks, {
|
|
mysqlConfigured: false,
|
|
mqttConfigured: true,
|
|
mqttConnected: true,
|
|
mqttSubscriptionsReady: true
|
|
});
|
|
await app.close();
|
|
|
|
await assert.rejects(
|
|
service.publishDeviceCommand('../bad', '{}'),
|
|
/Invalid MQTT DeviceID/
|
|
);
|
|
await assert.rejects(
|
|
service.publishDeviceCommand('BOX_001', Buffer.alloc(1025)),
|
|
/exceeds 1024 bytes/
|
|
);
|
|
|
|
await service.stop();
|
|
assert.equal(fake.ended, true);
|
|
assert.equal(service.health().connected, false);
|
|
|
|
assert.throws(
|
|
() => loadConfig({
|
|
NODE_ENV: 'test',
|
|
QIPAI_MQTT_USERNAME: 'only-user'
|
|
}),
|
|
/must be configured together/
|
|
);
|
|
|
|
console.log('PASS: MQTT 3.1 transport, QoS 1, resubscribe, health and size limits work.');
|