837 lines
27 KiB
Dart
837 lines
27 KiB
Dart
import 'dart:async';
|
||
import 'dart:convert';
|
||
|
||
import 'package:flutter_test/flutter_test.dart';
|
||
import 'package:fpdart/fpdart.dart';
|
||
import 'package:shared_preferences/shared_preferences.dart';
|
||
import 'package:latlong2/latlong.dart';
|
||
import 'package:maibu_satabot_v2/core/error/failure.dart';
|
||
import 'package:maibu_satabot_v2/core/logging/work_trace_diagnostics.dart';
|
||
import 'package:maibu_satabot_v2/core/network/mqtt/domain/interfaces/mqtt_client.dart';
|
||
import 'package:maibu_satabot_v2/core/network/mqtt/domain/models/mqtt_config.dart';
|
||
import 'package:maibu_satabot_v2/core/network/mqtt/domain/models/mqtt_message.dart';
|
||
import 'package:maibu_satabot_v2/features/autonomous_work/data/v2_work_store.dart';
|
||
import 'package:maibu_satabot_v2/features/autonomous_work/data/v2_work_task_tracker.dart';
|
||
import 'package:maibu_satabot_v2/features/autonomous_work/domain/entities/work_session_models.dart';
|
||
import 'package:maibu_satabot_v2/features/devices/domain/entities/device_schedule_plan_entity.dart';
|
||
import 'package:maibu_satabot_v2/features/devices/domain/entities/device_task_entity.dart';
|
||
import 'package:maibu_satabot_v2/features/devices/domain/repositories/device_task_repository.dart';
|
||
import 'package:maibu_satabot_v2/features/devices/domain/repositories/path_repository.dart';
|
||
import 'package:maibu_satabot_v2/features/devices/domain/usecases/create_device_task_usecase.dart';
|
||
import 'package:maibu_satabot_v2/features/pv_control/presentation/models/work_trace_recorder.dart';
|
||
|
||
Future<void> flush() async {
|
||
for (var i = 0; i < 15; i++) {
|
||
await Future<void>.delayed(Duration.zero);
|
||
}
|
||
}
|
||
|
||
class FakeMqtt implements MqttClient {
|
||
final messages = StreamController<MqttMessage>.broadcast(sync: true);
|
||
final topics = <String>{};
|
||
final configs = <MqttConfig>[];
|
||
@override
|
||
bool isConnected = false;
|
||
@override
|
||
Stream<MqttMessage> get messageStream => messages.stream;
|
||
@override
|
||
Future<void> connect(MqttConfig config) async {
|
||
configs.add(config);
|
||
isConnected = true;
|
||
}
|
||
|
||
@override
|
||
Future<void> disconnect() async {
|
||
isConnected = false;
|
||
}
|
||
|
||
@override
|
||
Future<void> subscribe(String topic) async {
|
||
topics.add(topic);
|
||
}
|
||
|
||
@override
|
||
Future<void> unsubscribe(String topic) async {
|
||
topics.remove(topic);
|
||
}
|
||
|
||
@override
|
||
Future<void> publish(String topic, String message) async {}
|
||
void status(
|
||
String status, {
|
||
int taskId = 42,
|
||
String deviceId = 'MC700-test',
|
||
int? topicId,
|
||
}) {
|
||
messages.add(
|
||
MqttMessage(
|
||
topic: 'task/${topicId ?? taskId}/status',
|
||
payload: jsonEncode({
|
||
'taskId': taskId,
|
||
'deviceId': deviceId,
|
||
'status': status,
|
||
}),
|
||
),
|
||
);
|
||
}
|
||
|
||
void arrive(dynamic entity, {int taskId = 42}) => messages.add(
|
||
MqttMessage(
|
||
topic: 'task/$taskId/arrive',
|
||
payload: jsonEncode({
|
||
'type': 'device_task_arrive_point',
|
||
'deviceId': 'MC700-test',
|
||
'taskId': taskId,
|
||
'entity': entity,
|
||
}),
|
||
),
|
||
);
|
||
}
|
||
|
||
class FakeTasks implements DeviceTaskRepository {
|
||
List<DeviceTaskEntity> tasks = [entity()];
|
||
Completer<Either<Failure, List<DeviceTaskEntity>>>? pending;
|
||
String? error;
|
||
int? operatedTask;
|
||
bool? force;
|
||
static DeviceTaskEntity entity({
|
||
int id = 42,
|
||
String state = 'EXECUTING',
|
||
int routeId = 8,
|
||
}) => DeviceTaskEntity(
|
||
id: id,
|
||
deviceId: 'MC700-test',
|
||
taskStatus: state,
|
||
taskStatusTranslate: '',
|
||
routeId: routeId,
|
||
siteId: 2,
|
||
orgId: 3,
|
||
);
|
||
@override
|
||
Future<Either<Failure, List<DeviceTaskEntity>>> getDeviceTaskPool({
|
||
required String userId,
|
||
required int siteId,
|
||
required int orgId,
|
||
int pageNum = 1,
|
||
int pageSize = 99999999,
|
||
}) async =>
|
||
pending?.future ?? (error == null ? Right(tasks) : Left(Failure(error!)));
|
||
@override
|
||
Future<Either<Failure, bool>> cancelTask({
|
||
required String deviceId,
|
||
required int taskId,
|
||
required int orgId,
|
||
required int siteId,
|
||
bool forceCancel = true,
|
||
}) async {
|
||
operatedTask = taskId;
|
||
force = forceCancel;
|
||
return error == null ? const Right(true) : Left(Failure(error!));
|
||
}
|
||
|
||
@override
|
||
Future<Either<Failure, bool>> pauseTask({
|
||
required String deviceId,
|
||
required int taskId,
|
||
required int orgId,
|
||
required int siteId,
|
||
}) async {
|
||
operatedTask = taskId;
|
||
return error == null ? const Right(true) : Left(Failure(error!));
|
||
}
|
||
|
||
@override
|
||
Future<Either<Failure, Map<String, dynamic>>> recoveryTask({
|
||
required String deviceId,
|
||
required int taskId,
|
||
required int orgId,
|
||
required int siteId,
|
||
}) async {
|
||
operatedTask = taskId;
|
||
return error == null ? const Right({}) : Left(Failure(error!));
|
||
}
|
||
|
||
@override
|
||
Future<Either<Failure, bool>> saveDeviceSchedulePlan({
|
||
required DeviceSchedulePlanEntity plan,
|
||
}) async {
|
||
return error == null ? const Right(true) : Left(Failure(error!));
|
||
}
|
||
}
|
||
|
||
class FakePath implements PathRepository {
|
||
int creations = 0;
|
||
@override
|
||
Future<int> createDeviceTask({
|
||
required String deviceId,
|
||
required int routeId,
|
||
required int siteId,
|
||
required int orgId,
|
||
}) async {
|
||
creations++;
|
||
return 100;
|
||
}
|
||
|
||
@override
|
||
dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
|
||
}
|
||
|
||
void main() {
|
||
TestWidgetsFlutterBinding.ensureInitialized();
|
||
late FakeTasks tasks;
|
||
late FakePath path;
|
||
late FakeMqtt mqtt;
|
||
late V2WorkTaskTracker tracker;
|
||
late StreamSubscription<WorkTaskSnapshot?> sub;
|
||
late List<WorkTaskSnapshot?> updates;
|
||
late List<String> traceLogs;
|
||
late bool failTraceLogs;
|
||
late int failedTraceWrites;
|
||
setUp(() {
|
||
SharedPreferences.setMockInitialValues({
|
||
'saved_device_task_ids': '{"legacy":77}',
|
||
});
|
||
tasks = FakeTasks();
|
||
path = FakePath();
|
||
mqtt = FakeMqtt();
|
||
updates = [];
|
||
traceLogs = [];
|
||
failTraceLogs = false;
|
||
failedTraceWrites = 0;
|
||
tracker = V2WorkTaskTracker(
|
||
repository: tasks,
|
||
createTask: CreateDeviceTaskUseCase(path),
|
||
resolveScope: (_) async => const WorkScope(siteId: 2, orgId: 3),
|
||
userId: () => 'test-user',
|
||
store: V2WorkStore('test-user'),
|
||
mqtt: mqtt,
|
||
diagnostics: WorkTraceDiagnostics(
|
||
now: () => DateTime(2026, 11, 5),
|
||
write: (line) {
|
||
if (failTraceLogs) {
|
||
failedTraceWrites++;
|
||
throw StateError('诊断输出失败');
|
||
}
|
||
traceLogs.add(line);
|
||
},
|
||
),
|
||
);
|
||
sub = tracker.watch('MC700-test').listen(updates.add);
|
||
});
|
||
tearDown(() async {
|
||
await sub.cancel();
|
||
await flush();
|
||
await mqtt.messages.close();
|
||
});
|
||
|
||
test(
|
||
'restore task subscribes with task ID, filters other devices and tasks',
|
||
() async {
|
||
final result = await tracker.load('MC700-test');
|
||
await flush();
|
||
expect(result.isOk, isTrue);
|
||
expect(mqtt.topics, {'task/42/status', 'task/42/arrive'});
|
||
expect(mqtt.configs.single.clientId, 'v2_work_task');
|
||
mqtt.status('FINISH', taskId: 8); // routeId cannot be used as taskId
|
||
mqtt.status('FINISH', deviceId: 'other-device');
|
||
mqtt.status('FINISH', taskId: 99, topicId: 42);
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.running);
|
||
mqtt.status('PAUSE');
|
||
await flush();
|
||
expect(updates.last!.state, WorkRunState.paused);
|
||
},
|
||
);
|
||
|
||
test(
|
||
'FINISH leaves terminal snapshot, removes active ID and preserves legacy cache',
|
||
() async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.status('EXECUTING');
|
||
mqtt.status('FINISH');
|
||
await flush();
|
||
expect(updates.last!.state, WorkRunState.finished);
|
||
expect(mqtt.topics, isEmpty);
|
||
expect(await tracker.history('MC700-test'), hasLength(1));
|
||
expect(
|
||
traceLogs.any((s) => s.contains('[MQTT收到]') && s.contains('FINISH')),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) =>
|
||
s.contains('[结束消息收到]') &&
|
||
s.contains('"source":"mqtt_status"') &&
|
||
s.contains('task/42/status'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) =>
|
||
s.contains('[结束状态落地]') &&
|
||
s.contains('"removedFromActive":true') &&
|
||
s.contains('"inHistory":true'),
|
||
),
|
||
isTrue,
|
||
);
|
||
final prefs = await SharedPreferences.getInstance();
|
||
expect(prefs.getString('saved_device_task_ids'), '{"legacy":77}');
|
||
final stored = await V2WorkStore('test-user').read('tasks');
|
||
expect(stored!['current'], isEmpty);
|
||
final reloaded = await tracker.load('MC700-test');
|
||
expect(
|
||
reloaded.data,
|
||
isEmpty,
|
||
reason: 'stale pool cannot revive a finished task',
|
||
);
|
||
expect(updates.last!.state, WorkRunState.finished);
|
||
},
|
||
);
|
||
|
||
test('MQTT finish wins over an earlier pending HTTP response', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
tasks.pending = Completer<Either<Failure, List<DeviceTaskEntity>>>();
|
||
final loading = tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.status('FINISH');
|
||
tasks.pending!.complete(Right([FakeTasks.entity()]));
|
||
expect((await loading).data, isEmpty);
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
|
||
});
|
||
|
||
test(
|
||
'arrival points accumulate only during execution and empty arrival completes',
|
||
() async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
|
||
mqtt.status('PAUSE');
|
||
mqtt.arrive({'lat': 31.2, 'lng': 120.2});
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
|
||
mqtt.arrive(null);
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) =>
|
||
s.contains('[结束消息收到]') &&
|
||
s.contains('"source":"mqtt_arrive_empty"') &&
|
||
s.contains('"entityPresent":true'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束状态落地]') && s.contains('task/42/arrive'),
|
||
),
|
||
isTrue,
|
||
);
|
||
},
|
||
);
|
||
|
||
test('NEW状态延迟时接受有效到达点，保留原始状态与诊断链路', () async {
|
||
tasks.tasks = [FakeTasks.entity(state: 'NEW')];
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[订阅登记]') && s.contains('task/42/arrive'),
|
||
),
|
||
isTrue,
|
||
);
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
|
||
expect(tracker.current('MC700-test')!.rawStatus, 'NEW');
|
||
expect(tracker.current('MC700-test')!.executionStarted, isTrue);
|
||
expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) =>
|
||
s.contains('[MQTT收到]') &&
|
||
s.contains('task/42/arrive') &&
|
||
s.contains('31.1'),
|
||
),
|
||
isTrue,
|
||
);
|
||
mqtt.status('EXECUTING');
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
mqtt.arrive({'lat': 31.2, 'lng': 120.2});
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(2));
|
||
expect(
|
||
traceLogs.any((s) => s.contains('[状态收到]') && s.contains('EXECUTING')),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any((s) => s.contains('[轨迹追加]') && s.contains('"points":2')),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[任务快照]') && s.contains('"selected":true'),
|
||
),
|
||
isTrue,
|
||
);
|
||
});
|
||
|
||
test('过滤日志说明 topic、ID、type、坐标和解析失败，拒绝消息不修改轨迹', () async {
|
||
tasks.tasks = [FakeTasks.entity(state: 'NEW')];
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
final valid = {
|
||
'type': 'device_task_arrive_point',
|
||
'taskId': 42,
|
||
'deviceId': 'MC700-test',
|
||
'entity': {'lat': 31.1, 'lng': 120.1},
|
||
};
|
||
void send(Map<String, dynamic> data, {String topic = 'task/42/arrive'}) =>
|
||
mqtt.messages.add(MqttMessage(topic: topic, payload: jsonEncode(data)));
|
||
send(valid, topic: 'task/99/arrive');
|
||
send({...valid, 'taskId': 8});
|
||
send({...valid, 'deviceId': 'other'});
|
||
send({...valid, 'type': 'unexpected'});
|
||
send({
|
||
...valid,
|
||
'entity': {'lat': 0, 'lng': 120.1},
|
||
});
|
||
send({...valid, 'entity': 'not a point'});
|
||
mqtt.messages.add(
|
||
MqttMessage(topic: 'task/42/arrive', payload: 'invalid json'),
|
||
);
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, isEmpty);
|
||
expect(tracker.current('MC700-test')!.executionStarted, isFalse);
|
||
expect(tracker.current('MC700-test')!.canRecordTrace, isFalse);
|
||
for (final reason in [
|
||
'topic格式不符或不在当前订阅列表',
|
||
'消息taskId与topic任务不一致',
|
||
'deviceId不匹配',
|
||
'到达点type不匹配',
|
||
'坐标为空、非数值、零或超出范围',
|
||
'entity不是坐标对象',
|
||
'[解析失败]',
|
||
]) {
|
||
expect(traceLogs.any((s) => s.contains(reason)), isTrue, reason: reason);
|
||
}
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
|
||
expect(traceLogs.any((s) => s.contains('与上一轨迹点重复')), isTrue);
|
||
});
|
||
|
||
test('诊断采样只减少日志，连续到达点完整保留；日志写入异常也不影响轨迹', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
for (var i = 0; i < 20; i++) {
|
||
mqtt.arrive({'lat': 31.1 + i * 0.00001, 'lng': 120.1});
|
||
}
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(20));
|
||
expect(traceLogs.where((s) => s.contains('[MQTT收到]')), hasLength(1));
|
||
failTraceLogs = true;
|
||
mqtt.status('PAUSE');
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.paused);
|
||
mqtt.status('EXECUTING');
|
||
mqtt.arrive({'lat': 31.2, 'lng': 120.2});
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(21));
|
||
expect(failedTraceWrites, greaterThan(0));
|
||
});
|
||
|
||
test(
|
||
'new task survives empty pool delay and a new watcher receives its state',
|
||
() async {
|
||
tasks.tasks = [];
|
||
final created = await tracker.start(
|
||
const WorkStartRequest(
|
||
deviceId: 'MC700-test',
|
||
route: WorkRouteOption(id: '8', name: 'route'),
|
||
),
|
||
);
|
||
await flush();
|
||
expect(created.data!.taskId, '100');
|
||
expect(created.data!.rawStatus, 'NEW');
|
||
expect(created.data!.executionStarted, isTrue);
|
||
expect(created.data!.canRecordTrace, isTrue);
|
||
expect(mqtt.topics, {'task/100/status', 'task/100/arrive'});
|
||
final secondUpdates = <WorkTaskSnapshot?>[];
|
||
final second = tracker.watch('MC700-test').listen(secondUpdates.add);
|
||
await flush();
|
||
expect(secondUpdates.single!.taskId, '100');
|
||
await second.cancel();
|
||
await flush();
|
||
expect(mqtt.topics, isNotEmpty);
|
||
expect(path.creations, 1);
|
||
final refreshed = await tracker.load('MC700-test');
|
||
expect(refreshed.data!.single.taskId, '100');
|
||
expect(tracker.current('MC700-test')!.taskId, '100');
|
||
},
|
||
);
|
||
|
||
test(
|
||
'pause resume stop use current task ID and stop cleans subscriptions',
|
||
() async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
var current = tracker.current('MC700-test')!;
|
||
current = (await tracker.command('MC700-test', current, 'pause')).data!;
|
||
expect(current.state, WorkRunState.paused);
|
||
current = (await tracker.command('MC700-test', current, 'resume')).data!;
|
||
expect(current.state, WorkRunState.running);
|
||
current = (await tracker.command('MC700-test', current, 'cancel')).data!;
|
||
await flush();
|
||
expect(tasks.operatedTask, 42);
|
||
expect(tasks.force, isTrue);
|
||
expect(current.rawStatus, 'CANCELED');
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束状态落地]') && s.contains('"source":"http_cancel"'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(traceLogs.where((s) => s.contains('[结束消息收到]')), isEmpty);
|
||
expect(mqtt.topics, isEmpty);
|
||
expect(
|
||
(await tracker.command('MC700-test', current, 'resume')).isOk,
|
||
isFalse,
|
||
);
|
||
},
|
||
);
|
||
|
||
test('只选中NEW任务不记录GPS，确认开始复用任务后才记录', () async {
|
||
tasks.tasks = [FakeTasks.entity(state: 'NEW')];
|
||
await tracker.load('MC700-test');
|
||
final selected = await tracker.select(tracker.current('MC700-test')!);
|
||
expect(selected.data!.canRecordTrace, isFalse);
|
||
final started = await tracker.start(
|
||
const WorkStartRequest(
|
||
deviceId: 'MC700-test',
|
||
route: WorkRouteOption(id: '8', name: 'route'),
|
||
reuseTaskId: '42',
|
||
),
|
||
);
|
||
expect(started.data!.rawStatus, 'NEW');
|
||
expect(started.data!.executionStarted, isTrue);
|
||
expect(started.data!.canRecordTrace, isTrue);
|
||
expect(path.creations, 0);
|
||
await tracker.load('MC700-test');
|
||
expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
|
||
});
|
||
|
||
test('开始标记持久化，重进页面遇到NEW也不会丢掉轨迹接收权限', () async {
|
||
tasks.tasks = [FakeTasks.entity(state: 'NEW')];
|
||
await tracker.load('MC700-test');
|
||
await tracker.start(
|
||
const WorkStartRequest(
|
||
deviceId: 'MC700-test',
|
||
route: WorkRouteOption(id: '8', name: 'route'),
|
||
reuseTaskId: '42',
|
||
),
|
||
);
|
||
final stored = await V2WorkStore('test-user').read('tasks');
|
||
expect((stored!['current'] as List).single['executionStarted'], isTrue);
|
||
await sub.cancel();
|
||
await flush();
|
||
tracker = V2WorkTaskTracker(
|
||
repository: tasks,
|
||
createTask: CreateDeviceTaskUseCase(path),
|
||
resolveScope: (_) async => const WorkScope(siteId: 2, orgId: 3),
|
||
userId: () => 'test-user',
|
||
store: V2WorkStore('test-user'),
|
||
mqtt: mqtt,
|
||
diagnostics: WorkTraceDiagnostics(write: traceLogs.add),
|
||
);
|
||
sub = tracker.watch('MC700-test').listen(updates.add);
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.rawStatus, 'NEW');
|
||
expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
|
||
});
|
||
|
||
for (final status in ['EXECUTING', 'PAUSE', 'FAILED']) {
|
||
test('$status之后的延迟NEW不能回退执行状态或解除暂停', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.status(status);
|
||
mqtt.status('NEW');
|
||
expect(tracker.current('MC700-test')!.rawStatus, status);
|
||
tasks.tasks = [FakeTasks.entity(state: 'NEW')];
|
||
await tracker.load('MC700-test');
|
||
expect(tracker.current('MC700-test')!.rawStatus, status);
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
expect(
|
||
tracker.current('MC700-test')!.completedPoints,
|
||
hasLength(status == 'EXECUTING' ? 1 : 0),
|
||
);
|
||
mqtt.status('EXECUTING');
|
||
mqtt.arrive({'lat': 31.2, 'lng': 120.2});
|
||
expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
|
||
mqtt.status('FINISH');
|
||
mqtt.status('NEW');
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
|
||
});
|
||
}
|
||
|
||
test('开始、NEW到达、实时车位、暂停与结束连通，GPS不写入任务到达点', () async {
|
||
tasks.tasks = [];
|
||
await tracker.start(
|
||
const WorkStartRequest(
|
||
deviceId: 'MC700-test',
|
||
route: WorkRouteOption(id: '8', name: 'route'),
|
||
),
|
||
);
|
||
await flush();
|
||
final trace = WorkTraceRecorder();
|
||
void sync() {
|
||
final current = tracker.current('MC700-test')!;
|
||
trace.synchronize(
|
||
deviceId: current.deviceId,
|
||
input: WorkTraceInput(
|
||
taskId: current.taskId,
|
||
recording: current.canRecordTrace,
|
||
visible: current.state != WorkRunState.finished,
|
||
completedPoints: current.completedPoints,
|
||
),
|
||
);
|
||
}
|
||
|
||
const a = LatLng(31.1, 120.1);
|
||
const b = LatLng(31.1001, 120.1001);
|
||
const c = LatLng(31.1002, 120.1002);
|
||
sync();
|
||
trace.addPosition(a);
|
||
trace.addPosition(b);
|
||
expect(trace.points, [a, b]);
|
||
mqtt.arrive({'lat': b.latitude, 'lng': b.longitude}, taskId: 100);
|
||
sync();
|
||
trace.addPosition(c);
|
||
expect(trace.points, [b, c]);
|
||
expect(tracker.current('MC700-test')!.completedPoints, [b]);
|
||
await tracker.command(
|
||
'MC700-test',
|
||
tracker.current('MC700-test')!,
|
||
'pause',
|
||
);
|
||
mqtt.status('NEW', taskId: 100);
|
||
sync();
|
||
expect(trace.addPosition(a), isFalse);
|
||
expect(trace.points, [b, c]);
|
||
mqtt.status('FINISH', taskId: 100);
|
||
sync();
|
||
expect(trace.points, isEmpty);
|
||
expect((await tracker.history('MC700-test')).single.completedPoints, [b]);
|
||
});
|
||
|
||
test('failed query cannot create or erase a running task', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
tasks.error = 'offline';
|
||
expect((await tracker.load('MC700-test')).isOk, isFalse);
|
||
expect(tracker.current('MC700-test')!.taskId, '42');
|
||
expect(path.creations, 0);
|
||
});
|
||
|
||
test(
|
||
'V2 draft store serializes writes and isolates users from legacy keys',
|
||
() async {
|
||
final store = V2WorkStore('first');
|
||
final points = <int>[1, 2, 3];
|
||
final first = store.write('editor/2/device', {'points': points});
|
||
points.clear();
|
||
await first;
|
||
expect((await store.read('editor/2/device'))!['points'], [1, 2, 3]);
|
||
expect(await V2WorkStore('second').read('editor/2/device'), isNull);
|
||
await store.write('editor/2/device', null);
|
||
expect(await store.read('editor/2/device'), isNull);
|
||
},
|
||
);
|
||
test(
|
||
'missed FINISH push is recovered from task pool with its arrival points',
|
||
() async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.arrive({'lat': 31.1, 'lng': 120.1});
|
||
await flush();
|
||
tasks.tasks = [FakeTasks.entity(state: 'FINISH')];
|
||
expect((await tracker.load('MC700-test')).data, isEmpty);
|
||
await flush();
|
||
expect(updates.last!.rawStatus, 'FINISH');
|
||
expect(updates.last!.completedPoints, hasLength(1));
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束任务池确认]') && s.contains('"source":"task_pool"'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束状态落地]') && s.contains('"source":"task_pool"'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(traceLogs.where((s) => s.contains('[结束消息收到]')), isEmpty);
|
||
},
|
||
);
|
||
|
||
test(
|
||
'queue receives other task completion without replacing current task',
|
||
() async {
|
||
tasks.tasks = [
|
||
FakeTasks.entity(),
|
||
FakeTasks.entity(id: 43, state: 'NEW'),
|
||
];
|
||
final changes = <WorkTaskSnapshot?>[];
|
||
final queue = tracker
|
||
.watch('MC700-test', allTasks: true)
|
||
.listen(changes.add);
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.status('FINISH', taskId: 43);
|
||
await flush();
|
||
expect(changes.last!.taskId, '43');
|
||
expect(changes.last!.state, WorkRunState.finished);
|
||
expect(updates.last!.taskId, '42');
|
||
expect(updates.last!.state, WorkRunState.running);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) =>
|
||
s.contains('[结束状态落地]') &&
|
||
s.contains('"taskId":"43"') &&
|
||
s.contains('"selected":false'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(mqtt.topics, {'task/42/status', 'task/42/arrive'});
|
||
await queue.cancel();
|
||
},
|
||
);
|
||
|
||
for (final status in ['CANCELED', 'CANCELLED']) {
|
||
test('$status结束信号独立记录并保留原有结束处理', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.status(status);
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
|
||
expect(
|
||
traceLogs.any((s) => s.contains('[结束消息收到]') && s.contains(status)),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束状态落地]') && s.contains('"source":"mqtt_status"'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(mqtt.topics, isEmpty);
|
||
});
|
||
}
|
||
|
||
test('结束消息过滤记录ID不匹配及重复原因，不重复完成任务', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.status('FINISH', deviceId: 'other-device');
|
||
mqtt.status('FINISH', taskId: 99, topicId: 42);
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.running);
|
||
expect(
|
||
traceLogs.any((s) => s.contains('[结束消息过滤]') && s.contains('deviceId不匹配')),
|
||
isTrue,
|
||
);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束消息过滤]') && s.contains('消息taskId与topic任务不一致'),
|
||
),
|
||
isTrue,
|
||
);
|
||
mqtt.status('FINISH');
|
||
mqtt.status('FINISH');
|
||
await flush();
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束消息过滤]') && s.contains('任务已结束，忽略重复或迟到消息'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(traceLogs.where((s) => s.contains('[结束状态落地]')), hasLength(1));
|
||
expect(
|
||
updates.where((t) => t?.state == WorkRunState.finished),
|
||
hasLength(1),
|
||
);
|
||
expect(await tracker.history('MC700-test'), hasLength(1));
|
||
});
|
||
|
||
test('空到达点日志记录entity是否缺失，保持原有结束语义', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
mqtt.messages.add(
|
||
MqttMessage(
|
||
topic: 'task/42/arrive',
|
||
payload: jsonEncode({
|
||
'type': 'device_task_arrive_point',
|
||
'deviceId': 'MC700-test',
|
||
'taskId': 42,
|
||
}),
|
||
),
|
||
);
|
||
await flush();
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) => s.contains('[结束消息收到]') && s.contains('"entityPresent":false'),
|
||
),
|
||
isTrue,
|
||
);
|
||
});
|
||
|
||
test('未收到结束消息时定期说明等待状态，不能把消息静默当成已结束', () async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
expect(mqtt.isConnected, isTrue);
|
||
// 真实异步订阅与计时器共用同一时钟，验证生产中的10秒诊断周期。
|
||
await Future<void>.delayed(const Duration(seconds: 11));
|
||
expect(tracker.current('MC700-test')!.state, WorkRunState.running);
|
||
expect(
|
||
traceLogs.any(
|
||
(s) =>
|
||
s.contains('[结束确认等待]') &&
|
||
s.contains('task/42/status') &&
|
||
s.contains('"statusReceived":0'),
|
||
),
|
||
isTrue,
|
||
);
|
||
expect(traceLogs.where((s) => s.contains('[结束状态落地]')), isEmpty);
|
||
await sub.cancel();
|
||
await flush();
|
||
expect(mqtt.topics, isEmpty);
|
||
});
|
||
|
||
test(
|
||
'last page exit unsubscribes and reentry resubscribes the active task',
|
||
() async {
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
await sub.cancel();
|
||
await flush();
|
||
expect(mqtt.topics, isEmpty);
|
||
expect(mqtt.isConnected, isFalse);
|
||
sub = tracker.watch('MC700-test').listen(updates.add);
|
||
await tracker.load('MC700-test');
|
||
await flush();
|
||
expect(mqtt.topics, {'task/42/status', 'task/42/arrive'});
|
||
mqtt.status('FINISH');
|
||
await flush();
|
||
expect(updates.last!.state, WorkRunState.finished);
|
||
},
|
||
);
|
||
}
|