318 lines
9.1 KiB
Dart
318 lines
9.1 KiB
Dart
import 'dart:async';
|
|
import 'dart:convert';
|
|
|
|
import 'package:archive/archive.dart';
|
|
import 'package:dart_mq/dart_mq.dart';
|
|
import 'package:flutter_test/flutter_test.dart';
|
|
import 'package:mqtt5_client/mqtt5_client.dart';
|
|
import 'package:mqtt5_client/mqtt5_server_client.dart';
|
|
import 'package:shared_preferences/shared_preferences.dart';
|
|
import 'package:stadtbote/main.dart';
|
|
import 'package:stadtbote/model/enqueued_message.dart';
|
|
import 'package:stadtbote/model/job.dart';
|
|
import 'package:stadtbote/model/log_entry.dart';
|
|
import 'package:stadtbote/objectbox.g.dart';
|
|
import 'package:stadtbote/tools/constants.dart';
|
|
import 'package:stadtbote/tools/message_header.dart';
|
|
import 'package:stadtbote/tools/messages/publish_payload_message.dart';
|
|
import 'package:stadtbote/tools/mqtt_core.dart';
|
|
import 'package:stadtbote/tools/util.dart';
|
|
// ignore: depend_on_referenced_packages
|
|
import 'package:typed_data/typed_buffers.dart';
|
|
|
|
void main() {
|
|
TestWidgetsFlutterBinding.ensureInitialized();
|
|
|
|
late _MessageStore store;
|
|
late _MqttClient client;
|
|
|
|
setUp(() async {
|
|
SharedPreferences.setMockInitialValues({});
|
|
await Util.init();
|
|
MQClient.initialize();
|
|
for (final topic in [
|
|
Constants.messageTopicPublishPayload,
|
|
Constants.messageTopicDisconnectClient,
|
|
Constants.messageTopicConnectClient,
|
|
Constants.messageTopicRequestConnectionStatus,
|
|
Constants.messageTopicConnectionStatusChanged,
|
|
]) {
|
|
MQClient.instance.declareQueue(topic);
|
|
}
|
|
store = _MessageStore();
|
|
MyApp.objectBoxStore = store;
|
|
client = _MqttClient();
|
|
MqttCore.withClient(
|
|
client: client,
|
|
clientId: 'test-device',
|
|
fromServerTopic: 'server/a/b/c/d/e/',
|
|
toServerTopic: 'client/a/b/c/d/e/',
|
|
);
|
|
});
|
|
|
|
tearDown(() async {
|
|
await client.incoming.close();
|
|
MQClient.instance.close();
|
|
});
|
|
|
|
Future<void> publish(WidgetTester tester, {bool requiresAck = true}) async {
|
|
MQClient.instance.sendMessage(
|
|
routingKey: Constants.messageTopicPublishPayload,
|
|
message: Message(
|
|
payload: PublishPayloadMessage(
|
|
topic: 'mssg',
|
|
message: {'text': 'Auftrag erledigt'},
|
|
acknowledgeRequired: requiresAck,
|
|
),
|
|
),
|
|
);
|
|
await tester.pump();
|
|
}
|
|
|
|
Future<void> retry(WidgetTester tester) async {
|
|
client.receive('sync_request', {});
|
|
await tester.pump();
|
|
}
|
|
|
|
testWidgets('behält ID, Inhalt und Bestätigungspflicht bei Wiederholungen', (
|
|
tester,
|
|
) async {
|
|
// Auch beim ersten Sendeversuch muss die Nachricht bereits gespeichert sein.
|
|
List<EnqueuedMessage>? queuedAtFirstSend;
|
|
client.onPublish = () => queuedAtFirstSend = store.messages.getAll();
|
|
await publish(tester);
|
|
client.onPublish = null;
|
|
final original = client.chatMessages.single;
|
|
final queued = store.messages.rows.single;
|
|
expect(queuedAtFirstSend, hasLength(1));
|
|
expect(queuedAtFirstSend!.single.messageId, original.messageId);
|
|
expect(queued.messageId, original.messageId);
|
|
expect(jsonDecode(queued.payload), original.payload);
|
|
|
|
await retry(tester);
|
|
await retry(tester);
|
|
|
|
expect(client.chatMessages, hasLength(3));
|
|
for (final message in client.chatMessages) {
|
|
expect(message.toJson(), original.toJson());
|
|
expect(message.requiresAck, isTrue);
|
|
}
|
|
expect(store.messages.rows.single.id, queued.id);
|
|
expect(store.messages.putCount, 1);
|
|
});
|
|
|
|
testWidgets('sendet bestätigte Nachrichten nicht erneut', (tester) async {
|
|
await publish(tester);
|
|
final messageId = client.chatMessages.single.messageId;
|
|
await retry(tester);
|
|
expect(store.messages.rows, hasLength(1));
|
|
|
|
client.receive('acknowledge', {'messageId': messageId});
|
|
await tester.pump();
|
|
expect(store.messages.rows, isEmpty);
|
|
|
|
await retry(tester);
|
|
expect(client.chatMessages, hasLength(2));
|
|
});
|
|
|
|
testWidgets('behält unbestätigte Nachrichten nach fehlgeschlagenem Versand', (
|
|
tester,
|
|
) async {
|
|
client.failPublish = true;
|
|
await publish(tester);
|
|
final messageId = store.messages.rows.single.messageId;
|
|
await retry(tester);
|
|
expect(store.messages.rows.single.messageId, messageId);
|
|
expect(client.chatMessages, isEmpty);
|
|
|
|
client.failPublish = false;
|
|
await retry(tester);
|
|
expect(client.chatMessages.single.messageId, messageId);
|
|
expect(client.chatMessages.single.requiresAck, isTrue);
|
|
expect(store.messages.rows, hasLength(1));
|
|
});
|
|
|
|
testWidgets('verwendet die ID bereits gespeicherter Nachrichten', (
|
|
tester,
|
|
) async {
|
|
store.messages.put(
|
|
EnqueuedMessage()
|
|
..messageId = 'persisted-message-id'
|
|
..topic = 'mssg'
|
|
..payload = '{"text":"Vor dem Neustart gespeichert"}',
|
|
);
|
|
|
|
await retry(tester);
|
|
|
|
final message = client.chatMessages.single;
|
|
expect(message.messageId, 'persisted-message-id');
|
|
expect(message.requiresAck, isTrue);
|
|
expect(message.payload['text'], 'Vor dem Neustart gespeichert');
|
|
expect(store.messages.rows, hasLength(1));
|
|
expect(store.messages.putCount, 1);
|
|
});
|
|
|
|
testWidgets('wiederholt Nachrichten ohne Bestätigungspflicht nicht', (
|
|
tester,
|
|
) async {
|
|
await publish(tester, requiresAck: false);
|
|
await retry(tester);
|
|
|
|
expect(client.chatMessages, hasLength(1));
|
|
expect(client.chatMessages.single.requiresAck, isFalse);
|
|
expect(store.messages.rows, isEmpty);
|
|
});
|
|
|
|
testWidgets(
|
|
'vergibt für neue Nachrichten auch bei gleichem Inhalt eigene IDs',
|
|
(tester) async {
|
|
await publish(tester);
|
|
await publish(tester);
|
|
|
|
final ids = client.chatMessages
|
|
.map((message) => message.messageId)
|
|
.toList();
|
|
expect(ids.toSet(), hasLength(2));
|
|
await retry(tester);
|
|
expect(client.chatMessages.map((message) => message.messageId), [
|
|
...ids,
|
|
...ids,
|
|
]);
|
|
expect(store.messages.rows, hasLength(2));
|
|
},
|
|
);
|
|
|
|
testWidgets('sendet bei fehlgeschlagener Speicherung keine neue Nachricht', (
|
|
tester,
|
|
) async {
|
|
store.messages.failPut = true;
|
|
await publish(tester);
|
|
|
|
expect(client.chatMessages, isEmpty);
|
|
expect(store.messages.rows, isEmpty);
|
|
});
|
|
}
|
|
|
|
class _MqttClient extends Fake implements MqttServerClient {
|
|
final incoming =
|
|
StreamController<List<MqttReceivedMessage<MqttMessage>>>.broadcast();
|
|
final chatMessages = <MessageHeader>[];
|
|
bool failPublish = false;
|
|
void Function()? onPublish;
|
|
|
|
@override
|
|
Stream<List<MqttReceivedMessage<MqttMessage>>> get updates => incoming.stream;
|
|
|
|
@override
|
|
MqttSubscription? subscribe(String topic, MqttQos qosLevel) => null;
|
|
|
|
@override
|
|
int publishMessage(
|
|
String topic,
|
|
MqttQos qualityOfService,
|
|
Uint8Buffer data, {
|
|
bool retain = false,
|
|
List<MqttUserProperty>? userProperties,
|
|
}) {
|
|
onPublish?.call();
|
|
if (failPublish) throw StateError('Keine Verbindung');
|
|
if (topic.endsWith('/mssg')) {
|
|
chatMessages.add(
|
|
MessageHeader.fromJson(
|
|
jsonDecode(utf8.decode(GZipDecoder().decodeBytes(data))),
|
|
),
|
|
);
|
|
}
|
|
return 1;
|
|
}
|
|
|
|
void receive(String topic, Map<String, dynamic> payload) {
|
|
final data = Uint8Buffer()
|
|
..addAll(
|
|
GZipEncoder().encode(
|
|
utf8.encode(
|
|
jsonEncode({
|
|
'messageId': 'server-message-id',
|
|
'requiresAck': false,
|
|
'payload': payload,
|
|
}),
|
|
),
|
|
),
|
|
);
|
|
final message = MqttPublishMessage()
|
|
.toTopic('server/a/b/c/d/e/test-device/$topic')
|
|
.publishData(data);
|
|
incoming.add([MqttReceivedMessage(topic, message)]);
|
|
}
|
|
}
|
|
|
|
class _MessageStore extends Fake implements Store {
|
|
final messages = _MessageBox();
|
|
final logs = _MemoryBox<LogEntry>();
|
|
final jobs = _MemoryBox<Job>();
|
|
|
|
@override
|
|
Box<T> box<T>() {
|
|
if (T == EnqueuedMessage) return messages as Box<T>;
|
|
if (T == LogEntry) return logs as Box<T>;
|
|
if (T == Job) return jobs as Box<T>;
|
|
throw StateError('Unerwarteter Entity-Typ: $T');
|
|
}
|
|
}
|
|
|
|
class _MemoryBox<T> extends Fake implements Box<T> {
|
|
final rows = <T>[];
|
|
|
|
@override
|
|
List<T> getAll() => List.of(rows);
|
|
|
|
@override
|
|
int put(T object, {PutMode mode = PutMode.put}) => 0;
|
|
|
|
@override
|
|
QueryBuilder<T> query([Condition<T>? qc]) => _MemoryQueryBuilder(rows);
|
|
}
|
|
|
|
class _MessageBox extends _MemoryBox<EnqueuedMessage> {
|
|
int putCount = 0;
|
|
bool failPut = false;
|
|
|
|
@override
|
|
int put(EnqueuedMessage object, {PutMode mode = PutMode.put}) {
|
|
if (failPut) throw StateError('Speicherung fehlgeschlagen');
|
|
object.id = ++putCount;
|
|
rows.add(object);
|
|
return object.id;
|
|
}
|
|
|
|
@override
|
|
int removeMany(List<int> ids) {
|
|
final count = rows.length;
|
|
rows.removeWhere((row) => ids.contains(row.id));
|
|
return count - rows.length;
|
|
}
|
|
}
|
|
|
|
// Die nativen Datenbankzugriffe sind ersetzt. Im Bestätigungstest enthält
|
|
// die Warteschlange genau die eine zum ACK passende Nachricht.
|
|
class _MemoryQueryBuilder<T> extends Fake implements QueryBuilder<T> {
|
|
final List<T> rows;
|
|
|
|
_MemoryQueryBuilder(this.rows);
|
|
|
|
@override
|
|
Query<T> build() => _MemoryQuery(rows);
|
|
}
|
|
|
|
class _MemoryQuery<T> extends Fake implements Query<T> {
|
|
final List<T> rows;
|
|
|
|
_MemoryQuery(this.rows);
|
|
|
|
@override
|
|
List<T> find() => List.of(rows);
|
|
|
|
@override
|
|
void close() {}
|
|
}
|