Skip to content

Testing Your Project

Jump to bottom
Stevie Chryn edited this page Jul 31, 2026 · 1 revision

Testing Your Project

The emulator is designed to be fully testable. The server runs in a background thread, tags can be reset between tests, and on_change listeners fire synchronously (unless defer=True), making assertions straightforward.

This guide uses Python's built-in unittest framework, but the same patterns apply to pytest.

Test Layout Convention

your_project/
├── tags.py
├── actions.py
├── __main__.py
└── tests/
    ├── __init__.py
    └── test_myfeature.py

Tags and action listeners defined in your project are imported directly into your test files.

Port Helper

Each test class should start its own server instance on a unique port to avoid conflicts when tests run in parallel or in sequence. You can either keep track of this and use set ports for each test or a simple counter in tests/__init__.py can handle this:

# tests/__init__.py
import itertools

_port_counter = itertools.count(44818) # use 44819 or something if your main server is actively running

def next_port() -> int:
    return next(_port_counter)

Server Lifecycle Helpers

Two module-level functions handle starting and stopping the server. Define them once and reuse across all test files.

import threading
import time

from ethernetip_emulator.server.device import AttributeDevice, apidict, device_controller
from ethernetip_emulator.server.tag_specs import tag_registry


def _start_server(port: int) -> tuple:
    server_control = apidict(timeout=1.0)
    AttributeDevice.set_server_control(server_control)

    def background_task():
        with AttributeDevice._actions.bind(AttributeDevice):
            device_controller(
                argv=tag_registry.build_argv(base_args=["--address", f":{port}"]),
                attribute_class=AttributeDevice,
                server={"control": server_control},
            )

    thread = threading.Thread(target=background_task, daemon=True)
    thread.start()
    time.sleep(0.5)  # allow the server to finish initialising
    return server_control, thread


def _stop_server(server_control, thread) -> None:
    server_control["done"] = True
    thread.join(timeout=5.0)
    tag_registry.invalidate()
    tag_registry._raw.clear()
    AttributeDevice.reset_defaults()

The tag_registry and AttributeDevice are module-level singletons. Without cleanup between test classes, tags registered in one class bleed into the next. _stop_server ensures a clean slate by:

  1. Signalling the server to exit (server_control["done"] = True)
  2. Waiting for the thread to finish (thread.join)
  3. Invalidating the built tag cache (tag_registry.invalidate())
  4. Clearing the raw tag list (tag_registry._raw.clear())
  5. Resetting device defaults and the attribute registry (AttributeDevice.reset_defaults())

Test Class Structure

Testing tag validation (no server needed)

Use a separate test class for type_validator tests.

class TestMyTagRegistry(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.server_control, cls.thread = _start_server(next_port())

    @classmethod
    def tearDownClass(cls):
        _stop_server(cls.server_control, cls.thread)

    def test_invalid_default_raises(self):
        with self.assertRaises((ValueError, TypeError)):
            tag_registry.register(
                lambda: [("my_tag", actions.type.BOOL(42))]
            )

Testing actions against a live server

Register your tags in setUpClass, start the server, and reset tag values to a known state in setUp before each test.

from ethernetip_emulator.server.device import actions
from your_project import tags    # triggers @tag_registry.register
from your_project import actions as app_actions  # triggers actions


class TestMyTagActions(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        # If your tags.py uses @tag_registry.register at import time,
        # importing it here is enough — no manual registration needed.
        cls.server_control, cls.thread = _start_server(next_port())

    @classmethod
    def tearDownClass(cls):
        _stop_server(cls.server_control, cls.thread)

    def setUp(self):
        # Reset tags to a known state before every test
        AttributeDevice._ensure_defaults()
        actions._lookup("O_BOOL")[slice(0, 1)] = [False]

Tip: Use actions._lookup(tag_name)[slice(...)] = [value] in setUp to make sure they you are only using tested methods on setUp. Never use a method that you or ethernetip-emulator hasn't tested in setUp.

Writing Tests

Reading and writing values

def test_get_val(self):
    self.assertEqual(actions.bool.get_val("O_BOOL"), False)

def test_set_val(self):
    actions.bool.set_val("O_BOOL", True)
    self.assertEqual(actions.bool.get_val("O_BOOL"), True)

Testing on_change callbacks

Because non-deferred listeners fire synchronously inside set_val, you can assert on results immediately after the write:

def test_on_change(self):
    changed = []

    @actions.bool.on_change("O_BOOL")
    def _(attr, key, value):
        changed.append(value[0])

    actions.bool.set_val("O_BOOL", True)
    self.assertEqual(changed[-1], True)

Testing deferred (defer=True) callbacks

Deferred listeners run in a background thread. Add a short time.sleep after the write to give the thread time to execute before asserting:

def test_on_change_deferred(self):
    changed = []

    @actions.bool.on_change("O_BOOL", defer=True)
    def _(attr, key, value):
        changed.append(value[0])

    actions.bool.set_val("O_BOOL", True)
    time.sleep(0.1)   # wait for the deferred thread
    self.assertEqual(changed[-1], True)

Testing missing tags

All get_val / set_val methods return None or [] for unknown tag names. Verify this explicitly:

def test_get_val_missing_tag(self):
    self.assertIsNone(actions.bool.get_val("no_such_tag"))

def test_set_val_missing_tag(self):
    result = actions.bool.set_val("no_such_tag", True)
    self.assertIsNone(result)

Testing array operations

Use setUp to reset the array, then test each operation in isolation:

def setUp(self):
    AttributeDevice._ensure_defaults()
    actions._lookup("O_Steps")[slice(0, 4)] = [False, False, False, False]

def test_append(self):
    actions.boolarray.append("O_Steps", True)
    self.assertEqual(
        actions.boolarray.get_val("O_Steps"),
        [True, False, False, False]
    )

def test_append_when_full(self):
    actions.boolarray.set_val("O_Steps", [True, True, True, True])
    result = actions.boolarray.append("O_Steps", True)
    self.assertFalse(result)

Full Example

# tags.py
from ethernetip_emulator.server.tag_specs import tag_registry
@tag_registry.register
def _():
    return [
        ("O_BOOL", actions.type.BOOL(False)),
        ("O_Steps", actions.type.BOOLARRAY([False, False, False, False])),
    ]
# tests/test_bool.py
import unittest
import time
import itertools

from ethernetip_emulator.server.device import AttributeDevice, apidict, device_controller, actions
from ethernetip_emulator.server.tag_specs import tag_registry
from tests import next_port

# --- helpers ---

_port_counter = itertools.count(44818) # use 44819 or something if your main server is actively running

def _next_port() -> int:
    return next(_port_counter)

def _start_server(port):
    server_control = apidict(timeout=1.0)
    AttributeDevice.set_server_control(server_control)
    def _run():
        with AttributeDevice._actions.bind(AttributeDevice):
            device_controller(
                argv=tag_registry.build_argv(base_args=["--address", f":{port}"]),
                attribute_class=AttributeDevice,
                server={"control": server_control},
            )
    import threading
    t = threading.Thread(target=_run, daemon=True)
    t.start()
    time.sleep(0.5)
    return server_control, t

def _stop_server(ctrl, t):
    ctrl["done"] = True
    t.join(timeout=5.0)
    tag_registry.invalidate()
    tag_registry._raw.clear()
    AttributeDevice.reset_defaults()

# --- tests ---

class TestBoolActions(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.server_control, cls.thread = _start_server(_next_port())

    @classmethod
    def tearDownClass(cls):
        _stop_server(cls.server_control, cls.thread)

    def setUp(self):
        AttributeDevice._ensure_defaults()
        actions._lookup("O_BOOL")[slice(0, 1)] = [False]
        actions._lookup("O_Steps")[slice(0, 4)] = [False, False, False, False]

    def test_set_and_get(self):
        actions.bool.set_val("O_BOOL", True)
        self.assertTrue(actions.bool.get_val("O_BOOL"))

    def test_on_change_fires(self):
        fired = []
        @actions.bool.on_change("O_BOOL")
        def _(attr, key, value):
            fired.append(value[0])
        actions.bool.set_val("O_BOOL", True)
        self.assertEqual(fired[-1], True)

    def test_array_toggle(self):
        new_val = actions.boolarray.toggle_bit("O_Steps", 0)
        self.assertTrue(new_val)
        self.assertTrue(actions.boolarray.get_bit("O_Steps", 0))


if __name__ == "__main__":
    unittest.main()

Common Pitfalls

Problem Cause Fix
Tags from a previous test class appear in the next _raw not cleared between classes Always call _stop_server in tearDownClass
on_change fires during setUp Using set_val to reset state Never rely on on_change events between tests
Deferred callback assertion fails Thread hasn't run yet Add time.sleep(0.1) after set_val when testing defer=True listeners
Port already in use Two test classes share a port Use next_port() for every _start_server call
Tag not found after server restart AttributeDevice.reset_defaults() clears the registry Re-register tags in the next class's setUpClass

Clone this wiki locally