Evergreen-Health/Admin.NET.Test/Health/Ib43MqttIntegrationTests.cs

using Evergreen.Health.Application.Health.Integrations.Ib43;
using Evergreen.Health.Application.Health.Entities;
using Evergreen.Health.Application.Health.Devices.Curtain;
using Evergreen.Health.Application.Health.Integrations.HomeAssistant;
using Evergreen.Health.Application.Health.Scenes;
using Evergreen.Health.Application.Health.Sleep;
using SqlSugar;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System.Net;
using System.Net.Sockets;
using System.Text.Json.Nodes;
using System.Text.Json;
using Xunit;

namespace Admin.NET.Test.Health;

public sealed class Ib43MqttIntegrationTests
{
    private const string LeftBinding = "a98788f2aeeb40b68f145a182d99f832";
    private const string TargetCurtain = "cover.test_curtain";

    [Fact]
    public async Task Real_MQTT_receives_supplier_examples_and_resubscribes_after_broker_restart()
    {
        // 只监听 loopback，随机端口；不访问用户的 EMQX、数据库或家具。
        using var reservation = new TcpListener(IPAddress.Loopback, 0);
        reservation.Start();
        var port = ((IPEndPoint)reservation.LocalEndpoint).Port;
        reservation.Stop();
        var factory = new MqttFactory();
        using var broker = factory.CreateMqttServer(new MqttServerOptionsBuilder()
            .WithDefaultEndpoint().WithDefaultEndpointPort(port)
            .WithDefaultEndpointBoundIPAddress(IPAddress.Loopback).Build());
        await broker.StartAsync();
        var binding = new Ib43Binding { ProductKey = "evergreen-test", SerialNumber = "IB43SIM001" };
        var options = Options.Create(new Ib43Options { Enabled = true, Host = "127.0.0.1", Port = port,
            Username = "local-test", Password = "local-test-only", Devices = [binding] });
        var store = new Ib43StateStore(options, new(), TimeProvider.System);
        using var sink = new DatabaseSink(binding);
        var gateway = new FakeHaGateway { CoverCloseResult = "closed" };
        gateway.States[TargetCurtain] = new() { EntityId = TargetCurtain, State = "open",
            Attributes = JsonSerializer.SerializeToElement(new { supported_features = 2 }) };
        var haOptions = Options.Create(new HomeAssistantOptions { Entities = new() { Curtain = TargetCurtain } });
        var execution = new DeviceExecutionOptions { VerificationDelay = TimeSpan.Zero };
        var runner = new DeviceActionRunner(gateway, TimeProvider.System, execution);
        var curtainAction = new BedPresenceCurtainCloseAction(new CurtainModule(runner, haOptions), runner, haOptions);
        using var automation = new BedPresenceAutomation(Options.Create(new BedPresenceAutomationOptions
            { Mode = BedAutomationMode.Execute, BindingId = LeftBinding, ActionKey = "curtain-close", TargetEntityId = TargetCurtain }),
            [curtainAction], TimeProvider.System, NullLogger<BedPresenceAutomation>.Instance);
        using var service = new Ib43MqttService(options, new Ib43MqttTransportFactory(), store,
            TimeProvider.System, NullLogger<Ib43MqttService>.Instance, sink, automation);
        using var publisher = factory.CreateMqttClient();
        try
        {
            await automation.StartAsync(default);
            await service.StartAsync(default);
            await WaitFor(() => store.Snapshot()[0].Transport == "connected");
            await publisher.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", port)
                .WithClientId("ib43-local-simulator").Build());
            foreach (var name in new[] { "startup.pdf-example.json", "realtime.pdf-example.json", "sleep-event.pdf-keys.json" })
            {
                var json = JsonNode.Parse(File.ReadAllText(Path.Combine(AppContext.BaseDirectory, "Ib43Examples", name)));
                json["params"]["deviceName"] = binding.SerialNumber;
                if (json["params"]["productKey"] != null) json["params"]["productKey"] = binding.ProductKey;
                await publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(binding.PostTopic)
                    .WithPayload(json.ToJsonString()).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce).Build());
            }
            await WaitFor(() => store.Snapshot()[0].Readings.TryGetValue("left.heartRate", out var value) && value.Value == 74);
            Assert.Equal(80d, store.Snapshot()[0].Readings["right.heartRate"].Value);
            Assert.Equal(60, store.Snapshot()[0].Fields["characteristic"].Value.GetProperty("reportIerVal").GetInt32());
            Assert.All(store.Diagnostics(), entry => Assert.Empty(entry.Issues));
            await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 2);
            Assert.Equal(74d, (await sink.Repository.Latest(1, LeftBinding))["heartRate"].Value);
            Assert.Equal(80d, (await sink.Repository.Latest(2, "right-test"))["heartRate"].Value);
            Assert.Empty(gateway.Calls);

            // 左侧空床后上床才触发。连续重复在床不会重启，右侧的变化不触发左侧规则。
            await PublishPresence(publisher, binding, 0, 0);
            await PublishPresence(publisher, binding, 0, 1);
            await PublishPresence(publisher, binding, 1, 1);
            await PublishPresence(publisher, binding, 1, 1);
            await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 6 && gateway.Calls.Count == 1);
            var close = Assert.Single(gateway.Calls);
            Assert.Equal(TargetCurtain, close.EntityId);
            Assert.Equal("close_cover", close.Service);
            Assert.Equal("closed", gateway.States[TargetCurtain].State);
            Assert.Equal(1d, (await sink.Repository.Latest(1, LeftBinding))["presence"].Value);

            await broker.StopAsync();
            await WaitFor(() => store.Snapshot()[0].Transport != "connected");
            await broker.StartAsync();
            await WaitFor(() => store.Snapshot()[0].Transport == "connected");
            using var publisherAfterRestart = factory.CreateMqttClient();
            await publisherAfterRestart.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", port).Build());
            await publisherAfterRestart.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(binding.PostTopic)
                .WithPayload("{\"method\":\"thing.property.post\",\"params\":{\"deviceName\":\"IB43SIM001\",\"HR\":[75,16,0,81,18,0]}}")
                .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce).Build());
            await WaitFor(() => store.Snapshot()[0].Readings["left.heartRate"].Value == 75);
            Assert.Equal(1d, store.Snapshot()[0].Readings["left.presence"].Value);
            await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 7);
            Assert.Equal(75d, (await sink.Repository.Latest(1, LeftBinding))["heartRate"].Value);
            Assert.Single(gateway.Calls);
        }
        finally
        {
            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
            await service.StopAsync(timeout.Token);
            await automation.StopAsync(timeout.Token);
            await broker.StopAsync();
        }
    }

    private sealed class DatabaseSink : IIb43ReportSink, IDisposable
    {
        private readonly SqlSugarClient _db;
        public readonly Ib43DataRepository Repository;
        public int SavedReports;
        public DatabaseSink(Ib43Binding binding)
        {
            _db = new(new ConnectionConfig { ConnectionString = "DataSource=:memory:", DbType = DbType.Sqlite,
                IsAutoCloseConnection = false, InitKeyType = InitKeyType.Attribute,
                ConfigureExternalServices = new ConfigureExternalServices { EntityService = (p, c) =>
                {
                    if (p.DeclaringType == typeof(HealthIb43Observation) && p.Name == "Id") c.DataType = "INTEGER";
                } } });
            _db.CodeFirst.InitTables<HealthIb43Binding, HealthIb43Observation>();
            Repository = new(_db);
            var key = Ib43DataRepository.DeviceKey(binding.ProductKey, binding.SerialNumber);
            foreach (var (side, user) in new[] { ("left", 1L), ("right", 2L) })
                _db.Insertable(new HealthIb43Binding { Id = side == "left" ? LeftBinding : "right-test", DeviceKey = key,
                    ProductKey = binding.ProductKey, SerialNumber = binding.SerialNumber, Side = side,
                    ActiveSlot = key + "." + side, UserId = user, Simulation = true }).ExecuteCommand();
        }
        public async Task<Ib43SavedReport> SaveAsync(string reportId, Ib43Binding binding, Ib43Report report, CancellationToken token)
        {
            var count = await Repository.Save(reportId, binding, report);
            Interlocked.Increment(ref SavedReports);
            return count;
        }
        public void Dispose() => _db.Dispose();
    }

    private static Task<MqttClientPublishResult> PublishPresence(IMqttClient publisher, Ib43Binding binding, int left, int right)
    {
        var payload = JsonSerializer.Serialize(new
        {
            method = "thing.property.post",
            @params = new { deviceName = binding.SerialNumber, SleepReportNew = new { ibNew = new[] { left, 0, 0, 16, 74, 0, right, 0, 0, 17, 80, 0 } } }
        });
        return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(binding.PostTopic).WithPayload(payload)
            .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce).Build());
    }

    private static async Task WaitFor(Func<bool> predicate)
    {
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(12));
        while (!predicate()) await Task.Delay(20, timeout.Token);
    }
}