mirror of
https://gitlab.com/veilid/veilid.git
synced 2024-12-27 08:19:27 -05:00
446 lines
14 KiB
Dart
446 lines
14 KiB
Dart
import 'dart:async';
|
|
import 'dart:convert';
|
|
|
|
import 'package:flutter_test/flutter_test.dart';
|
|
import 'package:veilid/veilid.dart';
|
|
|
|
final bogusKey =
|
|
TypedKey.fromString('VLD0:qD10lHHPD1_Qr23_Qy-1JnxTht12eaWwENVG_m2v7II');
|
|
|
|
Future<void> testGetDHTValueUnopened() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
await expectLater(() async => rc.getDHTValue(bogusKey, 0),
|
|
throwsA(isA<VeilidAPIException>()));
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testOpenDHTRecordNonexistentNoWriter() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
await expectLater(() async => rc.openDHTRecord(bogusKey),
|
|
throwsA(isA<VeilidAPIException>()));
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testCloseDHTRecordNonexistent() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
await expectLater(() async => rc.closeDHTRecord(bogusKey),
|
|
throwsA(isA<VeilidAPIException>()));
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testDeleteDHTRecordNonexistent() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
await expectLater(() async => rc.deleteDHTRecord(bogusKey),
|
|
throwsA(isA<VeilidAPIException>()));
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testCreateDeleteDHTRecordSimple() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
final rec = await rc.createDHTRecord(const DHTSchema.dflt(oCnt: 1));
|
|
await rc.closeDHTRecord(rec.key);
|
|
await rc.deleteDHTRecord(rec.key);
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testCreateDeleteDHTRecordNoClose() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
final rec = await rc.createDHTRecord(const DHTSchema.dflt(oCnt: 1));
|
|
await rc.deleteDHTRecord(rec.key);
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testGetDHTValueNonexistent() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
final rec = await rc.createDHTRecord(const DHTSchema.dflt(oCnt: 1));
|
|
expect(await rc.getDHTValue(rec.key, 0), isNull);
|
|
await rc.deleteDHTRecord(rec.key);
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testSetGetDHTValue() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
final rec = await rc.createDHTRecord(const DHTSchema.dflt(oCnt: 2));
|
|
expect(await rc.setDHTValue(rec.key, 0, utf8.encode('BLAH BLAH BLAH')),
|
|
isNull);
|
|
final vd2 = await rc.getDHTValue(rec.key, 0);
|
|
expect(vd2, isNotNull);
|
|
|
|
final vd3 = await rc.getDHTValue(rec.key, 0, forceRefresh: true);
|
|
expect(vd3, isNotNull);
|
|
|
|
final vd4 = await rc.getDHTValue(rec.key, 1);
|
|
expect(vd4, isNull);
|
|
|
|
expect(vd2, equals(vd3));
|
|
|
|
await rc.deleteDHTRecord(rec.key);
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testOpenWriterDHTValue() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
var rec = await rc.createDHTRecord(const DHTSchema.dflt(oCnt: 2));
|
|
final key = rec.key;
|
|
final owner = rec.owner;
|
|
final secret = rec.ownerSecret!;
|
|
|
|
final cs = await Veilid.instance.getCryptoSystem(rec.key.kind);
|
|
expect(await cs.validateKeyPair(owner, secret), isTrue);
|
|
final otherKeyPair = await cs.generateKeyPair();
|
|
|
|
final va = utf8.encode('Qwertyuiop Asdfghjkl Zxcvbnm');
|
|
final vb = utf8.encode('1234567890');
|
|
final vc = utf8.encode(r'!@#$%^&*()');
|
|
|
|
// Test subkey writes
|
|
expect(await rc.setDHTValue(key, 1, va), isNull);
|
|
|
|
var vdtemp = await rc.getDHTValue(key, 1);
|
|
expect(vdtemp, isNotNull);
|
|
expect(vdtemp!.data, equals(va));
|
|
expect(vdtemp.seq, equals(0));
|
|
expect(vdtemp.writer, equals(owner));
|
|
|
|
expect(await rc.getDHTValue(key, 0), isNull);
|
|
|
|
expect(await rc.setDHTValue(key, 0, vb), isNull);
|
|
|
|
expect(
|
|
await rc.getDHTValue(key, 0, forceRefresh: true),
|
|
equals(ValueData(
|
|
data: vb,
|
|
seq: 0,
|
|
writer: owner,
|
|
)));
|
|
|
|
expect(
|
|
await rc.getDHTValue(key, 1, forceRefresh: true),
|
|
equals(ValueData(
|
|
data: va,
|
|
seq: 0,
|
|
writer: owner,
|
|
)));
|
|
|
|
// Equal value should not trigger sequence number update
|
|
expect(await rc.setDHTValue(key, 1, va), isNull);
|
|
|
|
// Different value should trigger sequence number update
|
|
expect(await rc.setDHTValue(key, 1, vb), isNull);
|
|
|
|
await settle(rc, key, 0);
|
|
await settle(rc, key, 1);
|
|
|
|
// Now that we initialized some subkeys
|
|
// and verified they stored correctly
|
|
// Delete things locally and reopen and see if we can write
|
|
// with the same writer key
|
|
//
|
|
|
|
await rc.closeDHTRecord(key);
|
|
await rc.deleteDHTRecord(key);
|
|
|
|
rec = await rc.openDHTRecord(key,
|
|
writer: KeyPair(key: owner, secret: secret));
|
|
expect(rec, isNotNull);
|
|
expect(rec.key, equals(key));
|
|
expect(rec.owner, equals(owner));
|
|
expect(rec.ownerSecret, equals(secret));
|
|
expect(rec.schema, isA<DHTSchemaDFLT>());
|
|
expect(rec.schema.oCnt, equals(2));
|
|
|
|
// Verify subkey 1 can be set before it is get but newer is available online
|
|
vdtemp = await rc.setDHTValue(key, 1, vc);
|
|
expect(vdtemp, isNotNull);
|
|
expect(vdtemp!.data, equals(vb));
|
|
expect(vdtemp.seq, equals(1));
|
|
expect(vdtemp.writer, equals(owner));
|
|
|
|
// Verify subkey 1 can be set a second time
|
|
// and it updates because seq is newer
|
|
expect(await rc.setDHTValue(key, 1, vc), isNull);
|
|
|
|
// Verify the network got the subkey update with a refresh check
|
|
vdtemp = await rc.getDHTValue(key, 1, forceRefresh: true);
|
|
expect(vdtemp, isNotNull);
|
|
expect(vdtemp!.data, equals(vc));
|
|
expect(vdtemp.seq, equals(2));
|
|
expect(vdtemp.writer, equals(owner));
|
|
|
|
// Delete things locally and reopen and see if we can write
|
|
// with a different writer key (should fail)
|
|
await rc.closeDHTRecord(key);
|
|
await rc.deleteDHTRecord(key);
|
|
|
|
rec = await rc.openDHTRecord(key, writer: otherKeyPair);
|
|
expect(rec, isNotNull);
|
|
expect(rec.key, equals(key));
|
|
expect(rec.owner, equals(owner));
|
|
expect(rec.ownerSecret, isNull);
|
|
expect(rec.schema, isA<DHTSchemaDFLT>());
|
|
expect(rec.schema.oCnt, equals(2));
|
|
|
|
// Verify subkey 1 can NOT be set because we have the wrong writer
|
|
await expectLater(() async => rc.setDHTValue(key, 1, va),
|
|
throwsA(isA<VeilidAPIException>()));
|
|
|
|
// Verify subkey 0 can NOT be set because we have the wrong writer
|
|
await expectLater(() async => rc.setDHTValue(key, 0, va),
|
|
throwsA(isA<VeilidAPIException>()));
|
|
|
|
// Verify subkey 0 can be set because override with the right writer
|
|
expect(
|
|
await rc.setDHTValue(key, 0, va,
|
|
writer: KeyPair(key: owner, secret: secret)),
|
|
isNull);
|
|
|
|
// Clean up
|
|
await rc.closeDHTRecord(key);
|
|
await rc.deleteDHTRecord(key);
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|
|
|
|
Future<void> settle(VeilidRoutingContext rc, TypedKey key, int subkey) async {
|
|
// Wait for set to settle
|
|
do {
|
|
await Future<void>.delayed(const Duration(milliseconds: 100));
|
|
} while (
|
|
(await rc.inspectDHTRecord(key)).offlineSubkeys.containsSubkey(subkey));
|
|
}
|
|
|
|
Future<VeilidUpdateValueChange?> waitForValueChange(
|
|
Stream<VeilidUpdateValueChange> stream,
|
|
Duration duration,
|
|
Future<void> Function() closure) async {
|
|
final valueChangeQueueIterator = StreamIterator(stream);
|
|
|
|
try {
|
|
// Subscribe before call
|
|
final iterfut = valueChangeQueueIterator.moveNext();
|
|
|
|
// Call thing that might generate a value change
|
|
await closure();
|
|
|
|
// Wait for the first change
|
|
final hasChange =
|
|
await iterfut.timeout(duration, onTimeout: () async => false);
|
|
|
|
if (!hasChange) {
|
|
return null;
|
|
}
|
|
return valueChangeQueueIterator.current;
|
|
} finally {
|
|
// Stop waiting for changes
|
|
await valueChangeQueueIterator.cancel();
|
|
}
|
|
}
|
|
|
|
// XXX: Currently does not work because ValueChanged updates are suppressed
|
|
// for records that are the same sequence number locally as they are in the
|
|
// update. To properly test, you need two servers. Revisit this when we can
|
|
// make multiple veilid-core instantiations in a single process.
|
|
Future<void> testWatchDHTValues(Stream<VeilidUpdate> updateStream) async {
|
|
final valueChangeQueue =
|
|
StreamController<VeilidUpdateValueChange>.broadcast();
|
|
final valueChangeSubscription = updateStream.listen((update) {
|
|
if (update is VeilidUpdateValueChange) {
|
|
// print("valuechange: " + update.toString());
|
|
valueChangeQueue.sink.add(update);
|
|
}
|
|
});
|
|
|
|
try {
|
|
// Make two routing contexts, one with and one without safety
|
|
// So we can pretend to be a different node and get the watch updates
|
|
// Normally they would not get sent if the set comes from the same target
|
|
// as the watch's target
|
|
|
|
final rcSet = await Veilid.instance.routingContext();
|
|
final rcWatch = await Veilid.instance
|
|
.unsafeRoutingContext(sequencing: Sequencing.ensureOrdered);
|
|
try {
|
|
// Make a DHT record
|
|
var rec = await rcWatch.createDHTRecord(const DHTSchema.dflt(oCnt: 10));
|
|
|
|
// Set some subkey we care about
|
|
expect(
|
|
await rcWatch.setDHTValue(rec.key, 3, utf8.encode('BLAH BLAH BLAH')),
|
|
isNull);
|
|
|
|
// Wait for set to settle
|
|
await settle(rcWatch, rec.key, 3);
|
|
|
|
// Make a watch on that subkey
|
|
expect(await rcWatch.watchDHTValues(rec.key),
|
|
isNot(equals(Timestamp.zero())));
|
|
|
|
// Reopen without closing to change routing context and not lose watch
|
|
rec = await rcSet.openDHTRecord(rec.key, writer: rec.ownerKeyPair());
|
|
|
|
// Now we should NOT get an update because the update
|
|
// is the same as our local copy
|
|
final update1 = await waitForValueChange(
|
|
valueChangeQueue.stream, const Duration(seconds: 10), () async {
|
|
// Now set the subkey and trigger an update
|
|
expect(await rcSet.setDHTValue(rec.key, 3, utf8.encode('BLAH BLAH')),
|
|
isNull);
|
|
|
|
// Wait for set to settle
|
|
await settle(rcSet, rec.key, 3);
|
|
});
|
|
if (update1 != null) {
|
|
fail('should not have a change');
|
|
}
|
|
|
|
// Wait for the update
|
|
final update2 = await waitForValueChange(
|
|
valueChangeQueue.stream, const Duration(seconds: 10), () async {
|
|
// Now set a subkey and trigger an update
|
|
expect(
|
|
await rcSet.setDHTValue(rec.key, 3, utf8.encode('BLAH')), isNull);
|
|
|
|
await settle(rcSet, rec.key, 3);
|
|
});
|
|
if (update2 == null) {
|
|
fail('should have a change');
|
|
}
|
|
|
|
// Verify the update
|
|
expect(update2.key, equals(rec.key));
|
|
expect(update2.count, equals(0xFFFFFFFD));
|
|
expect(update2.subkeys, equals([ValueSubkeyRange.single(3)]));
|
|
expect(update2.value, isNull);
|
|
|
|
// Reopen without closing to change routing context and not lose watch
|
|
rec = await rcWatch.openDHTRecord(rec.key, writer: rec.ownerKeyPair());
|
|
|
|
// Cancel some subkeys we don't care about
|
|
expect(
|
|
await rcWatch
|
|
.cancelDHTWatch(rec.key, subkeys: [ValueSubkeyRange.make(0, 2)]),
|
|
isTrue);
|
|
|
|
// Reopen without closing to change routing context and not lose watch
|
|
rec = await rcSet.openDHTRecord(rec.key, writer: rec.ownerKeyPair());
|
|
|
|
// Wait for the update
|
|
final update3 = await waitForValueChange(
|
|
valueChangeQueue.stream, const Duration(seconds: 10), () async {
|
|
// Now set multiple subkeys and trigger an update on one of them
|
|
expect(
|
|
await [
|
|
rcSet.setDHTValue(rec.key, 3, utf8.encode('BLART')),
|
|
rcSet.setDHTValue(rec.key, 1, utf8.encode('BZORT BZORT'))
|
|
].wait,
|
|
equals([null, null]));
|
|
|
|
await settle(rcSet, rec.key, 3);
|
|
await settle(rcSet, rec.key, 1);
|
|
});
|
|
if (update3 == null) {
|
|
fail('should have a change');
|
|
}
|
|
|
|
// Verify the update came back but we don't get a new value because the
|
|
// sequence number is the same
|
|
expect(update3.key, equals(rec.key));
|
|
expect(update3.count, equals(0xFFFFFFFC));
|
|
expect(update3.subkeys, equals([ValueSubkeyRange.single(3)]));
|
|
expect(update3.value, isNull);
|
|
|
|
// Reopen without closing to change routing context and not lose watch
|
|
rec = await rcWatch.openDHTRecord(rec.key, writer: rec.ownerKeyPair());
|
|
|
|
// Now cancel the update
|
|
expect(
|
|
await rcWatch
|
|
.cancelDHTWatch(rec.key, subkeys: [ValueSubkeyRange.make(3, 9)]),
|
|
isFalse);
|
|
|
|
// Reopen without closing to change routing context and not lose watch
|
|
rec = await rcSet.openDHTRecord(rec.key, writer: rec.ownerKeyPair());
|
|
|
|
// Wait for the update
|
|
final update4 = await waitForValueChange(
|
|
valueChangeQueue.stream, const Duration(seconds: 10), () async {
|
|
// Now set multiple subkeys that should not trigger an update
|
|
expect(
|
|
await [
|
|
rcSet.setDHTValue(rec.key, 3, utf8.encode('BLAH BLAH BLAH BLAH')),
|
|
rcSet.setDHTValue(rec.key, 5, utf8.encode('BZORT BZORT BZORT'))
|
|
].wait,
|
|
equals([null, null]));
|
|
|
|
await settle(rcSet, rec.key, 3);
|
|
await settle(rcSet, rec.key, 5);
|
|
});
|
|
if (update4 != null) {
|
|
fail('should not have a change');
|
|
}
|
|
|
|
// Clean up
|
|
await rcSet.closeDHTRecord(rec.key);
|
|
await rcSet.deleteDHTRecord(rec.key);
|
|
} finally {
|
|
rcWatch.close();
|
|
rcSet.close();
|
|
}
|
|
} finally {
|
|
await valueChangeSubscription.cancel();
|
|
await valueChangeQueue.close();
|
|
}
|
|
}
|
|
|
|
Future<void> testInspectDHTRecord() async {
|
|
final rc = await Veilid.instance.routingContext();
|
|
try {
|
|
final rec = await rc.createDHTRecord(const DHTSchema.dflt(oCnt: 2));
|
|
|
|
expect(await rc.setDHTValue(rec.key, 0, utf8.encode('BLAH BLAH BLAH')),
|
|
isNull);
|
|
|
|
final rr = await rc.inspectDHTRecord(rec.key);
|
|
expect(rr.subkeys, equals([ValueSubkeyRange.make(0, 1)]));
|
|
expect(rr.localSeqs, equals([0, 0xFFFFFFFF]));
|
|
expect(rr.networkSeqs, equals([]));
|
|
|
|
final rr2 =
|
|
await rc.inspectDHTRecord(rec.key, scope: DHTReportScope.syncGet);
|
|
expect(rr2.subkeys, equals([ValueSubkeyRange.make(0, 1)]));
|
|
expect(rr2.localSeqs, equals([0, 0xFFFFFFFF]));
|
|
expect(rr2.networkSeqs, equals([0, 0xFFFFFFFF]));
|
|
|
|
await rc.closeDHTRecord(rec.key);
|
|
await rc.deleteDHTRecord(rec.key);
|
|
} finally {
|
|
rc.close();
|
|
}
|
|
}
|