app-flutter/test/boundary_test.dart

import 'dart:async';
import 'dart:convert';
import 'dart:io';

import 'package:display/boundary.dart';
import 'package:flutter_test/flutter_test.dart';

/// A minimal stand-in for svc-gw: hello, then whatever the test sends.
class FakeGateway {
  late HttpServer server;
  final sockets = <WebSocket>[];
  final received = <Map<String, dynamic>>[];
  bool ackCommands = true;

  Future<void> start({bool mock = false}) async {
    server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
    server.transform(WebSocketTransformer()).listen((ws) {
      sockets.add(ws);
      ws.add(jsonEncode({
        't': 'hello', 'proto': 0, 'server': 'fake', 'host': 'test', 'source': mock ? 'mock' : 'none',
        'mock': mock, 'hb_ms': 100, 'timeout_ms': 400,
      }));
      ws.listen((raw) {
        final m = jsonDecode(raw as String) as Map<String, dynamic>;
        received.add(m);
        if (m['t'] == 'cmd' && ackCommands) {
          ws.add(jsonEncode({'t': 'ack', 'id': m['id'], 'ok': true, 'detail': 'ok'}));
        }
      });
    });
  }

  String get url => 'ws://127.0.0.1:${server.port}/';
}

Future<void> until(bool Function() cond, {Duration timeout = const Duration(seconds: 5)}) async {
  final end = DateTime.now().add(timeout);
  while (!cond()) {
    if (DateTime.now().isAfter(end)) throw TimeoutException('condition not met');
    await Future<void>.delayed(const Duration(milliseconds: 20));
  }
}

void main() {
  test('connects, takes state, flags mock, heartbeats', () async {
    final gw = FakeGateway();
    await gw.start(mock: true);
    final b = Boundary(gw.url)..start();
    await until(() => b.state == LinkState.up);
    expect(b.mock, isTrue);
    gw.sockets.first.add(jsonEncode({
      't': 'state',
      'seq': 1,
      'signals': {
        'engine_rpm': {'timestamp': 0, 'monotonic': 0, 'value': 820, 'unit': 'rpm', 'mock': true},
      },
    }));
    await until(() => b.signals.containsKey('engine_rpm'));
    await until(() => gw.received.where((m) => m['t'] == 'hb').length >= 2);
    await b.stop();
    await gw.server.close(force: true);
  });

  test('silence past timeout_ms drops the link and reconnects', () async {
    final gw = FakeGateway();
    await gw.start();
    final b = Boundary(gw.url)..start();
    await until(() => b.state == LinkState.up);
    // The fake never sends heartbeats, so the watchdog must fire after 400 ms.
    await until(() => gw.sockets.length >= 2, timeout: const Duration(seconds: 4));
    await b.stop();
    await gw.server.close(force: true);
  });

  test('commands are retried under the same id until acked', () async {
    final gw = FakeGateway()..ackCommands = false;
    await gw.start();
    final b = Boundary(gw.url)..start();
    await until(() => b.state == LinkState.up);
    final id = b.command('mark', {'text': 'brake'});
    await until(() => gw.received.any((m) => m['t'] == 'cmd'));
    // Drop the connection unacked; the command must be resent after reconnect.
    gw.ackCommands = true;
    await gw.sockets.first.close();
    await until(() => b.pending.isEmpty);
    final sent = gw.received.where((m) => m['t'] == 'cmd').toList();
    expect(sent.length, greaterThanOrEqualTo(2));
    expect(sent.every((m) => m['id'] == id), isTrue);
    expect(sent.last['args']['delay_ms'], greaterThan(0));
    expect(b.acks.first.ok, isTrue);
    await b.stop();
    await gw.server.close(force: true);
  });
}