Add automated smoke tests

.github/workflows/test.yml

@@ -0,0 +1,104 @@
+name: Test
+
+on:
+  workflow_run:
+     workflows: [ Build ]
+     types:
+        - completed
+
+jobs:
+  build:
+    name: Test
+    runs-on: ubuntu-latest
+    
+    steps:
+      - name: Install VMs
+        run: |
+          set -x
+          sudo snap install multipass --edge # Edge is necessary to get 1.15 dev because of networks support
+          sudo chmod a+w /var/snap/multipass/common/multipass_socket # Fix 'multipass socket access denied'
+          multipass find
+
+          # Set up some bridges for the VMs
+          sudo ip link add lannetvm type bridge
+          sudo ip link add wannetvm type bridge
+          sudo ip link set dev lannetvm up
+          sudo ip link set dev wannetvm up
+          ip addr
+
+          # We need to disable the firewall otherwise the tap devices are not working properly
+          sudo ufw disable
+
+          # Start VMs
+          # Multipass and the OpenWrt container are using the same MAC QUI 52:54:00 which creates error. Let's use differnet MAC addresses here
+          multipass launch docker --network name=wannetvm,mac=06:00:00:11:22:33 --network name=lannetvm,mac=06:00:00:44:55:66 --name router-vm
+          multipass mount $GITHUB_WORKSPACE router-vm
+
+          multipass launch --network wannetvm --name wan-vm
+          multipass launch --network lannetvm --name lan-vm
+
+      - name: Show VM and and enviroments variables
+        run:  |
+          export
+          multipass exec router-vm -- lsb_release -a
+          multipass info
+
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Prepare VMs
+        run: |
+          # By default portainer is running at port 9000. But we need port 9000
+          multipass exec router-vm -- docker stop portainer
+
+          # Install python
+          multipass exec router-vm -- sudo apt-get -y install python3-pip python-is-python3
+
+          # Install everthing for pytest
+          multipass exec router-vm -- find .
+          #multipass exec router-vm -- export PATH=$PATH:~/.local/bin # not working
+          multipass exec router-vm -- pip install -r test/requirements.txt
+          
+          echo '******** router-vm IPs ********'
+          multipass exec router-vm -- ip addr
+
+          # Prepare WAN VM
+          echo '******** wan-vm IPs ********'
+          # Add IP address
+          multipass exec wan-vm -- sudo ip addr add 192.168.22.1/24 dev ens4
+          multipass exec wan-vm -- ip addr
+          # Install DHCP server
+          multipass exec wan-vm -- sudo sh -c 'apt-get update && apt-get install -y isc-dhcp-server && echo "default-lease-time 600;\nmax-lease-time 7200;\nsubnet 192.168.22.0 netmask 255.255.255.0 {\n  range 192.168.22.100 192.168.22.200;\n  option routers 192.168.22.1;\n  option domain-name-servers 8.8.8.8, 8.8.4.4;\n  option domain-name \"example.org\";\n}" > /etc/dhcp/dhcpd.conf && systemctl enable isc-dhcp-server && systemctl restart isc-dhcp-server'
+          multipass exec wan-vm -- cat /etc/dhcp/dhcpd.conf
+          # Enable NAT
+          multipass exec wan-vm -- sudo modprobe iptable_nat
+          multipass exec wan-vm -- sudo iptables -t nat -A POSTROUTING -o ens3 -j MASQUERADE
+          multipass exec wan-vm -- sudo sysctl -w net.ipv4.ip_forward=1
+
+          # Prepare LAN VM
+          echo '******** lan-vm IPs ********'
+          multipass exec lan-vm -- sudo ip addr add 172.31.1.2/24 dev ens4
+          multipass exec lan-vm -- ip addr
+
+          # Double check if virtual networks are working
+          echo '******** host IPs ********'
+          ip addr
+          sudo ufw status
+
+          echo '******** router-vm IPs ********'
+          multipass exec router-vm -- sudo ip addr add 192.168.22.99/24 dev ens4
+          multipass exec router-vm -- sudo ip addr add 172.31.1.99/24 dev ens5
+          multipass exec router-vm -- ip addr
+
+          echo '******** Test connection between router-vm, lan-vm and wan-vm ********'
+          multipass exec router-vm -- ping -c 1 192.168.22.1 # to wan-vm
+          multipass exec router-vm -- ping -c 1 172.31.1.2 # to lan-vm
+
+
+      - name: Run tests
+        #timeout-minutes: 10
+        run: |
+          # Run pytest
+          multipass exec router-vm -- sh -c 'cd test/ && ~/.local/bin/pytest -v -s'
+
+          

.gitignore

@@ -1 +1,5 @@
-
+.pytest_cache/
+test/tests/__pycache__/
+test/__pycache__/
+test/tests/.pytest_cache/
+.venv

.vscode/settings.json

@@ -0,0 +1,10 @@
+{
+    "python.testing.pytestArgs": [
+        "test/tests",
+        "--rootdir=test/",
+        "-v",
+        "-s"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

test/requirements.txt

@@ -0,0 +1,5 @@
+polling2
+pytest
+pytest-docker
+requests
+pyyaml

test/tests/conftest.py

@@ -0,0 +1,11 @@
+import pytest
+
+import os
+
+@pytest.hookimpl()
+def pytest_sessionfinish(session):
+    compose_file = os.path.join(str(session.path), "tests", "docker-compose.yml.generated")
+    
+    if os.path.isfile(compose_file):
+        os.remove(compose_file)
+        print(f"\nDeleted temporary file {compose_file} sucessfully")

test/tests/docker-compose.yml.template

@@ -0,0 +1,57 @@
+version: "3"
+services:
+  openwrt:
+    container_name: openwrt
+    image: albrechtloh/openwrt-docker:latest
+
+    environment:
+      #DEBUG: "y"
+      #IS_U_OS_APP: "true"
+
+      # Define number of VM CPUs
+      CPU_COUNT: 2
+
+      # Define which physical or virtual interfaces shall be attach direct to OpenWrt
+      WAN_IF: "host" # Usage of container host Ethernet connection as WAN. WARNING: Bad performace because of user space Ethernet implementation.
+      #WAN_IF: "none" # No WAN interface defined. This is useful if OpenWrt WAN acts a USB Wi-Fi client or as USB cellular modem
+      #WAN_IF: "enx00e04c0407a6" # Use this option to define a specific physical Ethernet interface
+
+      LAN_IF: "veth" # veth pair between OpenWrt and host system
+      #LAN_IF: "enx00e04d68033f" # Use this option to define a specific physical Ethernet interface
+
+      # Make LuCI OpenWrt web interface accessible via host LAN
+      FORWARD_LUCI: "true" 
+
+      # USB passthrough
+      # Format <vendorID>:<productID> like 'lsusb' is given
+      # Currently only two USB devices are supported
+      #USB_1: "0bda:8153"
+      #USB_2: "0e8d:7961"
+
+    ports:
+      # Port 8006 is console tty web VNC
+      - 8006:8006
+      # Port 8000 is available only when Docker WAN network is used.
+      # Use script /usr/bin/fw_wan_open_http to open the port
+      - 8000:8000
+      # SSH
+      # Use script /usr/bin/fw_wan_open_ssh to open the port
+      - 8022:8022
+    stop_grace_period: 2m
+    restart: on-failure
+    
+    # The following settings are mandantory to run the container
+    devices:
+      - /dev/kvm
+    device_cgroup_rules:
+      - 'c *:* rwm'
+    cap_add:
+      - NET_ADMIN
+    pid: "host"
+    privileged: true
+    volumes:
+      - /dev:/dev/
+      - data:/storage/
+
+volumes:
+  data:

test/tests/openwrt_docker_test.py

@@ -0,0 +1,334 @@
+import pytest
+
+import os
+import subprocess
+import polling2
+import json
+import requests
+import base64
+import time
+import yaml
+
+# Workaround to make parameters global accessible
+@pytest.fixture(scope='session')
+def parameter(request):
+    if hasattr(request,'param'):
+        return request.param
+    else:
+        return None
+
+# Adapt docker-compose.yml to our test needs
+@pytest.fixture(scope="session")
+def docker_compose_file(pytestconfig, parameter):
+        
+    compose_file_template = os.path.join(str(pytestconfig.rootdir), "tests", "docker-compose.yml.template")
+    compose_file = os.path.join(str(pytestconfig.rootdir), "tests", "docker-compose.yml.generated")
+
+    # Open template
+    with open(compose_file_template) as f:
+        list_doc = yaml.safe_load(f)
+
+    # Change enviroment variable
+    if parameter:
+        if len(parameter) == 2:
+            env_var = parameter[0]
+            value = parameter[1]
+            list_doc['services']['openwrt']['environment'][env_var] = value
+
+    # Save docker-compose file
+    with open(compose_file, "w") as f:
+        yaml.dump(list_doc, f)
+
+    return compose_file
+
+
+def is_container_running():
+    # Get current service list from supervisor
+    process = subprocess.run(['docker','exec','openwrt','supervisorctl','status'], 
+                         stdout=subprocess.PIPE, 
+                         universal_newlines=True)
+    
+    # Check if supervisord is running
+    if process.returncode == 0 \
+    or process.returncode == 3 : # Means a service has an issue
+        return True
+    else:
+        return False
+
+def get_service_status(service):
+    # Get current service list from supervisor
+    process = subprocess.run(['docker','exec','openwrt','supervisorctl','status'], 
+                         stdout=subprocess.PIPE, 
+                         universal_newlines=True)
+    
+    # Check if supervisord is running
+    if process.returncode == 0 \
+    or process.returncode == 3 : # Means a service has an issue
+        # Extract list from supervisorctl output
+        service_list = process.stdout.splitlines()
+
+        # Filter for service
+        service_status = [s for s in service_list if service in s][0].split()
+
+        return service_status[1]
+    else:
+        return None
+
+def is_service_started(service):
+    service_status = get_service_status(service)
+
+    if service_status != None:
+        # Check if service is running or in other states
+        if service_status == 'RUNNING' \
+        or service_status == 'BACKOFF' \
+        or service_status == 'EXITED' \
+        or service_status == 'FATAL' \
+        or service_status == 'UNKNOWN':
+            return True
+        else:
+            return False
+    else:
+        return False
+
+def run_openwrt_shell_command(command, *arg):
+    # Add double quotes
+    command = "\"" + command + "\""
+
+    # Build argument array
+    arguments = "["
+    for val in arg:
+        arguments += "\"" + val + "\","
+    arguments = arguments[:-1] # Remove the last ","
+    arguments += "]"
+
+    # Call qemu guest tools function 'guest-exec'. This is only working if OpenWrt is booted and the qemu guest tools are running
+    process = subprocess.run(['docker','exec','openwrt','sh','-c',r"""echo -ne '{"execute":"guest-exec", "arguments": { "path": """ + command + ", \"arg\":" + arguments + r""","capture-output": true}}' | nc -w 1 -U /run/qga.sock"""], 
+                         stdout=subprocess.PIPE,
+                         stderr=subprocess.PIPE, 
+                         universal_newlines=True)
+    
+    if process.returncode != 0:
+        return None
+    
+    if process.stdout == "":
+        return None
+    
+    # Assume json here, parse it
+    ret = json.loads(process.stdout)
+
+    if 'return' not in ret:
+        return None
+    
+    if 'pid' not in ret['return']:
+        return None
+
+    pid = ret['return']['pid']
+
+    # Get stdout of process
+    time.sleep(5) # Give command 5 seconds time to respond. It would be better to implement a real timeout here but I need to got he bed :-(
+    # Call qemu guest tools function 'guest-exec-status'. This is only working if OpenWrt is booted and the qemu guest tools are running
+    process = subprocess.run(['docker','exec','openwrt','sh','-c',r"""echo -ne '{"execute":"guest-exec-status", "arguments": { "pid": """ + str(pid) + r"""}}' | nc -w 1 -U /run/qga.sock"""], 
+                         stdout=subprocess.PIPE,
+                         stderr=subprocess.PIPE, 
+                         universal_newlines=True)
+    
+    if process.returncode != 0:
+        return None
+    
+    if process.stdout == "":
+        return None
+
+    # Assume json here, parse it
+    ret = json.loads(process.stdout)
+    data = ret['return']
+
+    # decode output
+    if 'err-data' in data:
+        data['err-data'] = base64.b64decode(data['err-data']).decode()
+
+    if 'out-data' in data:
+        data['out-data'] = base64.b64decode(data['out-data']).decode()
+
+    return data
+
+def get_openwrt_info():
+    # Call qemu guest tools function 'guest-get-osinfo'. This is only working if OpenWrt is booted and the qemu guest tools are running
+    process = subprocess.run(['docker','exec','openwrt','sh','-c',r"""echo -ne '{"execute":"guest-get-osinfo"}' | nc -w 1 -U /run/qga.sock"""], 
+                         stdout=subprocess.PIPE,
+                         stderr=subprocess.PIPE, 
+                         universal_newlines=True)
+    
+    if process.returncode != 0:
+        return None
+    
+    if process.stdout == "":
+        return None
+    
+    # Assume json here, parse it
+    info = json.loads(process.stdout)
+
+    if 'return' not in info:
+        return None
+    
+    if 'name' not in info['return']:
+        return None
+
+    return info['return']
+
+def is_openwrt_booted():
+    service_status = get_openwrt_info()
+
+    if service_status != None:
+        return True
+    else:
+        return False
+
+
+# ************************ Tests ************************
+
+def test_basic_container_start(docker_services):
+    docker_services.wait_until_responsive(
+        timeout=30.0, pause=0.5, check=lambda: is_container_running()
+    )
+    return
+
+
+def test_nginx_start(docker_services):
+    docker_services.wait_until_responsive(
+        timeout=30.0, pause=0.5, check=lambda: is_service_started('nginx')
+    )
+    
+    assert get_service_status('nginx') == 'RUNNING'
+
+
+def test_openwrt_start(docker_services):
+    docker_services.wait_until_responsive(
+        timeout=30.0, pause=0.5, check=lambda: is_service_started('openwrt')
+    )
+    
+    assert get_service_status('openwrt') == 'RUNNING'
+
+
+@pytest.mark.parametrize("parameter", 
+    [('FORWARD_LUCI','true'),('FORWARD_LUCI','false')], indirect=True,
+    ids=['FORWARD_LUCI=true', 'FORWARD_LUCI=false'])
+def test_caddy_start(docker_services, parameter):
+    try:
+        docker_services.wait_until_responsive(
+            timeout=30.0, pause=0.5, check=lambda: is_service_started('caddy')
+        )
+    except:
+        if parameter[1] == 'false':
+            return # We expect a timeout here. This is our test condition for FORWARD_LUCI=false
+    
+    # For FORWARD_LUCI=true
+    assert get_service_status('caddy') == 'RUNNING'
+
+
+def test_script_server_start(docker_services):
+    docker_services.wait_until_responsive(
+        timeout=30.0, pause=0.5, check=lambda: is_service_started('script-server')
+    )
+
+    assert get_service_status('script-server') == 'RUNNING'
+
+
+def test_openwrt_booted(docker_services):
+    docker_services.wait_until_responsive(
+        timeout=90.0, pause=1, check=lambda: is_openwrt_booted()
+    )
+
+    info = get_openwrt_info()
+    print(f"Running '{info['pretty-name'].rstrip()}'", end=' ')
+
+    assert 'OpenWrt' == info['name']
+
+
+@pytest.mark.parametrize("parameter", 
+    [('LAN_IF','veth'),('LAN_IF','ens5')], indirect=True,
+    ids=['LAN_IF=veth', 'LAN_IF=ens5'])
+def test_openwrt_lan(docker_services, parameter):
+    docker_services.wait_until_responsive(
+        timeout=90.0, pause=1, check=lambda: is_openwrt_booted()
+    )
+    
+    match parameter[1]:
+        case 'veth':
+            try:
+                response = polling2.poll(lambda: os.system("ping -c 1 172.31.1.1 >/dev/null") == 0, step=1, timeout=90)
+            except polling2.TimeoutException:
+                assert True, 'ping timeout'
+            return
+
+        case _: # Usage of real Ethernet interface e.g. 'eth0'
+            # This test is most likely only working in a github action enviroment because multiple VM are necessary to test it. See the action file, please.
+            # Try to ping LAN-VM
+            response = run_openwrt_shell_command("ping", "-c1", "-W2", "-w2", "172.31.1.2")
+            assert response['exitcode'] == 0
+            return
+    
+    assert False, 'Unknown parameter'
+
+
+@pytest.mark.parametrize("parameter", 
+    [('WAN_IF','host'),('WAN_IF','none'),('WAN_IF','ens4')], indirect=True,
+    ids=['WAN_IF=host', 'WAN_IF=none', 'WAN_IF=ens4'])
+def test_openwrt_wan(docker_services, parameter):
+    docker_services.wait_until_responsive(
+        timeout=90.0, pause=1, check=lambda: is_openwrt_booted()
+    )
+    
+    match parameter[1]:
+        case 'host':
+            # For some reason ping is not working at github actions, so use nslookup to test internet connection
+            response = run_openwrt_shell_command("nslookup", "google.com")
+            assert response['exitcode'] == 0
+            return
+
+        case 'none':
+            # We are looking for eth1. It should not existing
+            response = run_openwrt_shell_command("ip", "addr")        
+            assert ('eth1' in response['out-data']) == False
+            return
+        
+        case _: # Usage of real Ethernet interface e.g. 'eth0'
+            # This test is most likely only working in a github action enviroment because multiple VM are necessary to test it. See the action file, please
+            # Try to get IP address from WAN-VM
+            response = run_openwrt_shell_command("udhcpc", "-i", "eth1")
+            assert ('udhcpc: setting default routers: 192.168.22.1' in response['out-data']) == True
+
+            # Finally try Internet
+            response = run_openwrt_shell_command("nslookup", "google.com")
+            assert response['exitcode'] == 0
+
+            return
+
+    assert False, 'Unknown parameter'
+
+
+def test_openwrt_luci_forwarding(docker_services):
+    docker_services.wait_until_responsive(
+        timeout=90.0, pause=1, check=lambda: is_openwrt_booted()
+    )
+    
+    # Double check if caddy is still running
+    assert get_service_status('caddy') == 'RUNNING'
+
+    response = requests.get("http://localhost:9000")
+    
+    assert ('LuCI - Lua Configuration Interface' in response.content.decode()) == True
+
+
+@pytest.mark.parametrize("parameter", 
+    [('CPU_COUNT',1),('CPU_COUNT',2),('CPU_COUNT',3),('CPU_COUNT',4)], indirect=True,
+    ids=['CPU_COUNT=1', 'CPU_COUNT=2', 'CPU_COUNT=3', 'CPU_COUNT=4'])
+def test_cpu_num(docker_services, parameter):
+    docker_services.wait_until_responsive(
+        timeout=90.0, pause=1, check=lambda: is_openwrt_booted()
+    )
+    
+    # Get number of processors
+    response = run_openwrt_shell_command("cat", "/proc/cpuinfo")
+    cpu_num = response['out-data'].count('processor')
+        
+    assert parameter[1] == cpu_num