lmxopcua/tests/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/Snap7ServerFixture.cs

using System.Net.Sockets;

namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests;

/// <summary>
///     Reachability probe for a python-snap7 simulator (see
///     <c>PythonSnap7/serve.ps1</c>) or a real S7 PLC. Parses <c>S7_SIM_ENDPOINT</c>
///     (default <c>localhost:1102</c>) + TCP-connects once at fixture construction.
///     Tests check <see cref="SkipReason"/> + call <c>Assert.Skip</c> when unreachable, so
///     `dotnet test` stays green on a fresh box without the simulator installed —
///     mirrors the <c>ModbusSimulatorFixture</c> pattern.
/// </summary>
/// <remarks>
///     <para>
///         Default port is <b>1102</b>, not the S7-standard 102. 102 is a privileged port
///         on Linux (needs root) + triggers the Windows Firewall prompt on first bind;
///         1102 sidesteps both. S7netplus 0.20 supports the 5-arg <c>Plc</c> ctor that
///         takes an explicit port (verified + wired through <c>S7DriverOptions.Port</c>),
///         so the driver can reach the simulator on its non-standard port without
///         hacks.
///     </para>
///     <para>
///         The probe is a one-shot liveness check; tests open their own S7netplus
///         sessions against the same endpoint. Don't share a socket — S7 CPUs serialise
///         concurrent connections against the same mailbox anyway, but sharing would
///         couple test ordering to socket reuse in ways this harness shouldn't care
///         about.
///     </para>
///     <para>
///         Fixture is a collection fixture so the probe runs once per test session, not
///         per test.
///     </para>
/// </remarks>
public sealed class Snap7ServerFixture : IAsyncDisposable
{
    // Default 1102 (non-privileged) matches PythonSnap7/server.py. Override with
    // S7_SIM_ENDPOINT to point at a real PLC on its native 102.
    private const string DefaultEndpoint = "localhost:1102";
    private const string EndpointEnvVar = "S7_SIM_ENDPOINT";

    public string Host { get; }
    public int Port { get; }
    public string? SkipReason { get; }

    public Snap7ServerFixture()
    {
        var raw = Environment.GetEnvironmentVariable(EndpointEnvVar) ?? DefaultEndpoint;
        var parts = raw.Split(':', 2);
        Host = parts[0];
        Port = parts.Length == 2 && int.TryParse(parts[1], out var p) ? p : 102;

        try
        {
            // Force IPv4 — python-snap7 binds 0.0.0.0 (IPv4) and .NET's default
            // dual-stack "localhost" resolves IPv6 ::1 first then times out before
            // falling back. Same story the Modbus fixture hits.
            using var client = new TcpClient(AddressFamily.InterNetwork);
            var task = client.ConnectAsync(
                System.Net.Dns.GetHostAddresses(Host)
                    .FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetwork)
                    ?? System.Net.IPAddress.Loopback,
                Port);
            if (!task.Wait(TimeSpan.FromSeconds(2)) || !client.Connected)
            {
                SkipReason = $"python-snap7 simulator at {Host}:{Port} did not accept a TCP connection within 2s. " +
                             $"Start it (PythonSnap7\\serve.ps1 -Profile s7_1500) or override {EndpointEnvVar}.";
            }
        }
        catch (Exception ex)
        {
            SkipReason = $"python-snap7 simulator at {Host}:{Port} unreachable: {ex.GetType().Name}: {ex.Message}. " +
                         $"Start it (PythonSnap7\\serve.ps1 -Profile s7_1500) or override {EndpointEnvVar}.";
        }
    }

    public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}

[Xunit.CollectionDefinition(Name)]
public sealed class Snap7ServerCollection : Xunit.ICollectionFixture<Snap7ServerFixture>
{
    public const string Name = "Snap7Server";
}