464 lines
20 KiB
Dart
464 lines
20 KiB
Dart
import 'dart:convert';
|
|
import 'dart:typed_data';
|
|
import 'package:archive/archive.dart';
|
|
import 'package:crypto/crypto.dart';
|
|
import 'package:drift/drift.dart' as drift;
|
|
import 'package:drift/native.dart';
|
|
import 'package:flutter_test/flutter_test.dart';
|
|
import 'package:http/http.dart' as http;
|
|
import 'package:http/testing.dart';
|
|
import 'package:shared_preferences/shared_preferences.dart';
|
|
|
|
import 'package:motorcycle_dash/database/database.dart';
|
|
import 'package:motorcycle_dash/services/auth/auth_service.dart';
|
|
import 'package:motorcycle_dash/services/auth/auth_user.dart';
|
|
import 'package:motorcycle_dash/services/sync/r2_upload_service.dart';
|
|
import 'package:motorcycle_dash/services/sync/sync_bundle_worker.dart';
|
|
import 'package:motorcycle_dash/services/sync/sync_manager.dart';
|
|
import 'package:motorcycle_dash/services/sync/sync_models.dart';
|
|
import 'package:motorcycle_dash/services/sync/sync_outbox.dart';
|
|
import 'package:motorcycle_dash/services/sync/sync_provider.dart';
|
|
|
|
/// Architectural Boundaries & Verification Level for Offline Sync Fixture:
|
|
///
|
|
/// 1. REAL COMPONENTS EXERCISED:
|
|
/// - Drift SQLite Schema 7 [AppDatabase] with in-memory SQLite engine:
|
|
/// Executes real relational schema operations, queries, and immutable identity
|
|
/// contracts (`ensureCloudRideId()`, tables `rides`, `telemetry_samples`,
|
|
/// `field_test_events`, `field_test_summaries`).
|
|
/// - [SyncBundleWorker]: Generates authentic ZIP archives with DEFLATE/gzip compression,
|
|
/// serializing 20 Hz telemetry CSV, forensic events CSV, and session JSON.
|
|
/// - [SyncManager]: Full queue state machine, retry backoff calculation, app lifecycle
|
|
/// transitions, and multi-provider orchestration.
|
|
/// - [SyncOutbox]: Durable queue manifest serialization and deserialization via
|
|
/// [SharedPreferences] storage.
|
|
/// - [R2UploadService]: Real HTTP PUT request formatting, custom headers, SHA-256
|
|
/// checksum generation, and response parsing.
|
|
///
|
|
/// 2. REPLACED / MOCKED COMPONENTS:
|
|
/// - [_IsolatedContractSyncProvider]: Replaces production `FirebaseSyncProvider` to isolate
|
|
/// test execution from live Google Play Services / Firebase C++ native bindings, while
|
|
/// strictly adhering to the identical `SyncProvider` contract.
|
|
/// - [MockClient]: Replaces the live Cloudflare R2 Worker HTTP endpoint, intercepting
|
|
/// requests and verifying byte-level payload checksums and headers.
|
|
/// - Simulated `networkAvailable` flag: Software-level mock on the test provider.
|
|
/// Note: Distinguished from true OS-level network interface routing isolation
|
|
/// (e.g., netfilter/iptables rules or taking down eth0/wlan0 interfaces).
|
|
///
|
|
/// 3. UPLOAD SEQUENCING & CONSISTENCY CONTRACT:
|
|
/// - Sequential Execution: In the authoritative MotorCloud pipeline, the R2 binary bundle
|
|
/// upload executes first. Cloud Firestore write only executes after R2 returns HTTP 200/success.
|
|
/// This prevents orphan documents in Firestore if the R2 upload fails.
|
|
/// - NOT A DISTRIBUTED 2-PHASE COMMIT (2PC): Sequential execution ensures Firestore is
|
|
/// only updated upon R2 success, but is NOT a distributed two-phase commit across R2
|
|
/// and Firestore. If Firestore write fails after R2 succeeds, R2 holds the bundle while
|
|
/// Firestore remains unpopulated until the next retry cycle reconciles the state.
|
|
class _FakeAuthService implements AuthService {
|
|
AuthUser? _user;
|
|
|
|
@override
|
|
AuthUser? get currentUser => _user;
|
|
|
|
void setUser(AuthUser? user) {
|
|
_user = user;
|
|
}
|
|
|
|
@override
|
|
Future<String?> getIdToken([bool forceRefresh = false]) async {
|
|
return _user != null ? 'fake_jwt_token_${_user!.uid}' : null;
|
|
}
|
|
|
|
@override
|
|
dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
|
|
}
|
|
|
|
/// Test implementation of [SyncProvider] honoring the MotorCloud upload contract.
|
|
///
|
|
/// Replaces production `FirebaseSyncProvider` to run in headless CI/test environments
|
|
/// without initializing Firebase native plugins. Replaces live R2 worker network endpoints
|
|
/// with [MockClient].
|
|
class _IsolatedContractSyncProvider implements SyncProvider {
|
|
final AppDatabase _db;
|
|
final AuthService _authService;
|
|
final R2UploadService _r2UploadService;
|
|
final SyncBundleWorker _bundleWorker;
|
|
final List<Map<String, dynamic>> recordedUploads = [];
|
|
bool networkAvailable = true;
|
|
|
|
_IsolatedContractSyncProvider({
|
|
required this._db,
|
|
required this._authService,
|
|
required this._r2UploadService,
|
|
required this._bundleWorker,
|
|
});
|
|
|
|
@override
|
|
String get providerId => 'firebase';
|
|
|
|
@override
|
|
Future<bool> isAvailable() async => true;
|
|
|
|
@override
|
|
Future<bool> requiresAuthentication() async {
|
|
final user = _authService.currentUser;
|
|
return user == null || user.isAnonymous;
|
|
}
|
|
|
|
@override
|
|
Future<bool> isRideConfirmedInCloud(String remoteRideId) async => false;
|
|
|
|
@override
|
|
Future<CloudConfirmationResult> verifyCloudConfirmation(String remoteRideId) async =>
|
|
CloudConfirmationResult.missing(remoteRideId: remoteRideId);
|
|
|
|
@override
|
|
Future<SyncResult> uploadRide(SyncItem item) async {
|
|
if (!networkAvailable) {
|
|
return SyncResult.temporaryFailure('Network isolated (all interfaces down). Cannot reach R2 gateway.');
|
|
}
|
|
|
|
final user = _authService.currentUser;
|
|
if (user == null || user.isAnonymous) {
|
|
return SyncResult.authRequired(message: 'Sign in to MotorCloud is required for cloud backup.');
|
|
}
|
|
|
|
final remoteRideId = await _db.ensureCloudRideId(item.localRideId);
|
|
final idToken = await _authService.getIdToken();
|
|
if (idToken == null || idToken.isEmpty) {
|
|
return SyncResult.authRequired(message: 'Valid Firebase authentication token required for cloud sync.');
|
|
}
|
|
|
|
final zipBytes = await _bundleWorker.generateCompressedBundle(item.localRideId);
|
|
if (zipBytes == null || zipBytes.isEmpty) {
|
|
return SyncResult.permanentFailure('Failed generating compressed telemetry bundle for ride #${item.localRideId}.');
|
|
}
|
|
|
|
final sha256Hash = sha256.convert(zipBytes).toString();
|
|
final r2Result = await _r2UploadService.uploadTelemetryBundle(
|
|
idToken: idToken,
|
|
remoteRideId: remoteRideId,
|
|
zipBytes: zipBytes,
|
|
sha256Hash: sha256Hash,
|
|
appVersion: '1.0.2',
|
|
);
|
|
|
|
if (!r2Result.success) {
|
|
if (r2Result.requiresAuth) return SyncResult.authRequired(message: r2Result.errorMessage ?? 'R2 auth required');
|
|
if (r2Result.isRetryable) return SyncResult.temporaryFailure(r2Result.errorMessage ?? 'Temporary R2 upload error');
|
|
return SyncResult.permanentFailure(r2Result.errorMessage ?? 'Permanent R2 upload rejection');
|
|
}
|
|
|
|
// UPLOAD SEQUENCING CONTRACT:
|
|
// Sequential execution: In the authoritative MotorCloud pipeline, R2 bundle upload
|
|
// precedes Firestore document creation. This ordering ensures Firestore is only updated
|
|
// upon verified R2 success, avoiding orphan cloud records.
|
|
//
|
|
// Distributed Transaction Boundary: Sequential execution enforces error-gated ordering,
|
|
// but is NOT a distributed two-phase commit (2PC) across Cloudflare R2 and Firestore.
|
|
recordedUploads.add({
|
|
'localRideId': item.localRideId,
|
|
'remoteId': remoteRideId,
|
|
'objectKey': r2Result.objectKey,
|
|
'sha256': sha256Hash,
|
|
'bytesWritten': r2Result.bytesWritten,
|
|
'timestamp': DateTime.now(),
|
|
});
|
|
|
|
return SyncResult.success(remoteId: remoteRideId);
|
|
}
|
|
}
|
|
|
|
void main() {
|
|
TestWidgetsFlutterBinding.ensureInitialized();
|
|
|
|
late AppDatabase db;
|
|
late _FakeAuthService authService;
|
|
late SharedPreferences prefs;
|
|
|
|
setUp(() async {
|
|
db = AppDatabase.forTesting(NativeDatabase.memory());
|
|
authService = _FakeAuthService();
|
|
SharedPreferences.setMockInitialValues({});
|
|
prefs = await SharedPreferences.getInstance();
|
|
});
|
|
|
|
tearDown(() async {
|
|
await db.close();
|
|
});
|
|
|
|
test('Isolated Offline Sync Contract: Network Isolation, R2 Upload, Reconciliation & Idempotency', () async {
|
|
// 1. Stage an offline recorded ride with 2,788 telemetry samples and 8 events
|
|
final startTime = DateTime.utc(2026, 10, 3, 14, 0, 0);
|
|
final endTime = startTime.add(const Duration(minutes: 40));
|
|
|
|
final localRideId = await db.into(db.rides).insert(
|
|
RidesCompanion.insert(
|
|
startTime: startTime,
|
|
endTime: drift.Value(endTime),
|
|
distanceMiles: const drift.Value(42.5),
|
|
topSpeedMph: const drift.Value(85.0),
|
|
movingTime: const drift.Value(2400),
|
|
stoppedTime: const drift.Value(0),
|
|
averageSpeed: const drift.Value(63.75),
|
|
maxLeanLeft: const drift.Value(44.2),
|
|
maxLeanRight: const drift.Value(46.8),
|
|
maxAccelG: const drift.Value(0.78),
|
|
maxBrakingG: const drift.Value(-0.92),
|
|
elevationGain: const drift.Value(1450),
|
|
elevationLoss: const drift.Value(1450),
|
|
motorcycleMake: const drift.Value('Ducati'),
|
|
motorcycleModel: const drift.Value('Streetfighter V4'),
|
|
motorcycleYear: const drift.Value(2024),
|
|
),
|
|
);
|
|
|
|
// Insert 2,788 telemetry samples at 20 Hz
|
|
final samples = List.generate(2788, (i) {
|
|
final t = startTime.add(Duration(milliseconds: i * 50));
|
|
return TelemetrySamplesCompanion.insert(
|
|
rideId: localRideId,
|
|
timestamp: t,
|
|
latitude: drift.Value(37.3861 + (i * 0.0001)),
|
|
longitude: drift.Value(-122.2641 - (i * 0.0001)),
|
|
speed: drift.Value(45.0 + (i % 20)),
|
|
leanAngle: drift.Value(15.0 + (i % 30)),
|
|
accelX: const drift.Value(0.05),
|
|
accelY: const drift.Value(0.12),
|
|
accelZ: const drift.Value(0.98),
|
|
);
|
|
});
|
|
await db.batch((b) {
|
|
b.insertAll(db.telemetrySamples, samples);
|
|
});
|
|
|
|
// Insert 8 forensic events
|
|
final events = List.generate(8, (i) {
|
|
return FieldTestEventsCompanion.insert(
|
|
rideId: drift.Value(localRideId),
|
|
timestamp: startTime.add(Duration(minutes: (i + 1) * 4)),
|
|
eventType: i.isEven ? 'Hard Braking' : 'Extreme Lean Corner',
|
|
severity: 'MEDIUM',
|
|
message: 'Forensic event #$i detected at 20Hz',
|
|
);
|
|
});
|
|
await db.batch((b) {
|
|
b.insertAll(db.fieldTestEvents, events);
|
|
});
|
|
|
|
// Insert 1 field test summary
|
|
await db.into(db.fieldTestSummaries).insert(
|
|
FieldTestSummariesCompanion.insert(
|
|
rideId: localRideId,
|
|
startTime: startTime,
|
|
endTime: drift.Value(endTime),
|
|
durationSeconds: const drift.Value(2400),
|
|
rideQualityScore: const drift.Value(96),
|
|
rideQualityRating: const drift.Value('EXCELLENT'),
|
|
),
|
|
);
|
|
|
|
// 2. Claim ride for verified test rider
|
|
const testUid = 'pilot_qa_verified_123';
|
|
authService.setUser(const AuthUser(
|
|
uid: testUid,
|
|
email: 'pilot_qa@motorsignal.test',
|
|
isAnonymous: false,
|
|
));
|
|
await (db.update(db.rides)..where((r) => r.id.equals(localRideId))).write(
|
|
const RidesCompanion(cloudOwnerUid: drift.Value(testUid)),
|
|
);
|
|
|
|
final claimedRide = await (db.select(db.rides)..where((r) => r.id.equals(localRideId))).getSingle();
|
|
expect(claimedRide.cloudOwnerUid, testUid);
|
|
|
|
// 3. Set up R2 Worker mock implementing the authoritative upload contract
|
|
final uploadedR2Bundles = <String, Uint8List>{};
|
|
final r2Calls = <Map<String, dynamic>>[];
|
|
|
|
final r2MockClient = MockClient((request) async {
|
|
r2Calls.add({
|
|
'method': request.method,
|
|
'path': request.url.path,
|
|
'headers': request.headers,
|
|
'bodyLength': request.bodyBytes.length,
|
|
});
|
|
|
|
if (request.method == 'PUT' && request.url.path == '/r2/upload') {
|
|
final remoteRideId = request.headers['X-Remote-Ride-Id'];
|
|
final clientSha256 = request.headers['X-Bundle-Sha256'];
|
|
final authHeader = request.headers['Authorization'];
|
|
|
|
if (authHeader == null || !authHeader.startsWith('Bearer ')) {
|
|
return http.Response(jsonEncode({'error': 'Unauthorized'}), 401);
|
|
}
|
|
|
|
// Verify SHA-256 against body
|
|
final computedSha = sha256.convert(request.bodyBytes).toString();
|
|
if (clientSha256 != null && clientSha256 != computedSha) {
|
|
return http.Response(jsonEncode({'error': 'SHA mismatch'}), 422);
|
|
}
|
|
|
|
final objectKey = 'users/$testUid/rides/$remoteRideId/telemetry.zip';
|
|
uploadedR2Bundles[objectKey] = Uint8List.fromList(request.bodyBytes);
|
|
|
|
return http.Response(
|
|
jsonEncode({
|
|
'success': true,
|
|
'objectKey': objectKey,
|
|
'size': request.bodyBytes.length,
|
|
'etag': 'test-etag-verified',
|
|
'sha256': computedSha,
|
|
'uploadedAt': DateTime.now().toUtc().toIso8601String(),
|
|
}),
|
|
200,
|
|
headers: {'content-type': 'application/json'},
|
|
);
|
|
}
|
|
return http.Response('Not found', 404);
|
|
});
|
|
|
|
final r2Service = R2UploadService(
|
|
client: r2MockClient,
|
|
workerBaseUrl: 'https://r2-fixture.local',
|
|
);
|
|
|
|
final bundleWorker = SyncBundleWorker(db);
|
|
final contractProvider = _IsolatedContractSyncProvider(
|
|
db: db,
|
|
authService: authService,
|
|
r2UploadService: r2Service,
|
|
bundleWorker: bundleWorker,
|
|
);
|
|
|
|
final outbox = SyncOutbox(prefs);
|
|
final syncManager = SyncManager(outbox, initialProviders: [contractProvider]);
|
|
|
|
// 4. Enqueue ride for sync (without autoProcess so we control step-by-step)
|
|
await syncManager.enqueueRide(localRideId, autoProcess: false);
|
|
expect(syncManager.pendingItems.isNotEmpty, isTrue);
|
|
expect(syncManager.currentState.pendingCount, 1);
|
|
|
|
// 5. PHASE A: Network Isolated State (Simulating off-grid ride / offline interface)
|
|
contractProvider.networkAvailable = false;
|
|
await syncManager.processQueue();
|
|
|
|
// Verification: Outbox queue did NOT drain, ride remains pending with backoff
|
|
expect(syncManager.currentState.pendingCount, 1);
|
|
expect(uploadedR2Bundles, isEmpty);
|
|
expect(contractProvider.recordedUploads, isEmpty);
|
|
|
|
// 6. PHASE B: Network Connectivity Restored
|
|
contractProvider.networkAvailable = true;
|
|
// Advance simulated time past the 5-second initial retry backoff
|
|
syncManager.setNowProvider(() => DateTime.now().add(const Duration(seconds: 10)));
|
|
await syncManager.processQueue();
|
|
|
|
// Verification: Queue drains completely to 0 pending and 1 synced
|
|
expect(syncManager.currentState.pendingCount, 0);
|
|
expect(syncManager.currentState.syncedCount, 1);
|
|
expect(syncManager.pendingItems, isEmpty);
|
|
|
|
// Verification: R2 contains the authoritative telemetry bundle
|
|
expect(r2Calls, hasLength(1));
|
|
final expectedRemoteId = await db.ensureCloudRideId(localRideId);
|
|
final expectedObjectKey = 'users/$testUid/rides/$expectedRemoteId/telemetry.zip';
|
|
expect(uploadedR2Bundles.containsKey(expectedObjectKey), isTrue);
|
|
|
|
// Verification: Uploaded bundle size and checksum reconcile
|
|
final uploadedBytes = uploadedR2Bundles[expectedObjectKey]!;
|
|
expect(uploadedBytes.length, greaterThan(100)); // Real zip archive with dense telemetry
|
|
final uploadedSha = sha256.convert(uploadedBytes).toString();
|
|
expect(r2Calls.first['headers']['X-Bundle-Sha256'], uploadedSha);
|
|
|
|
// Unzip and verify telemetry.csv and forensic events inside the uploaded bundle
|
|
final archive = ZipDecoder().decodeBytes(uploadedBytes);
|
|
final fileNames = archive.files.map((f) => f.name).toSet();
|
|
expect(fileNames, contains('README.txt'));
|
|
expect(fileNames, contains('ride_summary.json'));
|
|
expect(fileNames, contains('field_test_session.json'));
|
|
expect(fileNames, contains('field_test_events.csv'));
|
|
expect(fileNames, contains('telemetry.csv'));
|
|
|
|
// Verification: Provider recorded single successful sync
|
|
expect(contractProvider.recordedUploads, hasLength(1));
|
|
expect(contractProvider.recordedUploads.first['remoteId'], expectedRemoteId);
|
|
|
|
// 7. PHASE C: Genuine Service Reconstruction & Restart Idempotency (Zero duplicate uploads)
|
|
// First, dispose the original in-memory SyncManager to simulate application termination / cold shutdown.
|
|
syncManager.dispose();
|
|
|
|
// Reconstruct brand-new SyncOutbox and SyncManager instances from the persisted SharedPreferences
|
|
// state and Drift SQLite AppDatabase.
|
|
// This genuinely tests that persistent storage (not in-memory cache) faithfully retains the
|
|
// authoritative queue state across an application restart.
|
|
final reconstructedOutbox = SyncOutbox(prefs);
|
|
expect(reconstructedOutbox.isLoaded, isTrue);
|
|
expect(reconstructedOutbox.count, 1,
|
|
reason: 'Reconstructed outbox must restore exactly 1 item from persistent SharedPreferences storage.');
|
|
|
|
final persistedItem = reconstructedOutbox.getItem(localRideId, 'firebase');
|
|
expect(persistedItem, isNotNull,
|
|
reason: 'Persisted item for ride #$localRideId must exist in reconstructed outbox.');
|
|
expect(persistedItem!.status, SyncStatus.synced,
|
|
reason: 'Persisted item status must be preserved as SyncStatus.synced.');
|
|
expect(persistedItem.remoteId, expectedRemoteId,
|
|
reason: 'Persisted remoteId must match authoritative cloudRideId.');
|
|
expect(persistedItem.uploadedAt, isNotNull,
|
|
reason: 'Persisted uploadedAt timestamp must be preserved.');
|
|
expect(reconstructedOutbox.getPendingItems(), isEmpty,
|
|
reason: 'Reconstructed outbox must have 0 pending items.');
|
|
|
|
final reconstructedContractProvider = _IsolatedContractSyncProvider(
|
|
db: db,
|
|
authService: authService,
|
|
r2UploadService: r2Service,
|
|
bundleWorker: bundleWorker,
|
|
);
|
|
|
|
final reconstructedSyncManager = SyncManager(
|
|
reconstructedOutbox,
|
|
initialProviders: [reconstructedContractProvider],
|
|
db: db,
|
|
authService: authService,
|
|
);
|
|
|
|
// Verify queue state on the reconstructed service immediately after initialization
|
|
expect(reconstructedSyncManager.currentState.pendingCount, 0);
|
|
expect(reconstructedSyncManager.currentState.syncedCount, 1);
|
|
expect(reconstructedSyncManager.currentState.failedCount, 0);
|
|
expect(reconstructedSyncManager.pendingItems, isEmpty);
|
|
|
|
// Trigger processQueue() on the reconstructed service.
|
|
// The service must recognize that the item is already synced and bypass upload.
|
|
await reconstructedSyncManager.processQueue();
|
|
expect(reconstructedContractProvider.recordedUploads, isEmpty,
|
|
reason: 'Reconstructed service must NOT perform any uploads for already-synced items.');
|
|
expect(r2Calls, hasLength(1),
|
|
reason: 'Total R2 HTTP PUT calls must remain strictly 1 across entire fixture run.');
|
|
|
|
// Attempt re-enqueuing the same ride into the reconstructed manager.
|
|
// SyncOutbox.enqueue() must safely detect the existing 'synced' record and return it without re-queueing.
|
|
await reconstructedSyncManager.enqueueRide(localRideId, autoProcess: false);
|
|
expect(reconstructedOutbox.getItem(localRideId, 'firebase')!.status, SyncStatus.synced);
|
|
expect(reconstructedSyncManager.currentState.pendingCount, 0);
|
|
expect(reconstructedSyncManager.currentState.syncedCount, 1);
|
|
|
|
// Trigger processQueue() again to verify post-enqueue idempotency
|
|
await reconstructedSyncManager.processQueue();
|
|
expect(reconstructedContractProvider.recordedUploads, isEmpty,
|
|
reason: 'No uploads may occur after re-enqueuing an already synced ride on reconstructed service.');
|
|
expect(r2Calls, hasLength(1),
|
|
reason: 'Total R2 HTTP PUT calls must remain strictly 1 (zero duplicate uploads).');
|
|
|
|
// The remote ID in SQLite AppDatabase must remain completely stable and immutable across re-queries
|
|
final secondRemoteId = await db.ensureCloudRideId(localRideId);
|
|
expect(secondRemoteId, expectedRemoteId,
|
|
reason: 'cloudRideId contract requires immutable identity across restarts and re-queries.');
|
|
|
|
expect(reconstructedSyncManager.currentState.pendingCount, 0);
|
|
expect(reconstructedSyncManager.currentState.syncedCount, 1);
|
|
|
|
reconstructedSyncManager.dispose();
|
|
});
|
|
}
|