195 lines
7.1 KiB
Python
195 lines
7.1 KiB
Python
"""External REST API test server for ScadaLink test infrastructure."""
|
|
|
|
import os
|
|
import time
|
|
import uuid
|
|
|
|
from flask import Flask, jsonify, request
|
|
|
|
app = Flask(__name__)
|
|
|
|
START_TIME = time.time()
|
|
API_KEY = "scadalink-test-key-1"
|
|
NO_AUTH = os.environ.get("API_NO_AUTH", "0") == "1"
|
|
|
|
|
|
@app.before_request
|
|
def check_auth():
|
|
if NO_AUTH:
|
|
return None
|
|
if request.path == "/api/Ping" and request.method == "GET":
|
|
return None # health check is unauthenticated
|
|
|
|
api_key = request.headers.get("X-API-Key")
|
|
if api_key == API_KEY:
|
|
return None
|
|
|
|
auth = request.authorization
|
|
if auth and auth.type == "basic":
|
|
return None # accept any basic auth credentials
|
|
|
|
return jsonify({"error": "Unauthorized", "message": "Provide X-API-Key header or Basic auth"}), 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Simple methods
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/Ping", methods=["GET"])
|
|
def ping():
|
|
return jsonify({"pong": True})
|
|
|
|
|
|
@app.route("/api/Add", methods=["POST"])
|
|
def add():
|
|
data = request.get_json(force=True)
|
|
a = data.get("a", 0)
|
|
b = data.get("b", 0)
|
|
return jsonify({"result": a + b})
|
|
|
|
|
|
@app.route("/api/Multiply", methods=["POST"])
|
|
def multiply():
|
|
data = request.get_json(force=True)
|
|
a = data.get("a", 0)
|
|
b = data.get("b", 0)
|
|
return jsonify({"result": a * b})
|
|
|
|
|
|
@app.route("/api/Echo", methods=["POST"])
|
|
def echo():
|
|
data = request.get_json(force=True)
|
|
return jsonify({"message": data.get("message", "")})
|
|
|
|
|
|
@app.route("/api/GetStatus", methods=["POST"])
|
|
def get_status():
|
|
uptime = round(time.time() - START_TIME, 1)
|
|
return jsonify({"status": "running", "uptime": uptime})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Complex methods (nested objects + lists)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/GetProductionReport", methods=["POST"])
|
|
def get_production_report():
|
|
data = request.get_json(force=True)
|
|
site_id = data.get("siteId", "Unknown")
|
|
return jsonify({
|
|
"siteName": f"Site {site_id}",
|
|
"totalUnits": 14250,
|
|
"lines": [
|
|
{"lineName": "Line-1", "units": 8200, "efficiency": 92.5},
|
|
{"lineName": "Line-2", "units": 6050, "efficiency": 88.1},
|
|
],
|
|
})
|
|
|
|
|
|
@app.route("/api/GetRecipe", methods=["POST"])
|
|
def get_recipe():
|
|
data = request.get_json(force=True)
|
|
recipe_id = data.get("recipeId", "R-000")
|
|
return jsonify({
|
|
"recipeId": recipe_id,
|
|
"name": "Standard Mix",
|
|
"version": 3,
|
|
"ingredients": [
|
|
{"name": "Material-A", "quantity": 45.0, "unit": "kg"},
|
|
{"name": "Material-B", "quantity": 12.5, "unit": "L"},
|
|
],
|
|
})
|
|
|
|
|
|
@app.route("/api/SubmitBatch", methods=["POST"])
|
|
def submit_batch():
|
|
data = request.get_json(force=True)
|
|
batch_id = f"BATCH-{uuid.uuid4().hex[:8].upper()}"
|
|
item_count = len(data.get("items", []))
|
|
return jsonify({
|
|
"batchId": batch_id,
|
|
"accepted": True,
|
|
"itemCount": item_count,
|
|
})
|
|
|
|
|
|
@app.route("/api/GetEquipmentStatus", methods=["POST"])
|
|
def get_equipment_status():
|
|
data = request.get_json(force=True)
|
|
site_id = data.get("siteId", "Unknown")
|
|
return jsonify({
|
|
"siteId": site_id,
|
|
"equipment": [
|
|
{
|
|
"equipmentId": "PUMP-001",
|
|
"name": "Feed Pump A",
|
|
"status": {"state": "running", "health": 98.5, "lastMaintenance": "2026-02-15"},
|
|
},
|
|
{
|
|
"equipmentId": "TANK-001",
|
|
"name": "Mix Tank 1",
|
|
"status": {"state": "idle", "health": 100.0, "lastMaintenance": "2026-03-01"},
|
|
},
|
|
{
|
|
"equipmentId": "CONV-001",
|
|
"name": "Conveyor B",
|
|
"status": {"state": "alarm", "health": 72.3, "lastMaintenance": "2026-01-20"},
|
|
},
|
|
],
|
|
})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error simulation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/SimulateTimeout", methods=["POST"])
|
|
def simulate_timeout():
|
|
data = request.get_json(force=True)
|
|
seconds = min(data.get("seconds", 5), 60) # cap at 60s
|
|
time.sleep(seconds)
|
|
return jsonify({"delayed": True, "seconds": seconds})
|
|
|
|
|
|
@app.route("/api/SimulateError", methods=["POST"])
|
|
def simulate_error():
|
|
data = request.get_json(force=True)
|
|
code = data.get("code", 500)
|
|
if code < 400 or code > 599:
|
|
code = 500
|
|
return jsonify({"error": True, "code": code}), code
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Method discovery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/methods", methods=["GET"])
|
|
def list_methods():
|
|
"""List all available API methods."""
|
|
methods = [
|
|
{"method": "Ping", "httpMethod": "GET", "path": "/api/Ping", "description": "Health check"},
|
|
{"method": "Add", "httpMethod": "POST", "path": "/api/Add", "params": {"a": "number", "b": "number"}, "description": "Add two numbers"},
|
|
{"method": "Multiply", "httpMethod": "POST", "path": "/api/Multiply", "params": {"a": "number", "b": "number"}, "description": "Multiply two numbers"},
|
|
{"method": "Echo", "httpMethod": "POST", "path": "/api/Echo", "params": {"message": "string"}, "description": "Echo back input"},
|
|
{"method": "GetStatus", "httpMethod": "POST", "path": "/api/GetStatus", "params": {}, "description": "Server status and uptime"},
|
|
{"method": "GetProductionReport", "httpMethod": "POST", "path": "/api/GetProductionReport", "params": {"siteId": "string", "startDate": "string", "endDate": "string"}, "description": "Production report with line details"},
|
|
{"method": "GetRecipe", "httpMethod": "POST", "path": "/api/GetRecipe", "params": {"recipeId": "string"}, "description": "Recipe with ingredients list"},
|
|
{"method": "SubmitBatch", "httpMethod": "POST", "path": "/api/SubmitBatch", "params": {"siteId": "string", "recipeId": "string", "items": "list"}, "description": "Submit a batch with items"},
|
|
{"method": "GetEquipmentStatus", "httpMethod": "POST", "path": "/api/GetEquipmentStatus", "params": {"siteId": "string"}, "description": "Equipment status with nested objects"},
|
|
{"method": "SimulateTimeout", "httpMethod": "POST", "path": "/api/SimulateTimeout", "params": {"seconds": "number"}, "description": "Delay response by N seconds"},
|
|
{"method": "SimulateError", "httpMethod": "POST", "path": "/api/SimulateError", "params": {"code": "number"}, "description": "Return specified HTTP error code"},
|
|
]
|
|
return jsonify({"methods": methods})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
port = int(os.environ.get("PORT", 5200))
|
|
no_auth_flag = "--no-auth" in __import__("sys").argv
|
|
if no_auth_flag:
|
|
NO_AUTH = True
|
|
if NO_AUTH:
|
|
print(" * Auth disabled (--no-auth or API_NO_AUTH=1)")
|
|
print(f" * API key: {API_KEY}")
|
|
app.run(host="0.0.0.0", port=port)
|