dohertj2/lmxopcua
1
0
Fork 0
Code Issues 17 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

test(mqtt): Mosquitto TLS+auth fixture + env-gated plain live suite

Browse Source 
Adds the MQTT driver integration-test project: an eclipse-mosquitto fixture running
auth AND TLS (allow_anonymous false on both listeners), a mosquitto_pub-based JSON
publisher emitting retained messages, a cert/password generator script whose output is
gitignored, and a live suite gated on MQTT_FIXTURE_ENDPOINT that skips clean offline.

The fixture is never anonymous by design: an anonymous broker would let the driver's
TLS/CA-pin and auth paths go untested, and those fail silently. The CA-pin leg carries
its own falsifiability control (a foreign CA generated in-process must be rejected).

Live gate on 10.100.0.35: 7/7 passed. It found a driver defect, reported not fixed
here (src/ is out of this task's scope): MqttConnection.ConnectAsync RETURNS NORMALLY
when Mosquitto rejects a CONNECT as not-authorized -- MQTTnet 5 does not throw on an
unsuccessful CONNACK, so a wrong broker password yields DriverState.Healthy from
InitializeAsync. The auth-negative test therefore asserts the invariant that holds
either way (no session is ever established) and documents the finding.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 17:49:14 -04:00
parent ff62b62a55
commit 033b3700c4
9 changed files with 1231 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ZB.MOM.WW.OtOpcUa.slnx
+1
View File
@@ -108,6 +108,7 @@
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests.csproj" />
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.IntegrationTests.csproj" />
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj" />
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj" />
</Folder>
<Folder Name="/tests/Drivers/Driver CLIs/">
<Project Path="tests/Drivers/Cli/ZB.MOM.WW.OtOpcUa.Driver.Cli.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Cli.Common.Tests.csproj" />
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore
+14
View File
@@ -0,0 +1,14 @@
# Fixture credential + TLS material produced by gen-fixture-material.sh.
#
# NEVER commit anything in here: it holds the broker password hash, the fixture CA's
# PRIVATE key and the server's private key. The generator script is the artifact in
# git; its output is regenerated per machine.
secrets/
# Belt-and-braces: catch key/cert material written directly into this directory by a
# hand-run openssl command that forgot the secrets/ prefix.
*.key
*.crt
*.csr
*.srl
passwd
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml
+77
View File
@@ -0,0 +1,77 @@
# MQTT integration-test fixture — Eclipse Mosquitto (auth + TLS) + a JSON publisher.
#
# PREREQUISITE — run this first, in this directory:
#
# MQTT_FIXTURE_PASSWORD='<pick-one>' ./gen-fixture-material.sh
#
# It writes ./secrets/ (password file + CA + server certificate/key), which is
# gitignored and which both services below mount. Without it the broker will not
# start: there is no anonymous fallback, by design.
#
# Bring up (on the shared Docker host — see infra/README.md §1):
#
# ssh dohertj2@10.100.0.35 'cd /opt/otopcua-mqtt \
# && MQTT_FIXTURE_USERNAME=otopcua MQTT_FIXTURE_PASSWORD=<same> docker compose up -d'
#
# Endpoints: 10.100.0.35:8883 (TLS + auth) · 10.100.0.35:1883 (plaintext + auth, smoke)
#
# Unlike the Modbus / S7 fixtures there are no compose profiles: both services are
# always wanted together (a broker with nothing publishing into it tests nothing), and
# they bind different ports so nothing contends.
services:
mosquitto:
image: eclipse-mosquitto:2.0.22
container_name: otopcua-mosquitto
restart: unless-stopped
# Every lmxopcua fixture container carries this so the shared Docker host stays
# discoverable with `docker ps --filter label=project=lmxopcua`.
labels:
project: lmxopcua
ports:
- "1883:1883"
- "8883:8883"
volumes:
- ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro
- ./secrets:/mosquitto/secrets:ro
healthcheck:
# Authenticated round-trip against the broker's own $SYS tree ($$ escapes the
# compose variable syntax). Proves the password file loaded and the listener
# accepts a real session — a bare port check would go green on a broker that
# rejects every CONNECT.
test:
- CMD-SHELL
- >-
mosquitto_sub -h 127.0.0.1 -p 1883
-u "$$MQTT_FIXTURE_USERNAME" -P "$$MQTT_FIXTURE_PASSWORD"
-t '$$SYS/broker/uptime' -C 1 -W 3
interval: 5s
timeout: 8s
retries: 12
start_period: 5s
environment:
# Consumed only by the healthcheck above; the broker itself reads the hashed
# password file. Supplied by the operator's environment at `docker compose up` —
# never defaulted here, so a missing value fails loudly instead of silently
# provisioning a known-password broker.
MQTT_FIXTURE_USERNAME: ${MQTT_FIXTURE_USERNAME:?set MQTT_FIXTURE_USERNAME (must match gen-fixture-material.sh)}
MQTT_FIXTURE_PASSWORD: ${MQTT_FIXTURE_PASSWORD:?set MQTT_FIXTURE_PASSWORD (must match gen-fixture-material.sh)}
publisher:
image: eclipse-mosquitto:2.0.22
container_name: otopcua-mqtt-publisher
restart: unless-stopped
labels:
project: lmxopcua
depends_on:
mosquitto:
condition: service_healthy
volumes:
- ./publisher/publish.sh:/publish.sh:ro
environment:
BROKER_HOST: mosquitto
BROKER_PORT: "1883"
TOPIC_PREFIX: ${MQTT_FIXTURE_TOPIC_PREFIX:-otopcua/fixture}
PUBLISH_INTERVAL: ${MQTT_FIXTURE_PUBLISH_INTERVAL:-2}
MQTT_FIXTURE_USERNAME: ${MQTT_FIXTURE_USERNAME:?set MQTT_FIXTURE_USERNAME}
MQTT_FIXTURE_PASSWORD: ${MQTT_FIXTURE_PASSWORD:?set MQTT_FIXTURE_PASSWORD}
entrypoint: ["/bin/sh", "/publish.sh"]
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/gen-fixture-material.sh
Executable
+137
View File
@@ -0,0 +1,137 @@
#!/usr/bin/env sh
# ---------------------------------------------------------------------------
# Generates the MQTT fixture's password file + TLS material into ./secrets/.
#
# NOTHING this script writes is committed: ./secrets/ is excluded by the sibling
# .gitignore. The SCRIPT is the artifact in git, the OUTPUT never is. Re-run it on any
# machine that needs the fixture (or on the Docker host); the material is throwaway by
# construction — a self-issued CA, a 2048-bit server key, and whatever password the
# operator supplies in the environment.
#
# Usage:
# MQTT_FIXTURE_PASSWORD='<pick-one>' ./gen-fixture-material.sh
#
# Environment:
# MQTT_FIXTURE_PASSWORD (required) broker password. No default — a committed or
# defaulted password is exactly what the plan's
# "secrets from env, never committed" rule forbids.
# MQTT_FIXTURE_USERNAME broker username (default: otopcua)
# MQTT_FIXTURE_SAN_IPS comma-separated IP SANs for the server certificate
# (default: 10.100.0.35,127.0.0.1)
# MQTT_FIXTURE_SAN_DNS comma-separated DNS SANs
# (default: localhost,mosquitto)
# MQTT_FIXTURE_CERT_CN server certificate CN (default: first IP SAN)
# MQTT_FIXTURE_CERT_DAYS certificate lifetime in days (default: 825)
# MQTT_FIXTURE_SECRETS_DIR output directory (default: <script dir>/secrets)
# MOSQUITTO_IMAGE image used to run mosquitto_passwd
# (default: eclipse-mosquitto:2.0.22)
#
# The SANs matter: MqttConnection pins the TLS target host with
# `.WithTargetHost(options.Host)`, so hostname verification is enforced and the
# certificate MUST carry a SAN matching whatever host the tests dial — the shared
# Docker host's IP literal (10.100.0.35) in the normal case. A cert without the right
# SAN fails the handshake even with a correct CA pin, and the failure looks like a pin
# bug.
#
# Requires: openssl, docker (only for hashing the password — mosquitto 2.0 will not
# read a plaintext password file, and mosquitto_passwd is the only supported way to
# produce its PBKDF2-SHA512 format).
# ---------------------------------------------------------------------------
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
OUT=${MQTT_FIXTURE_SECRETS_DIR:-"$SCRIPT_DIR/secrets"}
USERNAME=${MQTT_FIXTURE_USERNAME:-otopcua}
SAN_IPS=${MQTT_FIXTURE_SAN_IPS:-10.100.0.35,127.0.0.1}
SAN_DNS=${MQTT_FIXTURE_SAN_DNS:-localhost,mosquitto}
DAYS=${MQTT_FIXTURE_CERT_DAYS:-825}
MOSQUITTO_IMAGE=${MOSQUITTO_IMAGE:-eclipse-mosquitto:2.0.22}
if [ -z "${MQTT_FIXTURE_PASSWORD:-}" ]; then
echo "error: MQTT_FIXTURE_PASSWORD is not set." >&2
echo " Supply the broker password in the environment; this script never" >&2
echo " defaults or commits one. Example:" >&2
echo " MQTT_FIXTURE_PASSWORD='\$(openssl rand -base64 18)' $0" >&2
exit 2
fi
command -v openssl >/dev/null 2>&1 || { echo "error: openssl not found on PATH." >&2; exit 2; }
command -v docker >/dev/null 2>&1 || {
echo "error: docker not found on PATH — needed to run mosquitto_passwd." >&2
echo " Run this script on the Docker host (10.100.0.35) instead, or install docker." >&2
exit 2
}
CN=${MQTT_FIXTURE_CERT_CN:-$(printf '%s' "$SAN_IPS" | cut -d, -f1)}
mkdir -p "$OUT"
chmod 0755 "$OUT"
echo "==> generating fixture material into $OUT"
echo " username : $USERNAME"
echo " cert CN : $CN"
echo " IP SANs : $SAN_IPS"
echo " DNS SANs : $SAN_DNS"
# --- CA -------------------------------------------------------------------
openssl req -x509 -newkey rsa:2048 -nodes \
-keyout "$OUT/ca.key" -out "$OUT/ca.crt" -days "$DAYS" \
-subj "/CN=OtOpcUa MQTT fixture CA/O=OtOpcUa test fixture" \
-addext "basicConstraints=critical,CA:TRUE,pathlen:0" \
-addext "keyUsage=critical,keyCertSign,cRLSign" \
>/dev/null 2>&1
# --- server leaf ----------------------------------------------------------
# The SAN list is assembled into a temp extfile rather than a process substitution,
# which is a bashism this /bin/sh script cannot use.
EXT=$(mktemp)
trap 'rm -f "$EXT" "$OUT/server.csr" "$OUT/ca.srl"' EXIT
SAN=""
OLD_IFS=$IFS
IFS=,
for ip in $SAN_IPS; do [ -n "$ip" ] && SAN="$SAN,IP:$ip"; done
for dn in $SAN_DNS; do [ -n "$dn" ] && SAN="$SAN,DNS:$dn"; done
IFS=$OLD_IFS
SAN=${SAN#,}
cat >"$EXT" <<EOF
subjectAltName = $SAN
basicConstraints = critical,CA:FALSE
keyUsage = critical,digitalSignature,keyEncipherment
extendedKeyUsage = serverAuth
EOF
openssl req -newkey rsa:2048 -nodes \
-keyout "$OUT/server.key" -out "$OUT/server.csr" \
-subj "/CN=$CN/O=OtOpcUa test fixture" \
>/dev/null 2>&1
openssl x509 -req -in "$OUT/server.csr" \
-CA "$OUT/ca.crt" -CAkey "$OUT/ca.key" -CAcreateserial \
-out "$OUT/server.crt" -days "$DAYS" -sha256 -extfile "$EXT" \
>/dev/null 2>&1
# --- password file --------------------------------------------------------
# `--user` so the file lands owned by the invoking user rather than by root: the host
# user must be able to chmod it below, and (unlike a root-owned 0600 file) the broker
# container's `mosquitto` user must be able to READ it.
rm -f "$OUT/passwd"
docker run --rm --user "$(id -u):$(id -g)" -v "$OUT":/out "$MOSQUITTO_IMAGE" \
mosquitto_passwd -b -c /out/passwd "$USERNAME" "$MQTT_FIXTURE_PASSWORD" >/dev/null
# mosquitto runs as uid 1883 inside the container and reads these through a read-only
# bind mount, so they must be world-readable. This is a throwaway fixture key by
# construction — see the banner.
chmod 0644 "$OUT/ca.crt" "$OUT/server.crt" "$OUT/server.key" "$OUT/passwd"
chmod 0600 "$OUT/ca.key"
echo "==> done. Files written:"
ls -l "$OUT"
echo
echo "Next:"
echo " 1. Export the same creds for the test suite:"
echo " export MQTT_FIXTURE_USERNAME='$USERNAME'"
echo " export MQTT_FIXTURE_PASSWORD='<the same password>'"
echo " export MQTT_FIXTURE_CA_CERT='$OUT/ca.crt'"
echo " 2. Push this directory to the Docker host and bring the stack up."
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/mosquitto.conf
+65
View File
@@ -0,0 +1,65 @@
# Eclipse Mosquitto configuration for the OtOpcUa MQTT driver integration fixture.
#
# SECURITY POSTURE — deliberate, and the whole reason this fixture exists.
# `allow_anonymous false` is set GLOBALLY (with per_listener_settings false), so it
# governs BOTH listeners. An anonymous fixture would let the driver's auth and TLS
# code paths — including the CA-pin logic in MqttConnection.ConfigureTls /
# ValidateAgainstPinnedCa — go completely untested, which is exactly backwards: those
# are the paths whose failure mode is silent (a pin that never rejects anything looks
# identical to a pin that works).
#
# The password file and the TLS material are NOT in git. Generate them first with
# `./gen-fixture-material.sh` (see that script's banner); it writes ./secrets/, which
# .gitignore excludes.
# One global security policy rather than per-listener ones: with
# per_listener_settings true, `allow_anonymous`/`password_file` would have to be
# repeated under EVERY listener block, and a listener that forgot them would silently
# fall back to allowing anonymous connections. Global + fail-closed is the safer shape.
per_listener_settings false
allow_anonymous false
password_file /mosquitto/secrets/passwd
# No persistence: the fixture is throwaway, and a persisted retained-message store
# surviving a `docker compose down` would make the retained-seed test's outcome depend
# on a previous run rather than on this one's publisher.
persistence false
log_dest stdout
log_type error
log_type warning
log_type notice
log_type information
connection_messages true
# ---------------------------------------------------------------------------
# Plain (non-TLS) listener — smoke only.
#
# Still authenticated (allow_anonymous false above applies here too); it is the
# TRANSPORT that is plaintext, not the access policy. It exists so the subscribe /
# retained-seed / unauthored-topic legs can be exercised without dragging the TLS
# handshake into every assertion, and so an operator can `mosquitto_sub` the fixture
# by hand. Never a model for a deployment.
# ---------------------------------------------------------------------------
listener 1883 0.0.0.0
protocol mqtt
# ---------------------------------------------------------------------------
# TLS listener — the real one.
#
# `require_certificate false`: the driver authenticates with username/password over
# TLS (the shape MqttDriverOptions models today); mutual TLS is not a driver feature
# yet, so demanding a client certificate here would make the fixture untestable rather
# than more secure.
#
# tls_version is deliberately NOT pinned: mosquitto's `tls_version` selects exactly
# one version, so pinning 1.2 would refuse a .NET client that negotiated 1.3. Leaving
# it unset lets OpenSSL negotiate the best mutually-supported version.
# ---------------------------------------------------------------------------
listener 8883 0.0.0.0
protocol mqtt
cafile /mosquitto/secrets/ca.crt
certfile /mosquitto/secrets/server.crt
keyfile /mosquitto/secrets/server.key
require_certificate false
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/publisher/publish.sh
Executable
+67
View File
@@ -0,0 +1,67 @@
#!/bin/sh
# ---------------------------------------------------------------------------
# The fixture's JSON publisher. Runs inside a second eclipse-mosquitto container (the
# image already ships mosquitto_pub, so no bespoke image / paho dependency is needed)
# and feeds the broker over the PLAIN listener from inside the compose network —
# authenticated like every other client, since the broker is allow_anonymous false.
#
# Topics (all under $TOPIC_PREFIX, default otopcua/fixture):
#
# retained/seed published EXACTLY ONCE, retain=true, never republished.
# This is what makes the retained-seed test meaningful: a
# subscriber that connects minutes later and still receives a
# value on this topic can only have received the BROKER'S
# RETAINED COPY. On a topic that is also republished every few
# seconds, a retained seed is indistinguishable from a live
# publish that happened to arrive quickly.
# line1/temperature JSON object, retain=true, republished every $PUBLISH_INTERVAL
# line1/counter JSON object, retain=true, monotonically increasing
# line1/state bare scalar text (no JSON), retain=true
# noise/chatter JSON, retain=FALSE — deliberately never authored as a tag.
# A broker carrying traffic the deployment did not author is
# the normal case, and the driver must stay silent about it.
# ---------------------------------------------------------------------------
set -eu
HOST=${BROKER_HOST:-mosquitto}
PORT=${BROKER_PORT:-1883}
PREFIX=${TOPIC_PREFIX:-otopcua/fixture}
INTERVAL=${PUBLISH_INTERVAL:-2}
: "${MQTT_FIXTURE_USERNAME:?publisher needs MQTT_FIXTURE_USERNAME}"
: "${MQTT_FIXTURE_PASSWORD:?publisher needs MQTT_FIXTURE_PASSWORD}"
pub_retained() {
mosquitto_pub -h "$HOST" -p "$PORT" \
-u "$MQTT_FIXTURE_USERNAME" -P "$MQTT_FIXTURE_PASSWORD" \
-q 1 -r -t "$1" -m "$2"
}
pub_volatile() {
mosquitto_pub -h "$HOST" -p "$PORT" \
-u "$MQTT_FIXTURE_USERNAME" -P "$MQTT_FIXTURE_PASSWORD" \
-q 0 -t "$1" -m "$2"
}
echo "publisher: broker=$HOST:$PORT prefix=$PREFIX interval=${INTERVAL}s"
# The one-shot retained seed. Published before the loop starts and never again.
pub_retained "$PREFIX/retained/seed" \
'{"value":42.5,"unit":"C","note":"published once at fixture start; never republished"}'
echo "publisher: retained seed published on $PREFIX/retained/seed"
n=0
while true; do
n=$((n + 1))
temperature=$(awk "BEGIN { printf \"%.2f\", 20 + ($n % 50) / 10 }")
pub_retained "$PREFIX/line1/temperature" \
"{\"value\":$temperature,\"unit\":\"C\",\"seq\":$n}"
pub_retained "$PREFIX/line1/counter" "{\"value\":$n}"
pub_retained "$PREFIX/line1/state" "RUNNING"
# Unauthored chatter — the driver must never materialise a tag for this.
pub_volatile "$PREFIX/noise/chatter" "{\"value\":$n,\"source\":\"chatter\"}"
sleep "$INTERVAL"
done
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttFixture.cs
+205
View File
@@ -0,0 +1,205 @@
using System.Net.Sockets;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
/// <summary>
/// Environment gate + reachability probe for the Mosquitto fixture defined in
/// <c>Docker/docker-compose.yml</c> (auth + TLS on <c>:8883</c>, authenticated plaintext on
/// <c>:1883</c>, plus a JSON publisher emitting retained messages).
/// <para>
/// <b>Unset env ⇒ skip, never fail.</b> Unlike the Modbus / S7 fixtures — which default to
/// the shared Docker host and probe it unconditionally — this one has <b>no default
/// endpoint</b>: without <c>MQTT_FIXTURE_ENDPOINT</c> the whole suite reports
/// <see cref="NotConfigured"/> and every test calls <c>Assert.Skip</c>. That is deliberate.
/// The fixture needs credentials that are generated per machine and never committed, so
/// "probe a well-known host and hope" would produce auth failures rather than a clean skip
/// on a dev box that never provisioned it. `dotnet test` on macOS with no Docker therefore
/// stays green.
/// </para>
/// </summary>
/// <remarks>
/// <para>
/// Env vars, all read once at fixture construction:
/// </para>
/// <list type="bullet">
/// <item>
/// <c>MQTT_FIXTURE_ENDPOINT</c> — <c>host:port</c> of the <b>TLS</b> listener, e.g.
/// <c>10.100.0.35:8883</c>. <b>The gate</b>: absent ⇒ the whole suite skips.
/// </item>
/// <item>
/// <c>MQTT_FIXTURE_PLAIN_ENDPOINT</c> — <c>host:port</c> of the plaintext listener.
/// Defaults to the TLS host on port 1883.
/// </item>
/// <item>
/// <c>MQTT_FIXTURE_USERNAME</c> / <c>MQTT_FIXTURE_PASSWORD</c> — the broker credentials
/// <c>gen-fixture-material.sh</c> was run with. The password has no default and is
/// never committed; without it the suite skips.
/// </item>
/// <item>
/// <c>MQTT_FIXTURE_CA_CERT</c> — path to the fixture CA (<c>secrets/ca.crt</c>).
/// Optional: supplied, the TLS legs pin to it and the CA-pin test runs; absent, the
/// connect legs fall back to <c>AllowUntrustedServerCertificate</c> and the pin test
/// skips.
/// </item>
/// <item>
/// <c>MQTT_FIXTURE_TOPIC_PREFIX</c> — must match the publisher's. Default
/// <c>otopcua/fixture</c>.
/// </item>
/// </list>
/// <para>
/// The probe is one-shot and its socket is closed immediately: tests open their own real
/// <see cref="MqttConnection"/> / <see cref="MqttDriver"/> sessions, and sharing a socket
/// would serialise them.
/// </para>
/// </remarks>
public sealed class MqttFixture
{
/// <summary>The gate: absent ⇒ the whole suite skips.</summary>
public const string EndpointEnvVar = "MQTT_FIXTURE_ENDPOINT";
/// <summary>Optional override for the plaintext listener; defaults to the TLS host on 1883.</summary>
public const string PlainEndpointEnvVar = "MQTT_FIXTURE_PLAIN_ENDPOINT";
/// <summary>Broker username; defaults to the generator script's own default.</summary>
public const string UsernameEnvVar = "MQTT_FIXTURE_USERNAME";
/// <summary>Broker password. No default — a defaulted fixture password is a committed secret.</summary>
public const string PasswordEnvVar = "MQTT_FIXTURE_PASSWORD";
/// <summary>Path to the fixture CA PEM, for the CA-pin legs.</summary>
public const string CaCertEnvVar = "MQTT_FIXTURE_CA_CERT";
/// <summary>Topic prefix the fixture publisher emits under.</summary>
public const string TopicPrefixEnvVar = "MQTT_FIXTURE_TOPIC_PREFIX";
private const string DefaultUsername = "otopcua";
private const string DefaultTopicPrefix = "otopcua/fixture";
private const int DefaultPlainPort = 1883;
private static readonly TimeSpan ProbeTimeout = TimeSpan.FromSeconds(3);
/// <summary>Initializes a new instance of the <see cref="MqttFixture"/> class.</summary>
public MqttFixture()
{
Username = Environment.GetEnvironmentVariable(UsernameEnvVar) is { Length: > 0 } u ? u : DefaultUsername;
Password = Environment.GetEnvironmentVariable(PasswordEnvVar);
TopicPrefix = Environment.GetEnvironmentVariable(TopicPrefixEnvVar) is { Length: > 0 } p
? p.TrimEnd('/')
: DefaultTopicPrefix;
var caPath = Environment.GetEnvironmentVariable(CaCertEnvVar);
CaCertificatePath = string.IsNullOrWhiteSpace(caPath) ? null : caPath;
var tlsRaw = Environment.GetEnvironmentVariable(EndpointEnvVar);
if (string.IsNullOrWhiteSpace(tlsRaw))
{
SkipReason =
$"Skipped: set {EndpointEnvVar} (e.g. 10.100.0.35:8883) plus {UsernameEnvVar}/{PasswordEnvVar} to run "
+ "the MQTT fixture live suite. Bring the fixture up first — see "
+ "tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml.";
return;
}
(TlsHost, TlsPort) = ParseEndpoint(tlsRaw, defaultPort: 8883);
var plainRaw = Environment.GetEnvironmentVariable(PlainEndpointEnvVar);
(PlainHost, PlainPort) = string.IsNullOrWhiteSpace(plainRaw)
? (TlsHost, DefaultPlainPort)
: ParseEndpoint(plainRaw, DefaultPlainPort);
if (string.IsNullOrEmpty(Password))
{
// A blank password is NOT a valid "anonymous" configuration for this fixture — the
// broker runs allow_anonymous false — so treat it as unconfigured rather than letting
// every test fail with an indistinguishable "not authorized".
SkipReason =
$"Skipped: {EndpointEnvVar} is set but {PasswordEnvVar} is not. The fixture broker runs "
+ "allow_anonymous=false; supply the password gen-fixture-material.sh was run with.";
return;
}
// Probe the TLS listener: it is the endpoint the gate names, and a broker that is up serves
// both listeners from one process.
SkipReason = Probe(TlsHost, TlsPort);
}
/// <summary>Gets the TLS listener host.</summary>
public string TlsHost { get; } = string.Empty;
/// <summary>Gets the TLS listener port.</summary>
public int TlsPort { get; }
/// <summary>Gets the plaintext listener host.</summary>
public string PlainHost { get; } = string.Empty;
/// <summary>Gets the plaintext listener port.</summary>
public int PlainPort { get; }
/// <summary>Gets the broker username.</summary>
public string Username { get; }
/// <summary>Gets the broker password, or <see langword="null"/> when unconfigured.</summary>
public string? Password { get; }
/// <summary>Gets the fixture CA PEM path, or <see langword="null"/> when not supplied.</summary>
public string? CaCertificatePath { get; }
/// <summary>Gets the topic prefix the fixture publisher emits under (no trailing slash).</summary>
public string TopicPrefix { get; }
/// <summary>
/// Gets the reason every test in the suite must skip, or <see langword="null"/> when the
/// fixture is configured and reachable.
/// </summary>
public string? SkipReason { get; }
/// <summary>Gets a value indicating whether the suite must skip.</summary>
public bool NotConfigured => SkipReason is not null;
/// <summary>Composes a fixture topic under <see cref="TopicPrefix"/>.</summary>
/// <param name="suffix">The topic suffix, e.g. <c>line1/temperature</c>.</param>
/// <returns>The full topic.</returns>
public string Topic(string suffix) => $"{TopicPrefix}/{suffix.TrimStart('/')}";
private static (string Host, int Port) ParseEndpoint(string raw, int defaultPort)
{
var parts = raw.Trim().Split(':', 2);
var port = parts.Length == 2 && int.TryParse(parts[1], out var parsed) ? parsed : defaultPort;
return (parts[0], port);
}
private static string? Probe(string host, int port)
{
try
{
using var client = new TcpClient();
var connect = client.ConnectAsync(host, port);
if (!connect.Wait(ProbeTimeout) || !client.Connected)
{
return Unreachable(host, port, $"no TCP connection within {ProbeTimeout.TotalSeconds:0}s");
}
}
catch (Exception ex)
{
return Unreachable(host, port, $"{ex.GetType().Name}: {ex.Message}");
}
return null;
}
private static string Unreachable(string host, int port, string detail) =>
$"Skipped: the MQTT fixture broker at {host}:{port} is configured but unreachable ({detail}). "
+ "Bring the stack up on the Docker host: "
+ "`ssh dohertj2@10.100.0.35 'cd /opt/otopcua-mqtt && MQTT_FIXTURE_USERNAME=... MQTT_FIXTURE_PASSWORD=... docker compose up -d'`.";
}
/// <summary>
/// Collection definition so the reachability probe runs once per test session rather than once
/// per test — probing a firewalled endpoint per test would burn the probe timeout each time.
/// </summary>
[Xunit.CollectionDefinition(Name)]
public sealed class MqttFixtureCollection : Xunit.ICollectionFixture<MqttFixture>
{
/// <summary>The collection name.</summary>
public const string Name = "MqttFixture";
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs
+616
View File
@@ -0,0 +1,616 @@
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using MQTTnet;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
/// <summary>
/// Live validation of the plain-MQTT (P1) driver against the real Mosquitto fixture in
/// <c>Docker/docker-compose.yml</c> — auth + TLS on <c>:8883</c>, authenticated plaintext on
/// <c>:1883</c>, and a publisher emitting retained JSON. Every component exercised here is the
/// production one: <see cref="MqttConnection"/> (broker session, TLS, CA pin),
/// <see cref="MqttSubscriptionManager"/> (SUBSCRIBE, routing, extraction) and
/// <see cref="MqttDriver"/> (the <c>ISubscribable</c>/<c>IReadable</c> surface).
/// <para>
/// <b>Env-gated + skip-clean.</b> Without <c>MQTT_FIXTURE_ENDPOINT</c> (and the credentials)
/// every test calls <c>Assert.Skip</c>, so <c>dotnet test</c> is green on a laptop with no
/// Docker. See <see cref="MqttFixture"/> for the full env-var list.
/// </para>
/// <para>
/// <b>Why these four things.</b> The TLS/auth legs are the first time the Task-3 handshake
/// code meets a real broker — until now the CA-pin path had only ever seen synthetic certs,
/// and its failure mode (a pin that accepts everything) is invisible without a negative
/// control, which is why <see cref="Tls_connect_pinned_to_a_foreign_ca_is_rejected"/> exists
/// alongside the positive. The subscribe leg asserts the published <b>RawPath</b>, the single
/// most defect-prone identity in this plan. And the unauthored-topic leg is written so it
/// cannot pass vacuously — it proves an authored tag was flowing throughout the window in
/// which the unauthored topic produced nothing.
/// </para>
/// </summary>
[Collection(MqttFixtureCollection.Name)]
[Trait("Category", "LiveIntegration")]
public sealed class PlainMqttLiveTests(MqttFixture fixture)
{
/// <summary>OPC UA <c>BadWaitingForInitialData</c> — "no value has ever been observed".</summary>
private const uint BadWaitingForInitialData = 0x80320000u;
/// <summary>OPC UA <c>BadNodeIdUnknown</c> — "that reference is not an authored MQTT tag".</summary>
private const uint BadNodeIdUnknown = 0x80340000u;
/// <summary>OPC UA <c>Good</c>.</summary>
private const uint Good = 0x00000000u;
/// <summary>
/// Bound on every wait for broker traffic. The fixture publisher emits every 2s by default,
/// so 30s is ~15 publish cycles — generous enough to absorb a slow container start, short
/// enough that a genuinely dark subscription fails rather than hangs (the plan's per-op
/// bounded-deadline rule).
/// </summary>
private static readonly TimeSpan MessageTimeout = TimeSpan.FromSeconds(30);
private readonly MqttFixture _fx = fixture;
// -----------------------------------------------------------------------
// TLS + auth
// -----------------------------------------------------------------------
/// <summary>
/// Connects over the TLS listener with real credentials and asserts the session establishes.
/// Pins to the fixture CA when <c>MQTT_FIXTURE_CA_CERT</c> is supplied — so on a fully
/// provisioned rig this is simultaneously the CA-pin happy path — and falls back to
/// <c>AllowUntrustedServerCertificate</c> otherwise so the auth leg still runs.
/// </summary>
[Fact]
public async Task Tls_auth_connect_succeeds()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
await using var connection = new MqttConnection(TlsOptions(), "mqtt-live-tls");
await connection.ConnectAsync(ct);
connection.IsConnected.ShouldBeTrue("a TLS+auth connect against the fixture broker must establish a session");
connection.State.ShouldBe(MqttConnectionState.Connected);
TestContext.Current.SendDiagnosticMessage(
$"TLS+auth connect to {_fx.TlsHost}:{_fx.TlsPort} succeeded "
+ $"(ca pin: {_fx.CaCertificatePath ?? "<none — AllowUntrustedServerCertificate>"}).");
}
/// <summary>
/// The auth negative, and the assertion that gives every other test in this file its meaning:
/// a wrong password must never yield a usable session. The broker runs
/// <c>allow_anonymous false</c>, so if bad credentials produced a session the fixture would be
/// misconfigured and "connected" would prove nothing about auth.
/// <para>
/// <b>LIVE-GATE FINDING (2026-07-24) — the invariant asserted here is deliberately "no
/// session", not "ConnectAsync throws".</b> Against real Mosquitto,
/// <see cref="MqttConnection.ConnectAsync"/> <b>returns normally</b> for a CONNECT the
/// broker rejects as not-authorized: MQTTnet 5 does not throw on an unsuccessful CONNACK,
/// and <c>ConnectCoreAsync</c> then unconditionally publishes
/// <see cref="MqttConnectionState.Connected"/> and starts the reconnect supervisor.
/// Measured at +300 ms with a wrong password: <c>thrown=&lt;none&gt;</c>,
/// <c>IsConnected=False</c>, <c>State=Reconnecting</c> (the broker's socket close had
/// fired <c>DisconnectedAsync</c> by then). Broker side:
/// <c>Client … disconnected, not authorised.</c>
/// </para>
/// <para>
/// That is a driver-side defect, not a fixture one — <c>InitializeAsync</c> reports
/// <c>DriverState.Healthy</c> / <c>HostState.Running</c> off a connect that never
/// authenticated, so a deployment carrying the wrong broker password seals green and the
/// AdminUI "Test connect" probe is expected to agree. Fixing it belongs in the driver
/// (<c>MqttConnection</c> must inspect the connect result / opt into MQTTnet's
/// throw-on-unsuccessful-CONNACK), which is outside this task's file scope; it is reported
/// as a live-gate finding.
/// </para>
/// <para>
/// The test is written to hold either way: it records whether the call threw (as a
/// diagnostic, not an assertion) and asserts only the invariant that must be true under
/// both behaviours — no session is ever established. A driver fixed to throw still passes.
/// </para>
/// </summary>
[Fact]
public async Task Tls_connect_with_wrong_password_never_establishes_a_session()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
var options = TlsOptions() with { Password = _fx.Password + "-definitely-wrong" };
await using var connection = new MqttConnection(options, "mqtt-live-badpass");
Exception? thrown = null;
try
{
await connection.ConnectAsync(ct);
}
catch (Exception ex)
{
thrown = ex;
}
// Poll rather than sample once: the rejection arrives as a broker-initiated socket close a
// few milliseconds after ConnectAsync returns, so a single immediate read could catch the
// optimistic Connected state and pass for the wrong reason. Every sample must show no
// session — a correct password would make IsConnected true within the first sample.
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
while (DateTime.UtcNow < deadline)
{
connection.IsConnected.ShouldBeFalse(
"a CONNECT the broker rejected as not-authorized must never yield a usable session");
await Task.Delay(100, ct);
}
connection.State.ShouldNotBe(
MqttConnectionState.Connected,
"the settled state after a rejected CONNECT must not advertise a healthy session");
TestContext.Current.SendDiagnosticMessage(
$"wrong-password connect: threw={thrown?.GetType().Name ?? "<nothing — see the LIVE-GATE FINDING in this test's docs>"}, "
+ $"IsConnected={connection.IsConnected}, State={connection.State}.");
}
/// <summary>
/// CA-pin happy path: <c>CaCertificatePath</c> set to the fixture's own CA, with the
/// untrusted-certificate escape hatch explicitly OFF, so the handshake can only succeed by
/// chaining the broker's leaf up to the pinned root through
/// <c>MqttConnection.ValidateAgainstPinnedCa</c>.
/// </summary>
[Fact]
public async Task Tls_connect_pinned_to_fixture_ca_succeeds()
{
SkipIfUnconfigured();
if (_fx.CaCertificatePath is null)
{
Assert.Skip(
$"Skipped: set {MqttFixture.CaCertEnvVar} to the fixture CA (Docker/secrets/ca.crt) to run the CA-pin leg.");
}
File.Exists(_fx.CaCertificatePath).ShouldBeTrue(
$"{MqttFixture.CaCertEnvVar}='{_fx.CaCertificatePath}' must point at the fixture CA PEM");
var ct = TestContext.Current.CancellationToken;
var options = TlsOptions() with
{
CaCertificatePath = _fx.CaCertificatePath,
AllowUntrustedServerCertificate = false,
};
await using var connection = new MqttConnection(options, "mqtt-live-capin");
await connection.ConnectAsync(ct);
connection.IsConnected.ShouldBeTrue("the broker leaf must chain up to the pinned fixture CA");
}
/// <summary>
/// The pin's falsifiability control, and the reason the positive above means anything: pin to
/// a CA the broker's certificate was <b>not</b> issued by and assert the handshake fails. A
/// validator that returned <c>true</c> unconditionally — the silent failure mode for this
/// kind of code — would pass the positive test and fail here.
/// <para>
/// The foreign CA is generated in-process rather than taken from the fixture, so this leg
/// needs no extra rig provisioning and cannot accidentally be handed the real CA.
/// </para>
/// </summary>
[Fact]
public async Task Tls_connect_pinned_to_a_foreign_ca_is_rejected()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
var foreignCaPath = Path.Combine(Path.GetTempPath(), $"otopcua-mqtt-foreign-ca-{Guid.NewGuid():N}.pem");
await File.WriteAllTextAsync(foreignCaPath, CreateForeignCaPem(), ct);
try
{
var options = TlsOptions() with
{
CaCertificatePath = foreignCaPath,
AllowUntrustedServerCertificate = false,
// The handshake fails fast; no need to sit on the default 15s budget.
ConnectTimeoutSeconds = 10,
};
await using var connection = new MqttConnection(options, "mqtt-live-foreignca");
await Should.ThrowAsync<Exception>(async () => await connection.ConnectAsync(ct));
connection.IsConnected.ShouldBeFalse(
"a broker certificate that does not chain to the pinned CA must be refused — fail closed");
}
finally
{
File.Delete(foreignCaPath);
}
}
// -----------------------------------------------------------------------
// Plain subscribe / read
// -----------------------------------------------------------------------
/// <summary>
/// The identity assertion. Subscribes an authored tag whose RawPath is deliberately unrelated
/// to its MQTT topic, and asserts the driver publishes the change under the <b>RawPath</b> —
/// the v3 driver wire reference <c>DriverHostActor</c> fans out to the raw + UNS NodeIds. A
/// driver that keyed notifications by topic (or by the TagConfig blob, the retired pre-v3
/// shape) would deliver a value that no node in the address space is listening for, and every
/// "did a value arrive" assertion would still pass.
/// </summary>
[Fact]
public async Task Plain_subscribe_delivers_value_under_rawpath()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
const string rawPath = "Fixture/Mqtt/Line1/Temperature";
var tag = JsonTag(rawPath, _fx.Topic("line1/temperature"), "$.value", DriverDataType.Float64);
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-subscribe");
var recorder = new DataChangeRecorder(driver);
await driver.InitializeAsync(string.Empty, ct);
await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);
var change = await recorder.WaitForAsync(rawPath, MessageTimeout, ct);
change.FullReference.ShouldBe(
rawPath,
"the driver must publish under the tag's RawPath, never its topic or its TagConfig blob");
change.Snapshot.StatusCode.ShouldBe(Good);
change.Snapshot.Value.ShouldBeOfType<double>().ShouldBeInRange(-100d, 1000d);
// ...and the read path serves the same last value under the same key.
var read = await driver.ReadAsync([rawPath], ct);
read.Count.ShouldBe(1);
read[0].StatusCode.ShouldBe(Good);
read[0].Value.ShouldNotBeNull();
}
/// <summary>
/// Retained-message seeding. The fixture publisher emits <c>retained/seed</c> exactly once,
/// at container start, and never republishes it — so a subscriber connecting later can only
/// obtain a value from the broker's retained copy. Asserting the value on a topic that is
/// <i>also</i> republished every couple of seconds would prove nothing: a live publish
/// arriving quickly looks identical to a retained seed.
/// </summary>
[Fact]
public async Task Retained_message_seeds_value_on_subscribe()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
const string rawPath = "Fixture/Mqtt/Retained/Seed";
var tag = JsonTag(rawPath, _fx.Topic("retained/seed"), "$.value", DriverDataType.Float64);
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-retained");
var recorder = new DataChangeRecorder(driver);
await driver.InitializeAsync(string.Empty, ct);
await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);
var change = await recorder.WaitForAsync(rawPath, MessageTimeout, ct);
change.FullReference.ShouldBe(rawPath);
change.Snapshot.StatusCode.ShouldBe(Good);
change.Snapshot.Value.ShouldBeOfType<double>().ShouldBe(
42.5d,
"the retained seed carries the publisher's one-shot value; anything else means the "
+ "subscription was seeded from somewhere other than the broker's retained store");
}
/// <summary>
/// A chatty broker must not auto-provision. Publishes on a topic no tag was authored for and
/// asserts the driver never raises a notification for it.
/// <para>
/// <b>Deliberately falsifiable.</b> A version of this test that authored nothing would
/// pass vacuously — silence proves nothing if the pipeline is dead. So a real tag is
/// authored and subscribed first, and the test asserts that authored tag kept delivering
/// <i>throughout</i> the window in which the unauthored topic was being published. The
/// only way both assertions hold is if the driver was live and chose to ignore the
/// unauthored traffic.
/// </para>
/// </summary>
[Fact]
public async Task Unauthored_topic_is_silent_while_an_authored_tag_keeps_flowing()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
const string authoredRawPath = "Fixture/Mqtt/Line1/Counter";
const string unauthoredRawPath = "Fixture/Mqtt/Line1/NotAuthored";
// Unique per run so a previous run's retained message can never seed it.
var unauthoredTopic = _fx.Topic($"unauthored/{Guid.NewGuid():N}");
var tag = JsonTag(authoredRawPath, _fx.Topic("line1/counter"), "$.value", DriverDataType.Int64);
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-unauthored");
var recorder = new DataChangeRecorder(driver);
await driver.InitializeAsync(string.Empty, ct);
// Subscribing an unauthored reference alongside the authored one: the driver must say
// BadNodeIdUnknown rather than leaving it forever "waiting for initial data" — and must
// never later turn it Good off broker traffic.
await driver.SubscribeAsync([authoredRawPath, unauthoredRawPath], TimeSpan.FromSeconds(1), ct);
// Proof the pipeline is live BEFORE the unauthored traffic starts.
await recorder.WaitForAsync(authoredRawPath, MessageTimeout, ct);
var beforeCount = recorder.CountFor(authoredRawPath);
// Emit unauthored traffic with our own client — the fixture publisher's own noise topic is
// not enough, because a test must be able to point at the exact messages it caused.
await PublishUnauthoredAsync(unauthoredTopic, messages: 5, ct);
// Let the authored tag tick at least once more while the unauthored messages sit delivered.
await recorder.WaitForCountAsync(authoredRawPath, beforeCount + 1, MessageTimeout, ct);
var seen = recorder.Snapshot();
seen.ShouldAllBe(
e => e.FullReference == authoredRawPath,
"the driver must raise notifications ONLY for authored tags; a broker's other topics are none of its business. "
+ $"Saw: {string.Join(", ", seen.Select(e => e.FullReference).Distinct())}");
var read = await driver.ReadAsync([authoredRawPath, unauthoredRawPath], ct);
read[0].StatusCode.ShouldBe(Good, "the authored tag was flowing throughout — this is the falsifiability control");
read[1].StatusCode.ShouldBeOneOf(
BadNodeIdUnknown,
BadWaitingForInitialData);
read[1].Value.ShouldBeNull("an unauthored reference must never acquire a value from broker traffic");
}
// -----------------------------------------------------------------------
// helpers
// -----------------------------------------------------------------------
private void SkipIfUnconfigured()
{
if (_fx.NotConfigured)
{
Assert.Skip(_fx.SkipReason!);
}
}
/// <summary>Driver options for the TLS listener, pinning the fixture CA when one was supplied.</summary>
private MqttDriverOptions TlsOptions(params RawTagEntry[] rawTags) => new()
{
Host = _fx.TlsHost,
Port = _fx.TlsPort,
UseTls = true,
// With no CA to pin, the fixture's self-issued chain cannot validate against the OS trust
// store, so the documented dev escape hatch is the only way the auth leg can run at all.
CaCertificatePath = _fx.CaCertificatePath,
AllowUntrustedServerCertificate = _fx.CaCertificatePath is null,
Username = _fx.Username,
Password = _fx.Password,
ClientId = $"otopcua-live-{Guid.NewGuid():N}",
ConnectTimeoutSeconds = 15,
Mode = MqttMode.Plain,
Plain = new MqttPlainOptions { DefaultQos = 1 },
RawTags = rawTags,
};
/// <summary>Driver options for the plaintext listener — still authenticated.</summary>
private MqttDriverOptions PlainOptions(params RawTagEntry[] rawTags) => new()
{
Host = _fx.PlainHost,
Port = _fx.PlainPort,
UseTls = false,
Username = _fx.Username,
Password = _fx.Password,
ClientId = $"otopcua-live-{Guid.NewGuid():N}",
ConnectTimeoutSeconds = 15,
Mode = MqttMode.Plain,
Plain = new MqttPlainOptions { DefaultQos = 1 },
RawTags = rawTags,
};
private static RawTagEntry JsonTag(string rawPath, string topic, string jsonPath, DriverDataType dataType) =>
new(
rawPath,
$$"""
{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"{{jsonPath}}","dataType":"{{dataType}}","qos":1,"retainSeed":true}
""",
WriteIdempotent: false);
/// <summary>
/// Publishes on a topic no tag is authored for, with a raw MQTTnet client rather than through
/// the driver (the driver has no write path in P1, and the point is traffic the driver did
/// not originate).
/// </summary>
private async Task PublishUnauthoredAsync(string topic, int messages, CancellationToken ct)
{
using var client = new MqttClientFactory().CreateMqttClient();
var options = new MqttClientOptionsBuilder()
.WithTcpServer(_fx.PlainHost, _fx.PlainPort)
.WithCredentials(_fx.Username, _fx.Password)
.WithClientId($"otopcua-live-noise-{Guid.NewGuid():N}")
.WithTimeout(TimeSpan.FromSeconds(10))
.Build();
await client.ConnectAsync(options, ct);
try
{
for (var i = 0; i < messages; i++)
{
await client.PublishStringAsync(
topic,
$$"""{"value":{{i}},"source":"unauthored-live-test"}""",
MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce,
retain: false,
ct);
}
}
finally
{
await client.DisconnectAsync(cancellationToken: ct);
}
}
/// <summary>
/// A self-signed CA the fixture broker's certificate was definitively NOT issued by, as PEM.
/// Generated per call so nothing about it can be shared with — or mistaken for — the real
/// fixture CA.
/// </summary>
private static string CreateForeignCaPem()
{
using var key = RSA.Create(2048);
var request = new CertificateRequest(
"CN=OtOpcUa MQTT foreign CA (negative control)",
key,
HashAlgorithmName.SHA256,
RSASignaturePadding.Pkcs1);
request.CertificateExtensions.Add(new X509BasicConstraintsExtension(true, false, 0, true));
request.CertificateExtensions.Add(
new X509KeyUsageExtension(X509KeyUsageFlags.KeyCertSign | X509KeyUsageFlags.CrlSign, true));
using var certificate = request.CreateSelfSigned(
DateTimeOffset.UtcNow.AddDays(-1),
DateTimeOffset.UtcNow.AddDays(1));
var builder = new StringBuilder();
builder.AppendLine("-----BEGIN CERTIFICATE-----");
builder.AppendLine(Convert.ToBase64String(certificate.RawData, Base64FormattingOptions.InsertLineBreaks));
builder.AppendLine("-----END CERTIFICATE-----");
return builder.ToString();
}
/// <summary>
/// Records every <c>OnDataChange</c> the driver raises and lets a test await a specific
/// RawPath. Attached in the constructor — <b>before</b> the MQTT subscribe — because the
/// retained seed can arrive inside the SUBSCRIBE round-trip, and a handler attached
/// afterwards would miss exactly the message the retained-seed test is about.
/// </summary>
private sealed class DataChangeRecorder
{
private readonly List<DataChangeEventArgs> _events = [];
private readonly Lock _gate = new();
private readonly List<Waiter> _waiters = [];
public DataChangeRecorder(MqttDriver driver) => driver.OnDataChange += OnDataChange;
/// <summary>A point-in-time copy of everything recorded so far.</summary>
/// <returns>The recorded events.</returns>
public IReadOnlyList<DataChangeEventArgs> Snapshot()
{
lock (_gate)
{
return [.. _events];
}
}
/// <summary>How many notifications have been recorded for one RawPath.</summary>
/// <param name="rawPath">The RawPath to count.</param>
/// <returns>The count.</returns>
public int CountFor(string rawPath)
{
lock (_gate)
{
return _events.Count(e => e.FullReference == rawPath);
}
}
/// <summary>Awaits the first notification for <paramref name="rawPath"/>.</summary>
/// <param name="rawPath">The RawPath to await.</param>
/// <param name="timeout">Bound on the wait.</param>
/// <param name="cancellationToken">Caller cancellation.</param>
/// <returns>The first matching notification.</returns>
public Task<DataChangeEventArgs> WaitForAsync(
string rawPath,
TimeSpan timeout,
CancellationToken cancellationToken)
=> WaitForCountAsync(rawPath, count: 1, timeout, cancellationToken);
/// <summary>
/// Awaits the <paramref name="count"/>-th notification for <paramref name="rawPath"/>.
/// Pre-scans what has already been recorded, so a caller that registers the wait after
/// the message arrived is served immediately rather than timing out.
/// </summary>
/// <param name="rawPath">The RawPath to await.</param>
/// <param name="count">How many notifications must have been seen.</param>
/// <param name="timeout">Bound on the wait.</param>
/// <param name="cancellationToken">Caller cancellation.</param>
/// <returns>The <paramref name="count"/>-th matching notification.</returns>
public async Task<DataChangeEventArgs> WaitForCountAsync(
string rawPath,
int count,
TimeSpan timeout,
CancellationToken cancellationToken)
{
Waiter waiter;
lock (_gate)
{
var matches = _events.Where(e => e.FullReference == rawPath).ToList();
if (matches.Count >= count)
{
return matches[count - 1];
}
waiter = new Waiter(rawPath, count);
_waiters.Add(waiter);
}
try
{
return await waiter.Completion.Task.WaitAsync(timeout, cancellationToken);
}
catch (TimeoutException)
{
throw new TimeoutException(
$"No notification #{count} for RawPath '{rawPath}' within {timeout.TotalSeconds:0}s. "
+ $"Recorded so far: {DescribeRecorded()}. Is the fixture publisher container running "
+ "(`docker logs otopcua-mqtt-publisher`), and does MQTT_FIXTURE_TOPIC_PREFIX match it?");
}
finally
{
lock (_gate)
{
_waiters.Remove(waiter);
}
}
}
private string DescribeRecorded()
{
var snapshot = Snapshot();
return snapshot.Count == 0
? "(nothing)"
: string.Join(", ", snapshot.GroupBy(e => e.FullReference).Select(g => $"{g.Key}×{g.Count()}"));
}
private void OnDataChange(object? sender, DataChangeEventArgs args)
{
List<Waiter>? satisfied = null;
lock (_gate)
{
_events.Add(args);
foreach (var waiter in _waiters)
{
if (waiter.RawPath != args.FullReference) continue;
if (_events.Count(e => e.FullReference == waiter.RawPath) < waiter.Count) continue;
(satisfied ??= []).Add(waiter);
}
}
// Completed OUTSIDE the lock, and with RunContinuationsAsynchronously below, so a test
// continuation never runs on MQTTnet's dispatcher thread while holding this lock —
// that would stall delivery for every other subscription.
if (satisfied is null) return;
foreach (var waiter in satisfied)
{
waiter.Completion.TrySetResult(args);
}
}
private sealed record Waiter(string RawPath, int Count)
{
public TaskCompletionSource<DataChangeEventArgs> Completion { get; } =
new(TaskCreationOptions.RunContinuationsAsynchronously);
}
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj
+49
View File
@@ -0,0 +1,49 @@
<Project Sdk="Microsoft.NET.Sdk">
<!-- OTOPCUA0001 (arch-review 07/C-1): this suite invokes driver capability methods DIRECTLY to
exercise wire-level behavior against a real broker — the analyzer's documented intentional
case ("move the suppression into a NoWarn for the test project"). Not a resilience-dispatch
gap. Same suppression as the S7 / AB CIP integration suites. -->
<PropertyGroup>
<NoWarn>$(NoWarn);OTOPCUA0001</NoWarn>
</PropertyGroup>
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<IsPackable>false</IsPackable>
<IsTestProject>true</IsTestProject>
<RootNamespace>ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests</RootNamespace>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="xunit.v3"/>
<PackageReference Include="Shouldly"/>
<PackageReference Include="Microsoft.NET.Test.Sdk"/>
<PackageReference Include="xunit.runner.visualstudio">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<!--
Direct reference (not merely transitive through the driver) because the tests
publish into the fixture broker with their own raw MQTTnet client — the
unauthored-topic test has to emit traffic the DRIVER never emits.
-->
<PackageReference Include="MQTTnet"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/>
</ItemGroup>
<!--
Docker/ is DELIBERATELY not copied to the output directory (the sibling Modbus /
S7 suites copy theirs because their tests read simulator profile JSONs at
runtime; nothing here does). Copying it would also drag Docker/secrets/ — the
generated broker password hash and the CA private key — into bin/, i.e. out of
the one directory .gitignore protects.
-->
</Project>