MQTT · IIoT Sensor Simulator · Schemathesis

IIoT-Sensor-Simulator mit MQTT-Protokoll, FastAPI REST-Schnittstelle und automatisierten API-Tests mit Schemathesis — vollständig lauffähig mit Docker und GitHub Actions CI.

GitHub Repository

Kontext & Motivation

Im industriellen IoT müssen Sensoren kontinuierlich Daten über MQTT publizieren und diese über REST APIs zugänglich machen. Die Validierung solcher Schnittstellen gemäß OpenAPI-Spezifikationen ist eine zentrale Aufgabe in modernen IIoT-Projekten.

Ziel war es, einen vollständigen IIoT-Workflow zu implementieren: Sensorsimulation via MQTT, REST API mit automatischer OpenAPI 3.x Dokumentation und automatisierte API-Tests mit Schemathesis — inklusive echtem Bug-Finding und Fix.

Anwendungsbereich

IIoT, Industrie 4.0, Sensor-Monitoring, Predictive Maintenance

Protokoll

MQTT QoS 0/1/2 — Eclipse Mosquitto Broker via Docker

Werkzeuge

Python, FastAPI, Paho-MQTT, Schemathesis, Docker, GitHub Actions

Testing

Schemathesis — 2170 Tests generiert, OpenAPI 3.1 vollständig validiert

1 — Architektur

Die Pipeline besteht aus drei Schichten: Sensorsimulation, Datenvermittlung via MQTT und REST API — alles durch automatisierte Tests in GitHub Actions abgesichert.

MQTT IIoT Architektur Drei-Schichten Architektur: Sensor Simulator, MQTT Broker, FastAPI REST API, getestet durch Schemathesis und pytest via GitHub Actions. Sensor Simulator Python + Paho-MQTT 3 sensors · 5s interval temp · humidity · pressure QoS 1 · JSON payload Publish QoS 1 MQTT Broker Eclipse Mosquitto 2.0 Docker · Port 1883 Topic: sensors/{id}/{m} allow_anonymous true Subscribe sensors/# FastAPI REST API · Port 8000 OpenAPI 3.1 auto-gen GET /sensors · /health In-memory sensor store sensors/{id}/temperature sensors/{id}/humidity · pressure QoS 0 — status (best effort) QoS 1 — measurements (AT LEAST ONCE) GET /sensors → list GET /sensors/{id} → 200 / 404 GET /health → sensors_online CI/CD — GitHub Actions (on every push) Schemathesis API Fuzzing 2170 tests generiert Coverage · Fuzzing · Stateful pytest MQTT QoS 0/1/2 tests Topic structure validation Payload schema checks GitHub Actions Lint (flake8) Test (pytest) Fuzz (schemathesis) Ergebnisse 2170 Tests · No issues found · Coverage ✅ · Fuzzing ✅ · Stateful 758 ✅ · Bug found and fixed (404 undokumentiert)

MQTT Topic Struktur

TopicQoSPayload
sensors/{id}/temperature1{"value": 23.5, "timestamp": "...", "sensor_id": "..."}
sensors/{id}/humidity1{"value": 65.2, "timestamp": "...", "sensor_id": "..."}
sensors/{id}/pressure1{"value": 1013.2, "timestamp": "...", "sensor_id": "..."}
sensors/{id}/status0{"value": "online", ...}
sensors/{id}/full1Kompletter SensorReading Datensatz

2 — MQTT Sensor Simulator

Der Simulator publiziert alle 5 Sekunden realistische Sensordaten für 3 Sensoren — typisch für industrielle Umgebungssensoren.

MQTT Sensor Simulator läuft 
sensors/sensor-001/temperature | value=31.33 | QoS=1
sensors/sensor-001/humidity    | value=55.72 | QoS=1
sensors/sensor-001/pressure    | value=1015.46 | QoS=1
sensors/sensor-002/temperature | value=18.65 | QoS=1
sensors/sensor-003/temperature | value=29.68 | QoS=1
QoS 1 garantiert die Zustellung jeder Nachricht mindestens einmal — Standard für industrielle Sensordaten wo Datenverlust nicht akzeptabel ist.

3 — FastAPI REST API & OpenAPI 3.1

FastAPI generiert automatisch eine vollständige OpenAPI 3.1 Dokumentation aus dem Python-Code — kein manuelles Schreiben der Spec notwendig.

OpenAPI 3.1 Dokumentation

Endpoints

MethodEndpointResponseBeschreibung
GET/health200Health Check + sensors_online count
GET/sensors200Liste aller bekannten Sensoren
GET/sensors/{id}200 / 404Vollständiger Datensatz eines Sensors
GET/sensors/{id}/measurements200 / 404Einzelne Messwerte eines Sensors

Sensor Liste

Sensors List

Sensor Detail

Sensor Detail

Measurements Endpoint

Sensor Measurements

4 — Automatisierte Tests mit Schemathesis

Schemathesis generiert automatisch Testfälle aus der OpenAPI-Spezifikation und testet die API mit validen, invaliden und Edge-Case-Eingaben — API Fuzzing.

Erster Durchlauf — Bug gefunden

Schemathesis erste Ergebnisse
⚠️ Schemathesis sendete Unicode-Zeichen als sensor_id und erhielt 404 — aber die Spec dokumentierte nur 200 oder 422. Undokumentierter Status-Code = Spec-Fehler → sofort behoben.

Fix — 404 in OpenAPI Spec dokumentiert

@app.get("/sensors/{sensor_id}",
    response_model=SensorReading,
    tags=["Sensors"],
    responses={404: {"description": "Sensor not found"}}
)
def get_sensor(sensor_id: str): ...

Zweiter Durchlauf — No issues found

Schemathesis alle Tests bestehen
✅ No issues found — 2170 Tests generiert, alle bestehen. Coverage, Fuzzing und Stateful Tests vollständig grün.
2170
Tests generiert
4/4
Endpoints getestet
758
Stateful Scenarios
0
Failures (nach Fix)

5 — Zusammenfassung

KomponenteToolStatus
MQTT BrokerEclipse Mosquitto 2.0 (Docker)✅
Sensor SimulatorPython + Paho-MQTT✅
REST APIFastAPI + Uvicorn✅
OpenAPI SpecOpenAPI 3.1 (auto-generiert)✅
API FuzzingSchemathesis 4.21✅
Bug Found & Fixed404 response dokumentiert✅
CI/CDGitHub Actions✅
ContainerisierungDocker + Docker Compose✅

Skills

MQTTQoS LevelsPython FastAPIOpenAPI 3.xSchemathesis API FuzzingREST API TestingDocker GitHub ActionsIIoTPaho-MQTT JSON Schema

6 — Ausblick