Files
photo-live/apps/mobile/test/features/uploads/upload_engine_test.dart
T
浅念 29f3bde9c2 feat(app): wire server dedup - force passthrough + PHOTO_DUPLICATE classification & l10n
Task 8: close the gap between server-side duplicate detection (90dcc58)
and the app.

- ApiClient.mapDioException: normalize 409 wire body reusing
  VALIDATION_FAILED with message=PHOTO_DUPLICATE to
  ErrorCodes.photoDuplicate (message-based; auto-compat if the contract
  later adds a dedicated code)
- l10n zh/en errorPhotoDuplicate; error_l10n maps the code (queue page
  inherits via fallback)
- UploadEngine._classifyFailure: PHOTO_DUPLICATE joins the no-auto-retry
  bucket (retrying the same init body would 409 again); force to skip
  dedup travels only through the enqueue path
- UploadTasks drift table +force column (schema v2, addColumn migration;
  legacy rows default false); UploadTaskDraft/UploadTask/prepare carry it
- enqueueForUpload +forcePaths; duplicate dialog "全部上传" (capture page
  + album detail page) now returns DuplicateDialogChoice so confirmed
  duplicates enqueue with force=true (server adjudicates owner/admin)
- engine init body carries force only when true (absent otherwise)

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-28 17:06:19 +08:00

1165 lines
41 KiB
Dart
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/// 上传引擎(UploadEngine)测试(P3 计划 Task 6)。
///
/// 打桩面:FakeAdapter ×2(API 侧 /v1/* 与对象侧 presign PUT)+ 内存
/// TaskStore + 真实临时文件 + 可变时钟。覆盖:快乐路径调用序列、幂等
/// 重试、断网恢复、退避与自动重试上限、409/401/403 分类、取消、账号
/// 隔离、A15 归属不变、并行上限、崩溃恢复、进度回调、清理执行器。
library;
import 'dart:async';
import 'dart:collection';
import 'dart:io';
import 'package:dio/dio.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:photo_live/core/api_client.dart';
import 'package:photo_live/core/device_id.dart';
import 'package:photo_live/core/error_codes.dart';
import 'package:photo_live/core/session_store.dart';
import 'package:photo_live/features/uploads/native_transport.dart';
import 'package:photo_live/features/uploads/preparer.dart';
import 'package:photo_live/features/uploads/task_state.dart';
import 'package:photo_live/features/uploads/task_store.dart';
import 'package:photo_live/features/uploads/upload_engine.dart';
import '../../helpers/fakes.dart';
const String _kDeviceId = '0123456789abcdef0123456789abcdef';
/// 引擎测试装配:双适配器假服务 + 内存库 + 临时文件 + 可变时钟。
class Harness {
Harness() {
apiAdapter = FakeAdapter(_apiRouter);
objectAdapter = FakeAdapter(_objectRouter);
api = ApiClient(
sessionStore: sessionStore,
dio: Dio(BaseOptions(baseUrl: 'http://api.test.local'))
..httpClientAdapter = apiAdapter,
);
objectDio = Dio()..httpClientAdapter = objectAdapter;
store = TaskStore.memory();
deviceId = DeviceId(store: _FixedDeviceIdStore(_kDeviceId));
}
late final FakeAdapter apiAdapter;
late final FakeAdapter objectAdapter;
late final ApiClient api;
late final Dio objectDio;
late final TaskStore store;
late final DeviceId deviceId;
final FakeSessionStore sessionStore = FakeSessionStore()
..session = Session(token: 'tok-1', account: 'alice');
/// 可变时钟(测试推进)。
DateTime clock = DateTime.utc(2026, 9, 19, 12, 0, 0);
/// 服务端调用事件序列(跨适配器统一断言)。
final List<String> events = <String>[];
/// clientTaskId → init 响应体(幂等:同键回原会话)。
final Map<String, Map<String, dynamic>> sessionsByClientTaskId =
<String, Map<String, dynamic>>{};
/// uploadId → init 响应体(renew/complete 回显)。
final Map<String, Map<String, dynamic>> sessionsByUploadId =
<String, Map<String, dynamic>>{};
final List<Map<String, dynamic>> initBodies = <Map<String, dynamic>>[];
int renewCalls = 0;
// ── 故障注入旋钮(次数制)──
int initNetworkFailures = 0;
int initUnauthorized = 0;
int initAlbumDisabled = 0;
/// initUpload 查重 409 次数(Task 8:wire 体复用 VALIDATION_FAILED code、
/// message=PHOTO_DUPLICATE)。
int initDuplicateRejects = 0;
int completeVerifyFailures = 0;
int completeNetworkFailures = 0;
int putNetworkFailures = 0;
int put403Count = 0;
/// PUT 闸门(并行/取消测试:挂起全部 PUT 直到放开)。
Completer<void>? putGate;
UploadEngine buildEngine({
int maxParallel = 2,
int keepCompletedDays = 14,
void Function(String taskId, int sent, int total)? onProgress,
Transport? transport,
}) =>
UploadEngine(
store,
api,
deviceId,
objectDio: transport == null ? objectDio : null,
transport: transport,
now: () => clock,
maxParallel: maxParallel,
keepCompletedDays: keepCompletedDays,
onProgress: onProgress,
);
Map<String, dynamic> _initBodyFor(String uploadId, String photoId,
{bool renewed = false}) {
return <String, dynamic>{
'uploadId': uploadId,
'photoId': photoId,
'objectKey': 'photo-live/dev/staging/$uploadId/$photoId.jpg',
'credential': <String, dynamic>{
'mode': 'presigned',
'url':
'http://object.test.local/staging/$uploadId/$photoId.jpg'
'${renewed ? '?renewed=1' : ''}',
'method': 'PUT',
'headers': {'Content-Type': 'image/jpeg'},
'expiresAt': '2026-09-19T13:00:00.000Z',
},
};
}
FutureOr<(int, Object?)> _apiRouter(RequestOptions o) {
if (o.method == 'POST' && o.path == '/v1/uploads') {
final body = Map<String, dynamic>.from(o.data as Map<dynamic, dynamic>);
initBodies.add(body);
final ct = body['clientTaskId'] as String;
if (initAlbumDisabled > 0) {
initAlbumDisabled--;
return (409, errorBody(ErrorCodes.albumUploadDisabled));
}
if (initDuplicateRejects > 0) {
initDuplicateRejects--;
// P0-1 契约现状:查重拒绝 409 的 code 复用 VALIDATION_FAILED,
// 靠 message=PHOTO_DUPLICATE 识别(Task 8 客户端归一依据)。
return (
409,
<String, dynamic>{
'code': ErrorCodes.validationFailed,
'message': 'PHOTO_DUPLICATE',
'requestId': 'req-dup',
},
);
}
if (initUnauthorized > 0) {
initUnauthorized--;
return (401, errorBody(ErrorCodes.authTokenExpired));
}
if (initNetworkFailures > 0) {
initNetworkFailures--;
throw DioException(
requestOptions: o,
type: DioExceptionType.connectionError,
);
}
final existing = sessionsByClientTaskId[ct];
if (existing != null) {
events.add('init:$ct(idem)');
return (201, existing);
}
final n = sessionsByClientTaskId.length + 1;
final uploadId = 'upl$n';
final photoId = 'pho$n';
final initBody = _initBodyFor(uploadId, photoId);
sessionsByClientTaskId[ct] = initBody;
sessionsByUploadId[uploadId] = initBody;
events.add('init:$ct');
return (201, initBody);
}
final renew =
RegExp(r'^/v1/uploads/([^/]+)/renew$').firstMatch(o.path);
if (renew != null) {
renewCalls++;
final uploadId = renew.group(1)!;
events.add('renew:$uploadId');
final original = sessionsByUploadId[uploadId]!;
return (
200,
_initBodyFor(uploadId, original['photoId'] as String, renewed: true),
);
}
final complete =
RegExp(r'^/v1/uploads/([^/]+)/complete$').firstMatch(o.path);
if (complete != null) {
final uploadId = complete.group(1)!;
if (completeVerifyFailures > 0) {
completeVerifyFailures--;
events.add('complete-fail:$uploadId');
return (400, errorBody(ErrorCodes.uploadVerifyFailed));
}
if (completeNetworkFailures > 0) {
completeNetworkFailures--;
throw DioException(
requestOptions: o,
type: DioExceptionType.connectionError,
);
}
events.add('complete:$uploadId');
final original = sessionsByUploadId[uploadId]!;
return (
200,
<String, dynamic>{
'photoId': original['photoId'],
'processingState': 'processing',
},
);
}
throw StateError('意外的 API 请求:${o.method} ${o.path}');
}
FutureOr<(int, Object?)> _objectRouter(RequestOptions o) async {
events.add('put');
final gate = putGate;
if (gate != null) {
await gate.future;
}
if (put403Count > 0) {
put403Count--;
return (403, null);
}
if (putNetworkFailures > 0) {
putNetworkFailures--;
throw DioException(
requestOptions: o,
type: DioExceptionType.connectionError,
);
}
return (200, null);
}
/// 种子一条 queued 任务(真实临时文件做源)。
Future<UploadTask> seedTask(
File file, {
String accountId = 'alice',
String albumId = 'alb-1',
String? categoryId,
String clientTaskId = 'ct-default',
DateTime? createdAt,
bool force = false,
}) async {
final draft = UploadTaskDraft(
id: 'task-$clientTaskId',
accountId: accountId,
albumId: albumId,
categoryId: categoryId,
clientTaskId: clientTaskId,
sourceFilePath: file.path,
filename: file.uri.pathSegments.last,
bytes: await file.length(),
contentType: 'image/jpeg',
force: force,
);
final task = await store.create(draft, now: createdAt);
return store.updateState(
task.id,
UploadTaskState.queued,
now: createdAt,
);
}
}
class _FixedDeviceIdStore implements DeviceIdStore {
_FixedDeviceIdStore(this._value);
final String _value;
@override
Future<String?> read() async => _value;
@override
Future<void> write(String value) async {}
}
/// 一次 send 调用的快照(Task 7 传输委托断言)。
class ScriptedTransportCall {
const ScriptedTransportCall(
this.taskId,
this.url,
this.method,
this.headers,
this.filePath,
this.metadata,
);
final String taskId;
final String url;
final String method;
final Map<String, String> headers;
final String filePath;
final Map<String, Object?>? metadata;
}
/// 脚本化传输替身(P3 计划 Task 7:引擎 × Transport 接口集成)。
///
/// 每任务按队列依次弹出手本:[TransportResult] 即成功返回;其余视为
/// 异常抛出(TransportException 族)。无手本任务默认成功并补发一次
/// 全量进度回调。
class ScriptedTransport implements Transport {
final Map<String, ListQueue<Object>> script = {};
final List<ScriptedTransportCall> calls = [];
bool progressBeforeComplete = true;
void enqueue(String taskId, List<Object> outcomes) =>
script.putIfAbsent(taskId, ListQueue.new).addAll(outcomes);
@override
Future<TransportResult> send({
required String taskId,
required String url,
required String method,
required Map<String, String> headers,
required String filePath,
Map<String, Object?>? metadata,
void Function(int sent, int total)? onProgress,
CancelToken? cancelToken,
}) async {
calls.add(
ScriptedTransportCall(taskId, url, method, headers, filePath, metadata),
);
final queue = script[taskId];
final outcome = queue == null || queue.isEmpty
? const TransportResult()
: queue.removeFirst();
if (outcome is TransportResult) {
if (progressBeforeComplete) {
onProgress?.call(1024, 1024);
}
return outcome;
}
throw outcome;
}
@override
void dispose() {}
}
void main() {
late Harness h;
late Directory tmp;
late UploadEngine engine;
setUp(() async {
h = Harness();
tmp = await Directory.systemTemp.createTemp('engine-test-');
});
tearDown(() async {
engine.dispose();
await h.store.close();
// Windows:取消竞态下 dio 会孤儿化适配器 fetch(CancelableOperation
// 置 cancel),被孤儿化的 drain 仍短暂持有文件句柄(errno 32)——
// 短重试兜底等待句柄释放。
for (var i = 0; i < 20; i++) {
try {
await tmp.delete(recursive: true);
break;
} on FileSystemException {
await Future<void>.delayed(const Duration(milliseconds: 50));
}
}
});
File photoFile(String name, [int size = 1024]) {
final f = File('${tmp.path}/$name');
f.writeAsBytesSync(List<int>.generate(size, (i) => i % 251));
return f;
}
/// 驱动一轮完整调度并等待收敛(drain 折叠并发调用;双循环兜底
/// 完成回调触发的补位 drain)。
Future<void> cycle() async {
for (var i = 0; i < 3; i++) {
await engine.drain();
await engine.settle();
}
await engine.drain();
}
/// 泵事件循环直到条件成立(真实 IO/微任务链推进用)。
Future<void> until(bool Function() cond) async {
for (var i = 0; i < 100 && !cond(); i++) {
await pumpEventQueue();
}
expect(cond(), isTrue, reason: '等待条件在泵内未成立');
}
Future<UploadTask> task(String id) async {
final t = await h.store.get('alice', id);
if (t == null) {
throw StateError('任务不存在:$id');
}
return t;
}
group('快乐路径', () {
test('全序列 init → PUT → complete → completed;服务端调用与请求体断言',
() async {
final file = photoFile('a.jpg');
final progress = <String, List<int>>{};
engine = h.buildEngine(onProgress: (id, sent, total) => progress
.putIfAbsent(id, () => <int>[])
.add(sent));
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-1');
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(t.serverUploadId, 'upl1');
expect(t.serverPhotoId, 'pho1');
expect(t.lastError, isNull);
expect(t.nextRetryAt, isNull);
expect(h.events, ['init:ct-1', 'put', 'complete:upl1']);
// init 请求体:幂等键 + sourceDeviceId(A19)+ 元数据。
final body = h.initBodies.single;
expect(body['albumId'], 'alb-1');
expect(body['clientTaskId'], 'ct-1');
expect(body['filename'], 'a.jpg');
expect(body['bytes'], 1024);
expect(body['contentType'], 'image/jpeg');
expect(body['sourceDeviceId'], _kDeviceId);
expect(body.containsKey('force'), isFalse,
reason: '非强传任务不下发 force(缺省语义)');
// PUT:预签名 URL + 凭证头 + Content-Length;无 Authorization
// (裸 Dio——Bearer 会破坏 MinIO 预签名)。
final put = h.objectAdapter.requests.single;
expect(put.method, 'PUT');
expect(put.uri.toString(),
'http://object.test.local/staging/upl1/pho1.jpg');
expect(put.headers['Content-Type'], 'image/jpeg');
// dio 以原始键位(小写 const Headers.contentLengthHeader)落请求头。
expect(put.headers['content-length'], 1024);
expect(put.headers.containsKey('Authorization'), isFalse);
// 源文件即时保留(清理执行器按保留期处理,不随完成删除)。
expect(file.existsSync(), isTrue);
// 进度回调按真实字节计数上报(计满 1024)。
expect(progress[seeded.id]?.last, 1024);
});
test('force 任务:init 请求体携带 force:true(Task 8 查重强传透传)',
() async {
final file = photoFile('force-1.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(
file,
clientTaskId: 'ct-force-1',
force: true,
);
await cycle();
expect((await task(seeded.id)).state, UploadTaskState.completed);
expect(h.initBodies.single['force'], isTrue,
reason: '弹窗「全部上传」确认的强传必须到达服务端');
});
test('与准备器集成:preparing → queued → completed(受管副本上传)',
() async {
final src = photoFile('src.jpg');
final managed = Directory('${tmp.path}/managed')..createSync();
final preparer = UploadPreparer(
h.store,
baseDirProvider: () async => managed,
);
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final queued = await preparer.prepare(
accountId: 'alice',
albumId: 'alb-1',
sourcePath: src.path,
);
expect(queued.state, UploadTaskState.queued);
await cycle();
final t = await task(queued.id);
expect(t.state, UploadTaskState.completed);
// PUT 的 body 是受管副本(大小一致,路径在受管目录下)。
expect(t.sourceFilePath, contains('managed'));
expect(File(t.sourceFilePath).lengthSync(), 1024);
expect(h.events.last, 'complete:upl1');
});
});
group('幂等重试', () {
test('PUT 中途网络失败 → waitingNetwork → 到期重试:init 复用 '
'clientTaskId、服务端同会话同照片,最终单照片 completed', () async {
final file = photoFile('b.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.putNetworkFailures = 1;
final seeded = await h.seedTask(file, clientTaskId: 'ct-b');
await cycle();
var t = await task(seeded.id);
expect(t.state, UploadTaskState.waitingNetwork);
expect(t.attempt, 1);
expect(t.serverUploadId, 'upl1', reason: 'init 已成功,会话 ID 保留');
expect(h.events, ['init:ct-b', 'put']);
// 未到退避时间(5s·2^0):不重试。
h.clock = h.clock.add(const Duration(seconds: 2));
await cycle();
expect(h.events, ['init:ct-b', 'put'], reason: '退避未到不重试');
// 到期重试:init 幂等(同 clientTaskId 返回原会话)→ 单照片完成。
h.clock = h.clock.add(const Duration(seconds: 10));
await cycle();
t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(t.serverPhotoId, 'pho1');
expect(h.sessionsByClientTaskId, hasLength(1), reason: '服务端只建了一个会话');
expect(h.events, ['init:ct-b', 'put', 'init:ct-b(idem)', 'put', 'complete:upl1']);
expect(h.initBodies, hasLength(2));
expect(h.initBodies.map((b) => b['clientTaskId']).toSet(), ['ct-b']);
});
test('complete 网络失败 → waitingNetwork;重试只补 complete(PUT 已达)',
() async {
final file = photoFile('c.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.completeNetworkFailures = 1;
final seeded = await h.seedTask(file, clientTaskId: 'ct-c');
await cycle();
var t = await task(seeded.id);
expect(t.state, UploadTaskState.waitingNetwork);
// 网络失败(非核验失败)不产生服务端事件:序列止于 PUT。
expect(h.events, ['init:ct-c', 'put']);
expect(h.events.where((e) => e.startsWith('complete')), isEmpty);
h.clock = h.clock.add(const Duration(seconds: 30));
await cycle();
t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
// 重试路径:init 幂等 + PUT 重传(服务端对象覆盖写)+ complete。
expect(h.events, ['init:ct-c', 'put', 'init:ct-c(idem)', 'put', 'complete:upl1']);
});
});
group('断网恢复', () {
test('网络失败退避等待;onNetworkRestored 立即绕过退避完成', () async {
final file = photoFile('d.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.initNetworkFailures = 1;
final seeded = await h.seedTask(file, clientTaskId: 'ct-d');
await cycle();
var t = await task(seeded.id);
expect(t.state, UploadTaskState.waitingNetwork);
expect(t.lastError, ErrorCodes.networkError);
expect(t.nextRetryAt, h.clock.add(const Duration(seconds: 5)));
// 时钟未推进(退避未到),但网络恢复信号触发立即重试。
engine.onNetworkRestored();
await cycle();
t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(h.initBodies, hasLength(2));
});
});
group('退避曲线与自动重试上限', () {
test('backoffFor:5s·2^(n-1) 封顶 10min', () {
expect(UploadEngine.backoffFor(1), const Duration(seconds: 5));
expect(UploadEngine.backoffFor(2), const Duration(seconds: 10));
expect(UploadEngine.backoffFor(3), const Duration(seconds: 20));
expect(UploadEngine.backoffFor(7), const Duration(seconds: 320));
// 5s·2^7=640s 越过封顶 → 600s。
expect(UploadEngine.backoffFor(8), const Duration(minutes: 10));
expect(UploadEngine.backoffFor(100), const Duration(minutes: 10));
expect(UploadEngine.backoffFor(0), const Duration(seconds: 5));
});
test('UPLOAD_VERIFY_FAILED 三次后停止自动重试(nextRetryAt=null 终止符)',
() async {
final file = photoFile('e.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.completeVerifyFailures = 3;
final seeded = await h.seedTask(file, clientTaskId: 'ct-e');
await cycle();
var t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(t.attempt, 1);
expect(t.lastError, ErrorCodes.uploadVerifyFailed);
expect(t.nextRetryAt, isNotNull);
h.clock = h.clock.add(const Duration(seconds: 10));
await cycle();
t = await task(seeded.id);
expect(t.attempt, 2);
h.clock = h.clock.add(const Duration(seconds: 30));
await cycle();
t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(t.attempt, 3);
expect(t.nextRetryAt, isNull, reason: '达上限:不再自动重试');
final initCount = h.initBodies.length;
h.clock = h.clock.add(const Duration(hours: 1));
await cycle();
expect(h.initBodies.length, initCount, reason: '终止符后不再出队');
expect((await task(seeded.id)).state, UploadTaskState.retryableFailed);
});
test('retryNow 手动重试可复活终止符任务', () async {
final file = photoFile('e2.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.completeVerifyFailures = 3;
final seeded = await h.seedTask(file, clientTaskId: 'ct-e2');
await cycle();
h.clock = h.clock.add(const Duration(seconds: 10));
await cycle();
h.clock = h.clock.add(const Duration(seconds: 30));
await cycle();
var t = await task(seeded.id);
expect(t.nextRetryAt, isNull);
h.completeVerifyFailures = 0; // 服务端修复。
await engine.retryNow('alice', seeded.id);
await cycle();
t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
});
});
group('服务端状态分类', () {
test('409 ALBUM_UPLOAD_DISABLED → retryableFailed 不自动重试', () async {
final file = photoFile('f.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.initAlbumDisabled = 1;
final seeded = await h.seedTask(file, clientTaskId: 'ct-f');
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(t.lastError, ErrorCodes.albumUploadDisabled);
expect(t.nextRetryAt, isNull);
expect(t.attempt, 0, reason: '客户端重试无意义,不烧次数');
final initCount = h.initBodies.length;
h.clock = h.clock.add(const Duration(hours: 1));
await cycle();
expect(h.initBodies.length, initCount, reason: '无自动重试');
});
test('409 PHOTO_DUPLICATE(服务端查重拒绝)→ retryableFailed 不自动重试',
() async {
final file = photoFile('dup.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.initDuplicateRejects = 99; // 永远 409(重试同样被拒)。
final seeded = await h.seedTask(file, clientTaskId: 'ct-dup');
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
// 归一后 lastError=PHOTO_DUPLICATE(此前缺陷:VALIDATION_FAILED 落
// 桶 4 自动重试,文案误导)。
expect(t.lastError, ErrorCodes.photoDuplicate);
expect(t.nextRetryAt, isNull, reason: '重试必然再 409:不自动重试');
expect(t.attempt, 0, reason: '客户端重试无意义,不烧次数');
final initCount = h.initBodies.length;
h.clock = h.clock.add(const Duration(hours: 1));
await cycle();
expect(h.initBodies.length, initCount,
reason: '无自动重试(单次 init 即终判)');
});
test('401 会话过期 → waitingNetwork 短退避;重登录后(重试)完成,'
'attempt 不增长', () async {
final file = photoFile('g.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.initUnauthorized = 1;
final seeded = await h.seedTask(file, clientTaskId: 'ct-g');
await cycle();
var t = await task(seeded.id);
expect(t.state, UploadTaskState.waitingNetwork);
expect(t.lastError, ErrorCodes.authTokenExpired);
expect(t.attempt, 0);
expect(t.nextRetryAt, h.clock.add(const Duration(seconds: 30)));
h.clock = h.clock.add(const Duration(seconds: 31));
await cycle();
t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
});
});
group('凭证续期(presign 403)', () {
test('403 → renewUpload → 新 URL 重传成功', () async {
final file = photoFile('h.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.put403Count = 1;
final seeded = await h.seedTask(file, clientTaskId: 'ct-h');
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(h.renewCalls, 1);
expect(h.events,
['init:ct-h', 'put', 'renew:upl1', 'put', 'complete:upl1']);
// 第二次 PUT 用续期 URL。
expect(h.objectAdapter.requests.last.uri.query, 'renewed=1');
});
test('403 持续超元数上限 → 分类失败(retryableFailed)', () async {
final file = photoFile('h2.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
h.put403Count = 99; // 永远 403。
final seeded = await h.seedTask(file, clientTaskId: 'ct-h2');
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(h.renewCalls, kMaxCredentialRenewals);
expect(t.attempt, 1, reason: '第 4 次 403 落入通用分类桶');
});
});
group('取消', () {
test('未开始(queued):删行 + 删受管文件', () async {
final file = photoFile('i.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-i');
final cancelled = await engine.cancel('alice', seeded.id);
expect(cancelled, isTrue);
expect(await h.store.get('alice', seeded.id), isNull, reason: '行已删');
expect(file.existsSync(), isFalse, reason: '受管文件已删');
expect(h.events, isEmpty, reason: '从未发起传输');
});
test('进行中(uploading):尽力取消——置 cancelled、文件保留、PUT 收敛后不再迁移',
() async {
final file = photoFile('j.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final gate = Completer<void>();
h.putGate = gate;
final seeded = await h.seedTask(file, clientTaskId: 'ct-j');
// 只启动(PUT 挂起),不 cycle 收敛。泵事件循环直到 PUT 真正发出。
await engine.drain();
await until(() => h.objectAdapter.requests.isNotEmpty);
expect(h.objectAdapter.requests, isNotEmpty, reason: 'PUT 已发出');
final cancelled = await engine.cancel('alice', seeded.id);
expect(cancelled, isTrue);
expect((await task(seeded.id)).state, UploadTaskState.cancelled);
gate.complete();
await engine.settle();
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.cancelled, reason: '取消是终态,PUT 结果不复活');
expect(file.existsSync(), isTrue, reason: '进行中取消:文件保留给清理执行器');
expect(h.events.where((e) => e.startsWith('complete')), isEmpty);
});
test('已终态(completed):取消无效返回 false', () async {
final file = photoFile('i2.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-i2');
await cycle();
expect(await engine.cancel('alice', seeded.id), isFalse);
});
});
group('账号隔离与 A15', () {
test('引擎只出队当前账号任务(他人队列不动)', () async {
final mine = photoFile('k1.jpg');
final theirs = photoFile('k2.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final aliceTask = await h.seedTask(mine, clientTaskId: 'ct-k1');
final bobTask = await h.seedTask(
theirs,
accountId: 'bob',
clientTaskId: 'ct-k2',
);
await cycle();
expect((await task(aliceTask.id)).state, UploadTaskState.completed);
final bobRow = await h.store.get('bob', bobTask.id);
expect(bobRow!.state, UploadTaskState.queued, reason: 'bob 的任务不被触碰');
expect(h.initBodies.map((b) => b['clientTaskId']), ['ct-k1']);
});
test('A15:连拍中途切换相册/分类不影响既有任务(引擎只读行内目标)',
() async {
final f1 = photoFile('l1.jpg');
final f2 = photoFile('l2.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
// 第一张:相册 A 分类 X;随后"切到"相册 B 分类 Y 拍第二张。
final t1 = await h.seedTask(
f1,
albumId: 'alb-A',
categoryId: 'cat-X',
clientTaskId: 'ct-l1',
);
final t2 = await h.seedTask(
f2,
albumId: 'alb-B',
categoryId: 'cat-Y',
clientTaskId: 'ct-l2',
);
await cycle();
expect((await task(t1.id)).state, UploadTaskState.completed);
expect((await task(t2.id)).state, UploadTaskState.completed);
// 各自 init 携带各自入队时的目标(归属不变)。
expect(h.initBodies.firstWhere((b) => b['clientTaskId'] == 'ct-l1')
['albumId'], 'alb-A');
expect(h.initBodies.firstWhere((b) => b['clientTaskId'] == 'ct-l1')
['categoryId'], 'cat-X');
expect(h.initBodies.firstWhere((b) => b['clientTaskId'] == 'ct-l2')
['albumId'], 'alb-B');
expect(h.initBodies.firstWhere((b) => b['clientTaskId'] == 'ct-l2')
['categoryId'], 'cat-Y');
});
});
group('并行上限', () {
test('maxParallel=2:三任务闸门下同时只有 2 个 PUT,放开后全部完成',
() async {
final files = [
photoFile('m1.jpg'),
photoFile('m2.jpg'),
photoFile('m3.jpg'),
];
engine = h.buildEngine(maxParallel: 2);
engine.start('alice', period: const Duration(hours: 1));
final gate = Completer<void>();
h.putGate = gate;
for (var i = 0; i < files.length; i++) {
await h.seedTask(files[i], clientTaskId: 'ct-m${i + 1}');
}
await engine.drain();
// 泵到两个 PUT 都被适配器记录(协程 init 异步在途;第三个不发出)。
await until(() => h.objectAdapter.requests.length == 2);
expect(h.objectAdapter.requests, hasLength(2),
reason: '并行上限 2:第三个 PUT 不发出');
expect((await h.store.listByAccount('alice'))
.where((t) => t.state == UploadTaskState.uploading)
.length, 2);
gate.complete();
await cycle();
final states = await h.store.listByAccount('alice');
expect(states.map((t) => t.state),
everyElement(UploadTaskState.completed));
expect(h.objectAdapter.requests, hasLength(3));
});
});
group('崩溃恢复', () {
test('卡在 uploading 的任务(进程被杀痕迹)drain 前归位并完成', () async {
final file = photoFile('n.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
// 摆出"被杀时"的僵死状态:queued → uploading 后无 in-flight。
final seeded = await h.seedTask(file, clientTaskId: 'ct-n');
await h.store.updateState(seeded.id, UploadTaskState.uploading);
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(h.events, ['init:ct-n', 'put', 'complete:upl1']);
});
test('卡在 uploadedConfirming 的任务同样归位重试(complete 幂等)',
() async {
final file = photoFile('n2.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-n2');
await h.store.updateState(seeded.id, UploadTaskState.uploading);
await h.store.updateState(
seeded.id,
UploadTaskState.uploadedConfirming,
);
await cycle();
expect((await task(seeded.id)).state, UploadTaskState.completed);
});
});
group('清理执行器', () {
test('超过保留期的 completed:文件删除 + cleanedAt 标记 + 行保留为记录',
() async {
final file = photoFile('o.jpg');
engine = h.buildEngine(keepCompletedDays: 14);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-o');
await cycle();
expect((await task(seeded.id)).state, UploadTaskState.completed);
// 保留期内不清理。
expect(await engine.cleanup(), 0);
expect(file.existsSync(), isTrue);
// 20 天后(回写 updatedAt 早于截止)。
final future = h.clock.add(const Duration(days: 20));
expect(await engine.cleanup(now: future), 1);
expect(file.existsSync(), isFalse);
final t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(t.cleanedAt, future);
// 幂等:已清理不再重复。
expect(
await engine.cleanup(now: future.add(const Duration(days: 1))),
0,
);
});
test('未完成任务源文件永不清理(硬守护在引擎侧同样成立)', () async {
final queuedFile = photoFile('p1.jpg');
final waitingFile = photoFile('p2.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final ancient = h.clock.subtract(const Duration(days: 100));
// queued 任务(seedTask 落库即 queued,updatedAt=ancient)。
await h.seedTask(
queuedFile,
clientTaskId: 'ct-p1',
createdAt: ancient,
);
// waitingNetwork 的古老任务(updatedAt 同样古老)。
final w = await h.seedTask(
waitingFile,
clientTaskId: 'ct-p2',
createdAt: ancient,
);
await h.store.updateState(
w.id,
UploadTaskState.waitingNetwork,
patch: UploadTaskPatch(
nextRetryAt: h.clock.add(const Duration(hours: 1)),
),
now: ancient,
);
// q 的 updatedAt 已是 ancient(seedTask 落库时写入)。
expect(await engine.cleanup(), 0);
expect(queuedFile.existsSync(), isTrue);
expect(waitingFile.existsSync(), isTrue);
});
});
test('源文件缺失:分类为不自动重试的 retryableFailed', () async {
final file = photoFile('q.jpg');
engine = h.buildEngine();
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-q');
await file.delete(); // 受管副本被系统清理。
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(t.lastError, UploadEngine.sourceMissingCode);
expect(t.nextRetryAt, isNull);
// 文件检查在 PUT 阶段(init 已发生、无 PUT/complete)。
expect(h.events, ['init:ct-q']);
});
group('传输委托(Task 7 Transport 接口)', () {
late ScriptedTransport transport;
setUp(() {
transport = ScriptedTransport();
});
test('引擎 PUT 委托 Transport:凭证透传 + init/complete 仍在 Dart 引擎',
() async {
final file = photoFile('t1.jpg');
final progress = <String, List<int>>{};
engine = h.buildEngine(
transport: transport,
onProgress: (id, sent, total) =>
progress.putIfAbsent(id, () => <int>[]).add(sent),
);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-t1');
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(t.serverUploadId, 'upl1');
// 原生路径绕过 dio:对象适配器无请求,API 序列只含 init/complete。
expect(h.objectAdapter.requests, isEmpty);
expect(h.events, ['init:ct-t1', 'complete:upl1']);
// 委托参数:init 返回的凭证原样透传(URL/方法/头/文件路径/元数据)。
final call = transport.calls.single;
expect(call.taskId, seeded.id);
expect(call.url, 'http://object.test.local/staging/upl1/pho1.jpg');
expect(call.method, 'PUT');
expect(call.headers, {'Content-Type': 'image/jpeg'});
expect(call.filePath, file.path);
expect(call.metadata?['uploadId'], 'upl1');
expect(call.metadata?['attempt'], 0);
// 进度回调由 Transport 上报(队列 UI 挂接点不变)。
expect(progress[seeded.id], [1024]);
});
test('原生 needsRenew:引擎 renew 后携带新 URL 重新委托', () async {
final file = photoFile('t2.jpg');
engine = h.buildEngine(transport: transport);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-t2');
transport.enqueue(seeded.id, [
const TransportRenewRequiredException(),
const TransportResult(),
]);
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(h.renewCalls, 1);
expect(transport.calls, hasLength(2));
expect(transport.calls.first.url, endsWith('upl1/pho1.jpg'));
expect(transport.calls.last.url, endsWith('upl1/pho1.jpg?renewed=1'));
expect(h.events, ['init:ct-t2', 'renew:upl1', 'complete:upl1']);
});
test('原生 network 失败 → waitingNetwork 退避;恢复后重新委托完成',
() async {
final file = photoFile('t3.jpg');
engine = h.buildEngine(transport: transport);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-t3');
transport.enqueue(seeded.id, [
const TransportException(TransportErrorCodes.network),
]);
await cycle();
var t = await task(seeded.id);
expect(t.state, UploadTaskState.waitingNetwork);
expect(t.lastError, ErrorCodes.networkError);
expect(t.attempt, 1);
expect(t.nextRetryAt, h.clock.add(const Duration(seconds: 5)));
// 到期重试:无手本 → 默认成功。
h.clock = h.clock.add(const Duration(seconds: 10));
await cycle();
t = await task(seeded.id);
expect(t.state, UploadTaskState.completed);
expect(transport.calls, hasLength(2));
});
test('原生 HTTP_4XX(400) → retryableFailed(自动重试退避)', () async {
final file = photoFile('t4.jpg');
engine = h.buildEngine(transport: transport);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-t4');
transport.enqueue(seeded.id, [
const TransportException(TransportErrorCodes.http, statusCode: 400),
]);
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(t.lastError, ErrorCodes.httpError);
expect(t.attempt, 1);
expect(t.nextRetryAt, isNotNull, reason: 'attempt 1 < 3:仍自动重试');
});
test('原生 HTTP 401 → waitingNetwork 短退避(会话语义)', () async {
final file = photoFile('t5.jpg');
engine = h.buildEngine(transport: transport);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-t5');
transport.enqueue(seeded.id, [
const TransportException(TransportErrorCodes.http, statusCode: 401),
]);
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.waitingNetwork);
expect(t.lastError, ErrorCodes.authTokenExpired);
expect(t.attempt, 0, reason: '会话过期不烧次数');
expect(t.nextRetryAt, h.clock.add(const Duration(seconds: 30)));
});
test('原生 SOURCE_MISSING → 不自动重试的 retryableFailed', () async {
final file = photoFile('t6.jpg');
engine = h.buildEngine(transport: transport);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-t6');
transport.enqueue(seeded.id, [
const TransportException(TransportErrorCodes.sourceMissing),
]);
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(t.lastError, UploadEngine.sourceMissingCode);
expect(t.nextRetryAt, isNull);
});
test('原生 SERVER_ERROR(500) → retryableFailed(5xx 语义桶)', () async {
final file = photoFile('t7.jpg');
engine = h.buildEngine(transport: transport);
engine.start('alice', period: const Duration(hours: 1));
final seeded = await h.seedTask(file, clientTaskId: 'ct-t7');
transport.enqueue(seeded.id, [
const TransportException(TransportErrorCodes.http, statusCode: 500),
]);
await cycle();
final t = await task(seeded.id);
expect(t.state, UploadTaskState.retryableFailed);
expect(t.lastError, ErrorCodes.internalError);
expect(t.attempt, 1);
});
});
}