using System;
|
using System.Diagnostics;
|
using System.IO;
|
using System.Threading;
|
using System.Threading.Tasks;
|
using Aging.Infrastructure.Broker;
|
using Aging.Infrastructure.Core;
|
using Aging.Infrastructure.Drivers;
|
using Aging.ResourceContracts;
|
using CanSimulatorServer;
|
using DeviceBroker;
|
using Xunit;
|
|
namespace Aging.MvpTests
|
{
|
/// <summary>
|
/// Broker 全链路 e2e：客户端（arb.db 租约 + CanDeviceDriver BrokerRpc 代理）↔
|
/// DeviceBroker 宿主（命名管道 + 栅栏再校验 + configHash 对账 + SessionGate 原子片段）。
|
/// 宿主的物理 CAN 驱动经 TCP 连进程内托管的模拟器服务（固定端口 19555，与 App.config 一致）。
|
/// 每个测试使用唯一 brokerId → 独立管道名，避免并行串扰。
|
/// </summary>
|
public class BrokerE2ETests : IDisposable
|
{
|
private const string DevId = "CAN-B1";
|
private const int SimPort = 19555;
|
|
private readonly string _dir;
|
private readonly string _dataDb;
|
private readonly string _arbDb;
|
private readonly string _brokerId;
|
private readonly string _pipeName;
|
private readonly CancellationTokenSource _cts = new CancellationTokenSource();
|
private readonly Task _serverTask;
|
private readonly DbLeaseStore _clientLease;
|
private readonly TcpSimulatorHost _sim;
|
|
public BrokerE2ETests()
|
{
|
_dir = Path.Combine(Path.GetTempPath(), "brk_" + Guid.NewGuid().ToString("N"));
|
Directory.CreateDirectory(_dir);
|
_dataDb = Path.Combine(_dir, "data.db");
|
_arbDb = Path.Combine(_dir, "arb.db");
|
_brokerId = "broker-" + Guid.NewGuid().ToString("N").Substring(0, 8);
|
_pipeName = BrokerPipeName.For(_brokerId);
|
|
CreateDataDb();
|
|
// 模拟器服务先于宿主就位（宿主首次 CONNECT 会拉起物理 TCP 会话）
|
_sim = new TcpSimulatorHost();
|
_sim.StartAsync(SimPort).GetAwaiter().GetResult();
|
|
var server = new BrokerServer(_dataDb, _arbDb, _brokerId);
|
server.LoadDevices();
|
_serverTask = Task.Run(() => server.RunAsync(_cts.Token));
|
_clientLease = new DbLeaseStore(_arbDb, "local-main");
|
|
// 等监听就位（客户端本身有 3s 连接超时，另有 BROKER_UNREACHABLE 重试兜底）
|
Thread.Sleep(250);
|
}
|
|
public void Dispose()
|
{
|
try { _cts.Cancel(); } catch { }
|
try { _serverTask.Wait(1500); } catch { }
|
try { _sim?.Dispose(); } catch { }
|
_cts.Dispose();
|
foreach (var f in Directory.GetFiles(_dir))
|
{
|
try { File.Delete(f); } catch { }
|
}
|
try { Directory.Delete(_dir); } catch { }
|
}
|
|
// ==================== 正向链路 ====================
|
|
[Fact]
|
public async Task WithLease_Connect_Send_Query_Succeeds()
|
{
|
var grant = GrabLease();
|
var hash = DeviceConfigHash.Compute(_dataDb);
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
|
|
await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
|
|
var tx = await driver.SendCommandAsync("SEND:0x123:11 22", CancellationToken.None);
|
Assert.Contains("OK:TX", tx);
|
|
var status = await driver.QueryAsync("STATUS", CancellationToken.None);
|
Assert.Contains("STATUS", status);
|
|
// Broker 侧执行成功后租约仍有效（未被误释放）
|
Assert.Equal(LeasePhase.Active, _clientLease.PhaseOf(DevId));
|
_clientLease.Release(grant, "test-end");
|
}
|
|
[Fact]
|
public async Task TwoSequencedFragments_SameSession_ExecuteSerially()
|
{
|
var grant = GrabLease();
|
var hash = DeviceConfigHash.Compute(_dataDb);
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
|
await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
|
|
// SessionGate 下同会话片段严格串行（同进程单客户端本就串行，验证不产生交错错误）
|
var a = await driver.SendCommandAsync("SEND:0x100:01", CancellationToken.None);
|
var b = await driver.SendCommandAsync("SEND:0x200:02", CancellationToken.None);
|
Assert.Contains("OK:TX", a);
|
Assert.Contains("OK:TX", b);
|
_clientLease.Release(grant, "test-end");
|
}
|
|
// ==================== 负向链路 ====================
|
|
[Fact]
|
public async Task NoLease_ConnectAllowed_Send_RejectedStaleLease()
|
{
|
// CONNECT/DISCONNECT 仅登记/释放宿主物理会话，豁免租约；
|
// SEND 有总线输出，无租约必须拒绝（INV-01 禁止无租约直接 I/O）
|
var hash = DeviceConfigHash.Compute(_dataDb);
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
|
|
await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
|
|
var ex = await Record.ExceptionAsync(
|
() => driver.SendCommandAsync("SEND:0x100:01", CancellationToken.None));
|
Assert.NotNull(ex);
|
Assert.Contains("LEASE_STALE", ex.Message);
|
}
|
|
[Fact]
|
public async Task ConfigHashMismatch_Rejected()
|
{
|
GrabLease(); // 有租约，但哈希对不上 → INV-04 拦截
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, "deadbeefdeadbeef");
|
|
var ex = await Record.ExceptionAsync(
|
() => RetryConnectAsync(driver, expectThrow: true));
|
Assert.NotNull(ex);
|
Assert.Contains("CONFIG_MISMATCH", ex.Message);
|
}
|
|
[Fact]
|
public async Task ReleasedLease_Send_Rejected()
|
{
|
var grant = GrabLease();
|
var hash = DeviceConfigHash.Compute(_dataDb);
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
|
await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
|
|
// 租约归还后再 I/O → 栅栏必须拒绝（禁止无租约直接 I/O）
|
_clientLease.Release(grant, "released-by-test");
|
var ex = await Record.ExceptionAsync(
|
() => driver.SendCommandAsync("SEND:0x100:01", CancellationToken.None));
|
Assert.NotNull(ex);
|
Assert.Contains("LEASE_STALE", ex.Message);
|
}
|
|
[Fact]
|
public async Task MaintenanceLease_BrokeredWrite_Succeeds()
|
{
|
// 调试页维护写权链路：维护租约写入 arb.db（OwnerType=Maintenance）后，
|
// Broker 栅栏同样放行（占用可见/拒绝语义已由生产场景用例覆盖）
|
var outcome = _clientLease.Acquire(new AcquireRequest
|
{
|
OwnerId = "maint-session-1",
|
OwnerType = OwnerTypeV3.Maintenance,
|
ResourceKeys = new[] { DevId },
|
TtlSeconds = 30
|
});
|
Assert.Equal(AcquireResultCode.Acquired, outcome.Code);
|
|
var hash = DeviceConfigHash.Compute(_dataDb);
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
|
await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
|
var tx = await driver.SendCommandAsync("SEND:0x300:AB", CancellationToken.None);
|
Assert.Contains("OK:TX", tx);
|
|
_clientLease.Release(outcome.Grants[0], "maint-end");
|
}
|
|
// ==================== 辅助 ====================
|
|
[Fact]
|
public async Task BrokerReceive_PiggybackPush_DeliveredToProxyDriver()
|
{
|
// 验收点：共享设备(Broker)能收到同总线对端发来的帧（经宿主捎带推送）。
|
// 修复历史上 BrokeredCanDriver 无接收事件、面板收不到 RX 的顽症。
|
var grant = GrabLease();
|
var hash = DeviceConfigHash.Compute(_dataDb);
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
|
await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
|
|
CanFrameish received = null;
|
driver.DriverFrameReceived += (s, f) => received = new CanFrameish { Id = f.FrameIdHex, Data = f.DataHex };
|
|
// 对端：同模拟器同总线 CAN_E2E 的节点 1，本地直连发帧
|
using var peer = new CanDeviceDriver("CAN-PEER", "CAN_E2E:1", "127.0.0.1", SimPort);
|
peer.ConnectAsync(CancellationToken.None).GetAwaiter().GetResult();
|
peer.SendCommandAsync("SEND:0x7E7:EE FF", CancellationToken.None).GetAwaiter().GetResult();
|
|
// 触发宿主 drain 捎带：任一非接收类查询（STATUS）即可带回推送
|
for (int i = 0; i < 20 && received == null; i++)
|
{
|
await driver.QueryAsync("STATUS", CancellationToken.None);
|
if (received == null) await Task.Delay(50);
|
}
|
|
Assert.NotNull(received);
|
Assert.Equal("0x7E7", received.Id);
|
Assert.Contains("EE FF", received.Data);
|
|
_clientLease.Release(grant, "test-end");
|
}
|
|
private sealed class CanFrameish { public string Id; public string Data; public string Src; }
|
|
[Fact]
|
public async Task BrokerResponderEcho_ViaBusEcho_DeliveredToProxyDriver()
|
{
|
// 复现用户场景：仅 Broker 设备自己发帧 + 总线应答器开启 → 应收到 RESPONDER 回显。
|
var grant = GrabLease();
|
var hash = DeviceConfigHash.Compute(_dataDb);
|
using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
|
await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
|
|
CanFrameish echo = null;
|
driver.DriverFrameReceived += (s, f) => echo = new CanFrameish { Id = f.FrameIdHex, Data = f.DataHex, Src = f.SourceNodeId };
|
|
// 经 RPC 开启总线应答器
|
var on = await driver.QueryAsync("BUS_ECHO:ON", CancellationToken.None);
|
Assert.StartsWith("OK:", on);
|
|
// Broker 设备自己发一帧
|
await driver.SendCommandAsync("SEND:0x111:AA BB CC", CancellationToken.None);
|
|
// 轮询 STATUS 触发宿主 drain 捎带，等回显送达
|
for (int i = 0; i < 30 && echo == null; i++)
|
{
|
await driver.QueryAsync("STATUS", CancellationToken.None);
|
if (echo == null) await Task.Delay(50);
|
}
|
|
Assert.NotNull(echo);
|
Assert.Equal("0x111", echo.Id);
|
Assert.Contains("RESPONDER", echo.Src);
|
|
_clientLease.Release(grant, "test-end");
|
}
|
|
// ==================== 辅助（原有） ====================
|
|
private LeaseGrant GrabLease()
|
{
|
var outcome = _clientLease.Acquire(new AcquireRequest
|
{
|
OwnerId = "e2e-" + Guid.NewGuid().ToString("N"),
|
OwnerType = OwnerTypeV3.Production,
|
ResourceKeys = new[] { DevId },
|
TtlSeconds = 30
|
});
|
Assert.Equal(AcquireResultCode.Acquired, outcome.Code);
|
return outcome.Grants[0];
|
}
|
|
private static async Task RetryAsync(Func<Task> action, int attempts = 4)
|
{
|
Exception last = null;
|
for (int i = 0; i < attempts; i++)
|
{
|
try { await action().ConfigureAwait(false); return; }
|
catch (InvalidOperationException ex) when (ex.Message.Contains("BROKER_UNREACHABLE"))
|
{
|
last = ex; // 宿主尚未开始监听，短暂重试
|
await Task.Delay(400).ConfigureAwait(false);
|
}
|
}
|
throw last ?? new InvalidOperationException("重试仍失败");
|
}
|
|
/// <summary>负向用例：握手可重试（管道未就位），但栅栏/哈希拒绝属终态应直接上抛。</summary>
|
private async Task RetryConnectAsync(CanDeviceDriver driver, bool expectThrow)
|
{
|
Exception last = null;
|
for (int i = 0; i < 4; i++)
|
{
|
try
|
{
|
await driver.ConnectAsync(CancellationToken.None).ConfigureAwait(false);
|
return; // 意外成功——由外层断言捕捉
|
}
|
catch (InvalidOperationException ex) when (ex.Message.Contains("BROKER_UNREACHABLE"))
|
{
|
last = ex;
|
await Task.Delay(400).ConfigureAwait(false);
|
}
|
catch (InvalidOperationException)
|
{
|
throw; // LEASE_STALE / CONFIG_MISMATCH 终态
|
}
|
}
|
throw last ?? new InvalidOperationException("未获得预期拒绝");
|
}
|
|
private void CreateDataDb()
|
{
|
using var conn = new System.Data.SQLite.SQLiteConnection(
|
$"Data Source={_dataDb};Version=3;");
|
conn.Open();
|
using (var ddl = conn.CreateCommand())
|
{
|
ddl.CommandText = @"
|
CREATE TABLE DeviceDefinition (
|
DeviceId TEXT PRIMARY KEY, Kind TEXT NOT NULL, DriverType TEXT NOT NULL,
|
Transport TEXT NOT NULL, Endpoint TEXT NOT NULL,
|
ExecutionMode TEXT NOT NULL DEFAULT 'Local',
|
AuthorityId TEXT NOT NULL DEFAULT '', BrokerId TEXT NOT NULL DEFAULT '',
|
OwnerProcessId TEXT NOT NULL DEFAULT '', PhysicalSessionKey TEXT NOT NULL DEFAULT '',
|
EffectiveVersion TEXT NOT NULL DEFAULT '', CreatedAt TEXT NOT NULL, UpdatedAt TEXT NOT NULL);";
|
ddl.ExecuteNonQuery();
|
}
|
using (var ins = conn.CreateCommand())
|
{
|
ins.CommandText = @"
|
INSERT INTO DeviceDefinition VALUES (
|
'CAN-B1','CanDevice','CAN','Can','CAN_E2E:0',
|
'Broker',@auth,@broker,'','can/CAN_E2E/main',
|
'V1','2026-11-08 00:00:00','2026-11-08 00:00:00');";
|
ins.Parameters.AddWithValue("@auth", _brokerId);
|
ins.Parameters.AddWithValue("@broker", _brokerId);
|
ins.ExecuteNonQuery();
|
}
|
}
|
}
|
}
|
