import os

import asyncio

from unittest.mock import MagicMock

import pytest

import socketio

import requests

from openscada_lite.common.utils import SecurityUtils

from openscada_lite.modules.security.service import SecurityService

from openscada_lite.common.bus.event_bus import EventBus

from openscada_lite.modules.command.service import CommandService

from openscada_lite.common.config.config import Config

from openscada_lite.common.models.dtos import CommandFeedbackMsg, SendCommandMsg

from openscada_lite.modules.command.model import CommandModel

@pytest.fixture(autouse=True)

def set_config_env(monkeypatch):

monkeypatch.setenv("SCADA_CONFIG_PATH", "tests/config/system_config.json")

SERVER_URL = "http://localhost:5000"

@pytest.fixture(autouse=True)

def reset_event_bus(monkeypatch):

# Reset the singleton before each test

monkeypatch.setattr(EventBus, "_instance", None)

@pytest.fixture(autouse=True)

def allow_all_security(monkeypatch):

mock_security_service = MagicMock()

mock_security_service.is_allowed.return_value = True

monkeypatch.setattr(SecurityService, "_instance", mock_security_service)

@pytest.fixture(scope="session", autouse=True)

def run_server():

import subprocess

import time

import socket

from pathlib import Path

# Ensure SCADA_CONFIG_PATH is set

cfg_file = Path(__file__).parent / "config"

os.environ["SCADA_CONFIG_PATH"] = str(cfg_file.resolve())

# Start Uvicorn in a subprocess

process = subprocess.Popen(

[

"uvicorn",

"openscada_lite.app:asgi_app",

"--host",

"127.0.0.1",

"--port",

"5000",

],

env=os.environ.copy(), # Pass the current environment variables to the subprocess

)

# Wait for server readiness (avoid race conditions)

start = time.time()

while True:

try:

with socket.create_connection(("127.0.0.1", 5000), timeout=0.5):

break

except OSError:

if time.time() - start > 10:

raise RuntimeError("Server on 127.0.0.1:5000 did not start within 10s")

time.sleep(0.2)

yield

process.terminate()

process.wait()

def immediate_call(func, *args, **kwargs):

import asyncio

if asyncio.iscoroutinefunction(func):

asyncio.run(func(*args, **kwargs))

else:

func(*args, **kwargs)

@pytest.mark.asyncio

async def test_command_live_feed_and_feedback():

sio = socketio.Client()

received_initial = []

received_feedback = []

@sio.on("command_initial_state")

def on_initial_state(data):

received_initial.append(data)

@sio.on("command_commandfeedbackmsg")

def on_command_feedback(data):

received_feedback.append(data)

# Connect to the server

sio.connect(SERVER_URL)

sio.emit("command_subscribe_live_feed")

await asyncio.sleep(1) # Wait for initial state

token = SecurityUtils.create_jwt("admin", "test_group")

# Send a command

test_command = SendCommandMsg(

command_id="testcmd1",

datapoint_identifier="WaterTank@TANK",

value=42,

)

cookies = {"jwt": token}

print("Sending command:", test_command.to_dict())

response = requests.post(

f"{SERVER_URL}/command/sendcommandmsg",

json=test_command.to_dict(),

cookies=cookies,

)

assert response.status_code == 200

# Wait for feedback to be received

await asyncio.sleep(1.1) # Slightly longer than the server's processing time

# Assert that feedback was received

assert received_feedback, "No command feedback received after sending command"

# Flatten the received feedback (in case of batched messages)

all_feedback = [item for batch in received_feedback for item in batch]

# Check if the expected feedback is in the batch

feedback = next((f for f in all_feedback if f["command_id"] == "testcmd1"), None)

assert feedback is not None, "Expected feedback not found in received feedback"

assert feedback["datapoint_identifier"] == "WaterTank@TANK"

assert feedback["value"] == 42

assert feedback["feedback"] == "NOK-driver-offline"

# Disconnect the client

sio.disconnect()

def test_command_model_initial_load(monkeypatch):

# Mock Config.get_instance().get_allowed_command_identifiers()

allowed = ["WaterTank@CMD1", "AuxServer@CMD2"]

class DummyConfig:

def get_allowed_command_identifiers(self):

return allowed

monkeypatch.setattr(Config, "get_instance", lambda: DummyConfig())

model = CommandModel()

# The model should have all allowed commands in its _store

for cmd_id in allowed:

assert cmd_id in model._store

feedback = model._store[cmd_id]

assert feedback.datapoint_identifier == cmd_id

assert feedback.feedback is None

@pytest.mark.asyncio

async def test_command_service_handle_bus_message(monkeypatch):

allowed = ["WaterTank@CMD1"]

class DummyConfig:

def get_allowed_command_identifiers(self):

return allowed

monkeypatch.setattr(Config, "get_instance", lambda: DummyConfig())

bus = EventBus.get_instance()

model = CommandModel()

service = CommandService(bus, model, controller=None)

msg = CommandFeedbackMsg(

command_id="cmd123",

datapoint_identifier="WaterTank@CMD1",

value=42,

feedback="OK",

timestamp=None,

)

await service.handle_bus_message(msg)

stored = model._store.get("WaterTank@CMD1")

assert stored is not None

assert stored.command_id == "cmd123"

assert stored.value == 42

assert stored.feedback == "OK"