Pytest basic implementation (#73)

* Playing around with pytest * Playing around with pytest * Update pytest.yml * Update pytest.yml * Update pytest.yml * Update pytest.yml * Update pytest.yml * Update pytest.yml * Update pytest.yml install pytest in github actions * Update pytest.yml * Testing URLs for FUXA, OpenPLC, and physical & added requirements * Added testing for opcua and fixed docker-compose port exposure * Added status badge * Test if opc-ua is up and running --------- Co-authored-by: Matthias Niedermaier <matthias.niedermaier@googlemail.com>
mniedermaier · Oct 22, 2024 · f6058d5 · f6058d5
1 parent db01e39
commit f6058d5
Show file tree

Hide file tree

Showing 7 changed files with 199 additions and 7 deletions.
diff --git a/.devcontainer/virtual/docker-compose.yml b/.devcontainer/virtual/docker-compose.yml
@@ -30,6 +30,8 @@ services:
       context: ../../software/opcua
       dockerfile: Dockerfile
     restart: always
+    ports:
+      - 4840:4840
     depends_on:
     - openplc
 

diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
@@ -0,0 +1,47 @@
+name: pytest
+
+on:
+  push:
+  schedule:
+  - cron: "0 0 * * *"
+
+jobs:
+  docker:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+      with:
+          submodules: recursive
+
+    - name: Setup Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.12' 
+
+    - name: Setup docker compose
+      uses: KengoTODA/actions-setup-docker-compose@main
+      env:
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+    - name: Run prepre env
+      run: |
+         chmod +x ./.devcontainer/prepare-env.sh
+         .devcontainer/prepare-env.sh
+      shell: bash
+
+    - name: Start containers
+      run: docker-compose -f "docker-compose.yml" up -d --build
+      working-directory: ./.devcontainer/virtual
+
+
+    - name: Pytest Modbus
+      run: |
+        pip install --no-cache-dir -r tests/requirements.txt
+        pytest tests/test_modbus.py
+
+    - name: Stop containers
+      if: always()
+      run: docker-compose -f "docker-compose.yml" down
+      working-directory: ./.devcontainer/virtual
diff --git a/.gitignore b/.gitignore
@@ -76,6 +76,27 @@ fp-info-cache
 *.lck
 *auto_saved_files*
 
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+
 # do not push database
 historian.sqlite
 .env

diff --git a/README.md b/README.md
@@ -13,7 +13,7 @@
 [![flawFinder C](https://github.com/mniedermaier/CybICS/actions/workflows/flawfinder.yml/badge.svg)](https://github.com/mniedermaier/CybICS/actions/workflows/flawfinder.yml)
 [![TruffleHog](https://github.com/mniedermaier/CybICS/actions/workflows/trufflehog.yaml/badge.svg)](https://github.com/mniedermaier/CybICS/actions/workflows/trufflehog.yaml)
 [![devContainer](https://github.com/mniedermaier/CybICS/actions/workflows/devContainer.yml/badge.svg)](https://github.com/mniedermaier/CybICS/actions/workflows/devContainer.yml)
-
+[![pytest](https://github.com/mniedermaier/CybICS/actions/workflows/pytest.yml/badge.svg)](https://github.com/mniedermaier/CybICS/actions/workflows/pytest.yml)
 </div>
 
 ---

diff --git a/software/opcua/opcua.py b/software/opcua/opcua.py
@@ -4,6 +4,7 @@
 import logging
 import socket
 import user_manager
+import time
 
 from pathlib import Path
 from asyncua import Server, ua
@@ -19,6 +20,8 @@
 
 async def main():
     _logger = logging.getLogger(__name__)
+    _logger.info("Wait 5s that openplc can start")
+    time.sleep(5) # wait that openplc is up and running
 
     # Define Path for self-signed server certificate used by secure channel
     cert_base = Path(__file__).parent
@@ -115,16 +118,20 @@ async def main():
     )
     _logger.info("Starting server!")
     async with server:
+        _logger.info("Starting while True")
         while True:
             # read GST and HPT to the OpenPLC
             # Check if flag var is set and display flag
             _logger.info("Reading from modbus")
-            gst = client.read_holding_registers(1124)
-            hpt = client.read_holding_registers(1126)
-            systemSen = client.read_holding_registers(2)
-            boSen = client.read_holding_registers(3)
-            stop = client.read_holding_registers(1129)
-            manual = client.read_holding_registers(1131)
+            try:
+                gst = client.read_holding_registers(1124)
+                hpt = client.read_holding_registers(1126)
+                systemSen = client.read_holding_registers(2)
+                boSen = client.read_holding_registers(3)
+                stop = client.read_holding_registers(1129)
+                manual = client.read_holding_registers(1131)
+            except Exception as e:
+              logging.error("Read from OpenPLC failed - " + str(e))
             await gstvar.write_value(ua.UInt16(gst.registers[0]))
             await hptvar.write_value(ua.UInt16(hpt.registers[0]))
             await systemSenvar.write_value(ua.UInt16(systemSen.registers[0]))

diff --git a/tests/requirements.txt b/tests/requirements.txt
@@ -0,0 +1,7 @@
+pytest
+pytest-cov
+pytest-asyncio
+pymodbus
+asyncio
+asyncua
+requests
diff --git a/tests/test_modbus.py b/tests/test_modbus.py
@@ -0,0 +1,108 @@
+import pytest
+import requests
+import pytest_asyncio
+from pymodbus.client import ModbusTcpClient
+from pymodbus.exceptions import ConnectionException
+from asyncua import Client, ua
+
+# Define the IP and Ports
+SERVER_IP = '127.0.0.1'   # Replace with your server IP
+MODBUS_SERVER_PORT = 502  # Replace with your Modbus server Port (default is 502)
+
+# OPC UA server URL (with the correct address)
+OPCUA_SERVER_URL = "opc.tcp://"+SERVER_IP+":4840"
+USERNAME = "user1"
+PASSWORD = "test"
+
+# List of URLs to check
+URLS = [
+    "http://"+SERVER_IP+"",
+    "http://"+SERVER_IP+":1881",
+    "http://"+SERVER_IP+":8080"
+]
+
+@pytest.mark.parametrize("url", URLS)
+def test_website_is_up(url):
+    try:
+        response = requests.get(url, timeout=5)  # Set a timeout of 5 seconds
+        assert response.status_code == 200, f"Website {url} is not up. Status code: {response.status_code}"
+    except requests.exceptions.RequestException as e:
+        pytest.fail(f"Failed to reach {url}: {e}")
+
+@pytest.fixture
+def modbus_client():
+    """
+    Fixture to create and clean up the Modbus TCP client.
+    """
+    client = ModbusTcpClient(SERVER_IP, port=MODBUS_SERVER_PORT)
+    yield client
+    client.close()
+
+def test_modbus_connection(modbus_client):
+    """
+    Test to check if the Modbus client can successfully connect to the server.
+    """
+    assert modbus_client.connect(), "Failed to connect to the Modbus server"
+
+def test_read_holding_register(modbus_client):
+    """
+    Test to read a specific holding register.
+    Assuming register 1 exists and has a valid value.
+    """
+    # Address of the register (Modbus address is often 0-based)
+    register_address = 1
+    result = modbus_client.read_holding_registers(register_address, 1)
+
+    # Check if we received a valid response
+    assert result is not None, "No response from the Modbus server"
+    assert not result.isError(), f"Modbus error: {result}"
+    assert result.registers[0] is not None, "Invalid register value"
+
+def test_write_single_register(modbus_client):
+    """
+    Test to write a value to a single holding register.
+    """
+    register_address = 1  # Example register address
+    value_to_write = 123  # Example value to write
+
+    # Write the value to the register
+    result = modbus_client.write_register(register_address, value_to_write)
+
+    # Check if write was successful
+    assert result is not None, "No response from the Modbus server"
+    assert not result.isError(), f"Modbus error: {result}"
+
+    # Verify that the value was written correctly by reading it back
+    read_result = modbus_client.read_holding_registers(register_address, 1)
+    assert read_result.registers[0] == value_to_write, f"Expected {value_to_write}, got {read_result.registers[0]}"
+
+@pytest.mark.asyncio
+async def test_opcua_server_running():
+    """
+    Test to check if the OPC UA server is running and accessible.
+    """
+    client = Client(OPCUA_SERVER_URL)
+
+    # Set username and password for the asyncua client
+    client.set_user(USERNAME)
+    client.set_password(PASSWORD)
+
+    try:
+        # Increase the timeout for connection attempts
+        client.timeout = 10
+
+        # Try to connect to the server
+        await client.connect()
+        print("Successfully connected to the OPC UA server.")
+
+        # Optionally, check server status
+        server_status_node = client.get_node(ua.ObjectIds.Server_ServerStatus)
+        server_status = await server_status_node.read_value()
+        assert server_status is not None, "Server status is None."
+
+        print(f"Server Status: {server_status}")
+
+    except Exception as e:
+        pytest.fail(f"Failed to connect to the OPC UA server: {e}")
+    finally:
+        await client.disconnect()  # Ensure proper cleanup