-
- | Protocol |
- Plausible Use Cases |
- Operations |
-
-
- | HTTP |
- Web Apps |
-
- readproperty,
- writeproperty,
- observeproperty,
- unobserveproperty,
- invokeaction,
- subscribeevent,
- unsubscribeevent,
- readmultipleproperties,
- writemultipleproperties,
- readallproperties,
- writeallproperties
-
- properties and actions can be operated in a oneway and no-block manner (issue and query later format) as well
- |
-
-
- | ZMQ TCP |
- Networked Control Systems, subnet protected containerized apps like in Kubernetes |
-
-
- | ZMQ IPC |
- Desktop Applications, Python Dashboards without exposing device API directly on network |
-
-
- | ZMQ INPROC |
-
- High Speed Desktop Applications (again, not exposed on network), currently you will need some CPP magic or disable GIL to leverage it fully
- |
-
-
- | MQTT |
-
- Reliable pub-sub & incorporating into existing systems that use MQTT for
lightweight messaging
- |
-
- observeproperty,
- unobserveproperty,
- subscribeevent,
- unsubscribeevent
- |
-
-
- | MQTT with websockets |
-
- Reliable pub-sub for web applications, planned for November 2025 release.
- |
-
- observeproperty,
- unobserveproperty,
- subscribeevent,
- unsubscribeevent
- |
-
-
diff --git a/pyproject.toml b/pyproject.toml
index a14503e6..5373f3cd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -98,7 +98,7 @@ linux = [
minversion = "8.0"
addopts = "-ra --strict-markers --strict-config --ignore=lib64"
testpaths = ["tests/pytests-new"]
-python_files = ["test_*_pytest.py"]
+python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
markers = [
From 17720bd0293b373d54ee36875ac726329eb1a540 Mon Sep 17 00:00:00 2001
From: Vignesh Venkatasubramanian Vaidyanathan
<62492557+VigneshVSV@users.noreply.github.com>
Date: Sat, 1 Nov 2025 11:56:43 +0100
Subject: [PATCH 02/43] move upto test 3 to pytest
---
tests/pytests-new/conftest.py | 84 ++-
tests/pytests-new/test_01_message.py | 205 ++++++
tests/pytests-new/test_01_message_pytest.py | 353 ---------
tests/pytests-new/test_02_socket.py | 240 ++++++
tests/pytests-new/test_03_serializers.py | 145 ++++
tests/pytests-new/test_04_thing_init.py | 778 ++++++++++++++++++++
tests/pytests-new/things/__init__.py | 4 +
tests/pytests-new/things/spectrometer.py | 330 +++++++++
tests/pytests-new/things/starter.py | 124 ++++
tests/pytests-new/things/test_thing.py | 771 +++++++++++++++++++
10 files changed, 2643 insertions(+), 391 deletions(-)
create mode 100644 tests/pytests-new/test_01_message.py
delete mode 100644 tests/pytests-new/test_01_message_pytest.py
create mode 100644 tests/pytests-new/test_02_socket.py
create mode 100644 tests/pytests-new/test_03_serializers.py
create mode 100644 tests/pytests-new/test_04_thing_init.py
create mode 100644 tests/pytests-new/things/__init__.py
create mode 100644 tests/pytests-new/things/spectrometer.py
create mode 100644 tests/pytests-new/things/starter.py
create mode 100644 tests/pytests-new/things/test_thing.py
diff --git a/tests/pytests-new/conftest.py b/tests/pytests-new/conftest.py
index 9da1aed3..08629410 100644
--- a/tests/pytests-new/conftest.py
+++ b/tests/pytests-new/conftest.py
@@ -1,66 +1,74 @@
"""
Pytest configuration and shared fixtures for hololinked tests.
"""
+
import asyncio
import pytest
import zmq.asyncio
+import sys
+from typing import Generator
from uuid import uuid4
from faker import Faker
+from dataclasses import dataclass
from hololinked.config import global_config
+from hololinked.serializers import Serializers
+
+
+@dataclass
+class AppIDs:
+ """
+ Application related IDs generally used by end-user,
+ like server, client, and thing IDs.
+ """
+
+ server_id: str
+ """RPC server ID"""
+ client_id: str
+ """A client ID"""
+ thing_id: str
+ """A thing ID"""
@pytest.fixture(scope="session")
-def event_loop():
+def fake() -> Faker:
+ """Provide a Faker instance for generating test data."""
+ return Faker()
+
+
+@pytest.fixture()
+def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
"""Create an instance of the default event loop for the test session."""
loop = asyncio.get_event_loop_policy().new_event_loop()
+ if sys.platform.startswith("win"):
+ asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+ asyncio.set_event_loop(loop)
yield loop
loop.close()
-@pytest.fixture(scope="class")
-def zmq_context():
+@pytest.fixture()
+def zmq_context() -> Generator[zmq.asyncio.Context, None, None]:
"""Setup ZMQ context for test classes."""
global_config.ZMQ_CONTEXT = zmq.asyncio.Context()
yield global_config.ZMQ_CONTEXT
- # Cleanup is handled by the context manager
-
+ global_config.ZMQ_CONTEXT.term()
-@pytest.fixture(scope="class")
-def test_ids():
- """Generate unique test IDs for each test class."""
- return {
- "server_id": f"test-server-{uuid4().hex[:8]}",
- "client_id": f"test-client-{uuid4().hex[:8]}",
- "thing_id": f"test-thing-{uuid4().hex[:8]}"
- }
-
-@pytest.fixture(scope="session")
-def fake():
- """Provide a Faker instance for generating test data."""
- return Faker()
-
-
-@pytest.fixture(autouse=True)
-def setup_test_environment(zmq_context):
- """Automatically setup test environment for each test."""
+@pytest.fixture()
+def setup_test_environment(zmq_context, event_loop):
+ """Automatically setup test environment for each file"""
# This fixture runs automatically for every test
- pass
+ yield
+ # Reset serializers after each test
+ Serializers().reset()
-def pytest_configure(config):
- """Configure pytest with custom settings."""
- config.addinivalue_line(
- "markers", "order: mark test to run in a specific order"
+@pytest.fixture()
+def app_ids() -> AppIDs:
+ """Generate unique test IDs for server, client, and thing for each test"""
+ return AppIDs(
+ server_id=f"test-server-{uuid4().hex[:8]}",
+ client_id=f"test-client-{uuid4().hex[:8]}",
+ thing_id=f"test-thing-{uuid4().hex[:8]}",
)
-
-
-def pytest_collection_modifyitems(config, items):
- """Modify test collection to add ordering markers."""
- # Add order markers based on test file names
- for item in items:
- if "test_01_" in item.nodeid:
- item.add_marker(pytest.mark.order(1))
- elif "test_00_" in item.nodeid:
- item.add_marker(pytest.mark.order(0))
diff --git a/tests/pytests-new/test_01_message.py b/tests/pytests-new/test_01_message.py
new file mode 100644
index 00000000..324c6e4d
--- /dev/null
+++ b/tests/pytests-new/test_01_message.py
@@ -0,0 +1,205 @@
+"""
+Pytest tests for message validation and messaging contract.
+Converted from unittest to pytest format.
+"""
+
+import pytest
+from uuid import UUID, uuid4
+
+from hololinked.core.zmq.message import (
+ EXIT,
+ OPERATION,
+ HANDSHAKE,
+ PreserializedData,
+ SerializableData,
+ RequestHeader,
+ EventHeader,
+ RequestMessage,
+) # client to server
+from hololinked.core.zmq.message import (
+ TIMEOUT,
+ INVALID_MESSAGE,
+ ERROR,
+ REPLY,
+ ResponseMessage,
+ ResponseHeader,
+ EventMessage,
+) # server to client
+from hololinked.serializers.serializers import Serializers
+
+try:
+ from .conftest import AppIDs
+except ImportError:
+ from conftest import AppIDs
+
+
+def validate_request_message(request_message: RequestMessage, app_ids: AppIDs) -> None:
+ """call this method to validate request message"""
+ # req. 1. check message ID is a UUID
+ assert isinstance(request_message.id, UUID) or isinstance(UUID(request_message.id, version=4), UUID)
+ # req. 2. generated byte array must confine to predefined length (which is readonly & fixed)
+ assert len(request_message.byte_array) == request_message.length
+ # req. 3. receiver which must be the server ID
+ assert request_message.receiver_id == app_ids.server_id
+ # req. 4. sender_id is the client ID
+ assert request_message.sender_id == app_ids.client_id
+ # req. 5. all indices of byte array are bytes
+ for obj in request_message.byte_array:
+ assert isinstance(obj, bytes)
+ # req. 6. check that header is correct type (RequestHeader dataclass/struct)
+ assert isinstance(request_message.header, RequestHeader)
+ # req. 7 check that body is correct type (list of SerializableData and PreserializedData)
+ assert isinstance(request_message.body, list)
+ assert len(request_message.body) == 2
+ assert isinstance(request_message.body[0], SerializableData)
+ assert isinstance(request_message.body[1], PreserializedData)
+
+
+def validate_response_message(response_message: ResponseMessage, app_ids: AppIDs) -> None:
+ """call this method to validate response message"""
+
+ # check message ID is a UUID
+ assert isinstance(response_message.id, UUID) or isinstance(UUID(response_message.id, version=4), UUID)
+ # check message length
+ assert len(response_message.byte_array) == response_message.length
+ # check receiver which must be the client
+ assert response_message.receiver_id == app_ids.client_id
+ # sender_id is not set before sending message on the socket
+ assert response_message.sender_id == app_ids.server_id
+ # check that all indices are bytes
+ for obj in response_message.byte_array:
+ assert isinstance(obj, bytes)
+ # check that header is correct type
+ assert isinstance(response_message.header, ResponseHeader)
+ # check that body is correct type
+ assert isinstance(response_message.body, list)
+ assert len(response_message.body) == 2
+ assert isinstance(response_message.body[0], SerializableData)
+ assert isinstance(response_message.body[1], PreserializedData)
+
+
+def validate_event_message(event_message: EventMessage, app_ids: AppIDs) -> None:
+ """call this method to validate event message"""
+
+ # check message ID is a UUID
+ assert isinstance(event_message.id, UUID) or isinstance(UUID(event_message.id, version=4), UUID)
+ # check message length
+ assert len(event_message.byte_array) == event_message.length
+ # no receiver id for event message, only event id
+ assert isinstance(event_message.event_id, str)
+ # sender_id is not set before sending message on the socket
+ assert event_message.sender_id == app_ids.server_id
+ # check that all indices are bytes
+ for obj in event_message.byte_array:
+ assert isinstance(obj, bytes)
+ # check that header is correct type
+ assert isinstance(event_message.header, EventHeader)
+ # check that body is correct type
+ assert isinstance(event_message.body, list)
+ assert len(event_message.body) == 2
+ assert isinstance(event_message.body[0], SerializableData)
+ assert isinstance(event_message.body[1], PreserializedData)
+
+
+@pytest.mark.order(1)
+def test_1_request_message(app_ids: AppIDs) -> None:
+ """test the request message"""
+
+ # request messages types are OPERATION, HANDSHAKE & EXIT
+ request_message = RequestMessage.craft_from_arguments(
+ receiver_id=app_ids.server_id,
+ sender_id=app_ids.client_id,
+ thing_id=app_ids.thing_id,
+ objekt="some_prop",
+ operation="readproperty",
+ )
+ validate_request_message(request_message, app_ids)
+ # check message type for the above craft_from_arguments method
+ assert request_message.type == OPERATION
+
+ request_message = RequestMessage.craft_with_message_type(
+ receiver_id=app_ids.server_id, sender_id=app_ids.client_id, message_type=HANDSHAKE
+ )
+ validate_request_message(request_message, app_ids)
+ # check message type for the above craft_with_message_type method
+ assert request_message.type == HANDSHAKE
+
+ request_message = RequestMessage.craft_with_message_type(
+ receiver_id=app_ids.server_id, sender_id=app_ids.client_id, message_type=EXIT
+ )
+ validate_request_message(request_message, app_ids)
+ # check message type for the above craft_with_message_type method
+ assert request_message.type == EXIT
+
+
+def test_2_response_message(app_ids: AppIDs) -> None:
+ """test the response message"""
+
+ # response messages types are HANDSHAKE, TIMEOUT, INVALID_MESSAGE, ERROR and REPLY
+ response_message = ResponseMessage.craft_from_arguments(
+ receiver_id=app_ids.client_id,
+ sender_id=app_ids.server_id,
+ message_type=HANDSHAKE,
+ message_id=uuid4(),
+ )
+ validate_response_message(response_message, app_ids)
+ # check message type for the above craft_with_message_type method
+ assert response_message.type == HANDSHAKE
+
+ response_message = ResponseMessage.craft_from_arguments(
+ receiver_id=app_ids.client_id,
+ sender_id=app_ids.server_id,
+ message_type=TIMEOUT,
+ message_id=uuid4(),
+ )
+ validate_response_message(response_message, app_ids)
+ # check message type for the above craft_with_message_type method
+ assert response_message.type == TIMEOUT
+
+ response_message = ResponseMessage.craft_from_arguments(
+ receiver_id=app_ids.client_id,
+ sender_id=app_ids.server_id,
+ message_type=INVALID_MESSAGE,
+ message_id=uuid4(),
+ )
+ validate_response_message(response_message, app_ids)
+ # check message type for the above craft_with_message_type method
+ assert response_message.type == INVALID_MESSAGE
+
+ response_message = ResponseMessage.craft_from_arguments(
+ receiver_id=app_ids.client_id,
+ sender_id=app_ids.server_id,
+ message_type=ERROR,
+ message_id=uuid4(),
+ payload=SerializableData(Exception("test")),
+ )
+ validate_response_message(response_message, app_ids)
+ assert response_message.type == ERROR
+ assert isinstance(Serializers.json.loads(response_message._bytes[2]), dict)
+
+ request_message = RequestMessage.craft_from_arguments(
+ sender_id=app_ids.client_id,
+ receiver_id=app_ids.server_id,
+ thing_id=app_ids.thing_id,
+ objekt="some_prop",
+ operation="readProperty",
+ )
+ request_message._sender_id = app_ids.client_id # will be done by craft_from_self
+ response_message = ResponseMessage.craft_reply_from_request(
+ request_message=request_message,
+ )
+ validate_response_message(response_message, app_ids)
+ assert response_message.type == REPLY
+ assert Serializers.json.loads(response_message._bytes[3]) is None # INDEX_BODY = 3
+ assert request_message.id == response_message.id
+
+
+def test_3_event_message(app_ids: AppIDs) -> None:
+ """test the event message"""
+ event_message = EventMessage.craft_from_arguments(
+ event_id="test-event",
+ sender_id=app_ids.server_id,
+ payload=SerializableData("test"),
+ preserialized_payload=PreserializedData(b"test"),
+ )
+ validate_event_message(event_message, app_ids)
diff --git a/tests/pytests-new/test_01_message_pytest.py b/tests/pytests-new/test_01_message_pytest.py
deleted file mode 100644
index 122a38d1..00000000
--- a/tests/pytests-new/test_01_message_pytest.py
+++ /dev/null
@@ -1,353 +0,0 @@
-"""
-Pytest tests for message validation and messaging contract.
-Converted from unittest to pytest format.
-"""
-
-import pytest
-from uuid import UUID, uuid4
-
-from hololinked.core.zmq.message import (
- EXIT,
- OPERATION,
- HANDSHAKE,
- PreserializedData,
- SerializableData,
- RequestHeader,
- EventHeader,
- RequestMessage,
-) # client to server
-from hololinked.core.zmq.message import (
- TIMEOUT,
- INVALID_MESSAGE,
- ERROR,
- REPLY,
- ERROR,
- ResponseMessage,
- ResponseHeader,
- EventMessage,
-) # server to client
-from hololinked.serializers.serializers import Serializers
-
-
-class MessageValidatorMixin:
- """A mixin class to validate request and response messages"""
-
- @pytest.fixture(autouse=True)
- def setup_message_validator(self, test_ids):
- """Setup message validator with test IDs."""
- self.server_id = test_ids["server_id"]
- self.client_id = test_ids["client_id"]
- self.thing_id = test_ids["thing_id"]
-
- def validate_request_message(self, request_message: RequestMessage) -> None:
- """call this method to validate request message"""
-
- # req. 1. check message ID is a UUID
- assert isinstance(request_message.id, UUID) or isinstance(UUID(request_message.id, version=4), UUID)
- # req. 2. generated byte array must confine to predefined length (which is readonly & fixed)
- assert len(request_message.byte_array) == request_message.length
- # req. 3. receiver which must be the server ID
- assert request_message.receiver_id == self.server_id
- # req. 4. sender_id is the client ID
- assert request_message.sender_id == self.client_id
- # req. 5. all indices of byte array are bytes
- for obj in request_message.byte_array:
- assert isinstance(obj, bytes)
- # req. 6. check that header is correct type (RequestHeader dataclass/struct)
- assert isinstance(request_message.header, RequestHeader)
- # req. 7 check that body is correct type (list of SerializableData and PreserializedData)
- assert isinstance(request_message.body, list)
- assert len(request_message.body) == 2
- assert isinstance(request_message.body[0], SerializableData)
- assert isinstance(request_message.body[1], PreserializedData)
-
- def validate_response_message(self, response_message: ResponseMessage) -> None:
- """call this method to validate response message"""
-
- # check message ID is a UUID
- assert isinstance(response_message.id, UUID) or isinstance(UUID(response_message.id, version=4), UUID)
- # check message length
- assert len(response_message.byte_array) == response_message.length
- # check receiver which must be the client
- assert response_message.receiver_id == self.client_id
- # sender_id is not set before sending message on the socket
- assert response_message.sender_id == self.server_id
- # check that all indices are bytes
- for obj in response_message.byte_array:
- assert isinstance(obj, bytes)
- # check that header is correct type
- assert isinstance(response_message.header, ResponseHeader)
- # check that body is correct type
- assert isinstance(response_message.body, list)
- assert len(response_message.body) == 2
- assert isinstance(response_message.body[0], SerializableData)
- assert isinstance(response_message.body[1], PreserializedData)
-
- def validate_event_message(self, event_message: EventMessage) -> None:
- """call this method to validate event message"""
-
- # check message ID is a UUID
- assert isinstance(event_message.id, UUID) or isinstance(UUID(event_message.id, version=4), UUID)
- # check message length
- assert len(event_message.byte_array) == event_message.length
- # no receiver id for event message, only event id
- assert isinstance(event_message.event_id, str)
- # sender_id is not set before sending message on the socket
- assert event_message.sender_id == self.server_id
- # check that all indices are bytes
- for obj in event_message.byte_array:
- assert isinstance(obj, bytes)
- # check that header is correct type
- assert isinstance(event_message.header, EventHeader)
- # check that body is correct type
- assert isinstance(event_message.body, list)
- assert len(event_message.body) == 2
- assert isinstance(event_message.body[0], SerializableData)
- assert isinstance(event_message.body[1], PreserializedData)
-
-
-@pytest.mark.order(1)
-class TestMessagingContract(MessageValidatorMixin):
- """Tests request and response messages"""
-
- def test_1_request_message(self):
- """test the request message"""
-
- # request messages types are OPERATION, HANDSHAKE & EXIT
- request_message = RequestMessage.craft_from_arguments(
- receiver_id=self.server_id,
- sender_id=self.client_id,
- thing_id=self.thing_id,
- objekt="some_prop",
- operation="readproperty",
- )
- self.validate_request_message(request_message)
- # check message type for the above craft_from_arguments method
- assert request_message.type == OPERATION
-
- request_message = RequestMessage.craft_with_message_type(
- receiver_id=self.server_id, sender_id=self.client_id, message_type=HANDSHAKE
- )
- self.validate_request_message(request_message)
- # check message type for the above craft_with_message_type method
- assert request_message.type == HANDSHAKE
-
- request_message = RequestMessage.craft_with_message_type(
- receiver_id=self.server_id, sender_id=self.client_id, message_type=EXIT
- )
- self.validate_request_message(request_message)
- # check message type for the above craft_with_message_type method
- assert request_message.type == EXIT
-
- def test_2_response_message(self):
- """test the response message"""
-
- # response messages types are HANDSHAKE, TIMEOUT, INVALID_MESSAGE, ERROR and REPLY
- response_message = ResponseMessage.craft_from_arguments(
- receiver_id=self.client_id,
- sender_id=self.server_id,
- message_type=HANDSHAKE,
- message_id=uuid4(),
- )
- self.validate_response_message(response_message)
- # check message type for the above craft_with_message_type method
- assert response_message.type == HANDSHAKE
-
- response_message = ResponseMessage.craft_from_arguments(
- receiver_id=self.client_id,
- sender_id=self.server_id,
- message_type=TIMEOUT,
- message_id=uuid4(),
- )
- self.validate_response_message(response_message)
- # check message type for the above craft_with_message_type method
- assert response_message.type == TIMEOUT
-
- response_message = ResponseMessage.craft_from_arguments(
- receiver_id=self.client_id,
- sender_id=self.server_id,
- message_type=INVALID_MESSAGE,
- message_id=uuid4(),
- )
- self.validate_response_message(response_message)
- # check message type for the above craft_with_message_type method
- assert response_message.type == INVALID_MESSAGE
-
- response_message = ResponseMessage.craft_from_arguments(
- receiver_id=self.client_id,
- sender_id=self.server_id,
- message_type=ERROR,
- message_id=uuid4(),
- payload=SerializableData(Exception("test")),
- )
- self.validate_response_message(response_message)
- assert response_message.type == ERROR
- assert isinstance(Serializers.json.loads(response_message._bytes[2]), dict)
-
- request_message = RequestMessage.craft_from_arguments(
- sender_id=self.client_id,
- receiver_id=self.server_id,
- thing_id=self.thing_id,
- objekt="some_prop",
- operation="readProperty",
- )
- request_message._sender_id = self.client_id # will be done by craft_from_self
- response_message = ResponseMessage.craft_reply_from_request(
- request_message=request_message,
- )
- self.validate_response_message(response_message)
- assert response_message.type == REPLY
- assert Serializers.json.loads(response_message._bytes[3]) is None # INDEX_BODY = 3
- assert request_message.id == response_message.id
-
- def test_3_event_message(self):
- """test the event message"""
- event_message = EventMessage.craft_from_arguments(
- event_id="test-event",
- sender_id=self.server_id,
- payload=SerializableData("test"),
- preserialized_payload=PreserializedData(b"test"),
- )
- self.validate_event_message(event_message)
-
-
-# def validate_request_message(request_message, server_id, client_id):
-# """Validate request message contract."""
-# assert isinstance(request_message.id, UUID) or isinstance(UUID(request_message.id, version=4), UUID)
-# assert len(request_message.byte_array) == request_message.length
-# assert request_message.receiver_id == server_id
-# assert request_message.sender_id == client_id
-# for obj in request_message.byte_array:
-# assert isinstance(obj, bytes)
-# assert isinstance(request_message.header, RequestHeader)
-# assert isinstance(request_message.body, list)
-# assert len(request_message.body) == 2
-# assert isinstance(request_message.body[0], SerializableData)
-# assert isinstance(request_message.body[1], PreserializedData)
-
-# def validate_response_message(response_message, server_id, client_id):
-# """Validate response message contract."""
-# assert isinstance(response_message.id, UUID) or isinstance(UUID(response_message.id, version=4), UUID)
-# assert len(response_message.byte_array) == response_message.length
-# assert response_message.receiver_id == client_id
-# assert response_message.sender_id == server_id
-# for obj in response_message.byte_array:
-# assert isinstance(obj, bytes)
-# assert isinstance(response_message.header, ResponseHeader)
-# assert isinstance(response_message.body, list)
-# assert len(response_message.body) == 2
-# assert isinstance(response_message.body[0], SerializableData)
-# assert isinstance(response_message.body[1], PreserializedData)
-
-# def validate_event_message(event_message, server_id):
-# """Validate event message contract."""
-# assert isinstance(event_message.id, UUID) or isinstance(UUID(event_message.id, version=4), UUID)
-# assert len(event_message.byte_array) == event_message.length
-# assert isinstance(event_message.event_id, str)
-# assert event_message.sender_id == server_id
-# for obj in event_message.byte_array:
-# assert isinstance(obj, bytes)
-# assert isinstance(event_message.header, EventHeader)
-# assert isinstance(event_message.body, list)
-# assert len(event_message.body) == 2
-# assert isinstance(event_message.body[0], SerializableData)
-# assert isinstance(event_message.body[1], PreserializedData)
-
-# @pytest.fixture(scope="module")
-# def test_ids():
-# return {
-# "server_id": "server-uuid",
-# "client_id": "client-uuid",
-# "thing_id": "thing-uuid",
-# }
-
-# @pytest.mark.order(1)
-# class TestMessagingContract:
-# """Tests request and response messages"""
-
-# def test_1_request_message(self, test_ids):
-# request_message = RequestMessage.craft_from_arguments(
-# receiver_id=test_ids["server_id"],
-# sender_id=test_ids["client_id"],
-# thing_id=test_ids["thing_id"],
-# objekt="some_prop",
-# operation="readproperty",
-# )
-# validate_request_message(request_message, test_ids["server_id"], test_ids["client_id"])
-# assert request_message.type == OPERATION
-
-# request_message = RequestMessage.craft_with_message_type(
-# receiver_id=test_ids["server_id"], sender_id=test_ids["client_id"], message_type=HANDSHAKE
-# )
-# validate_request_message(request_message, test_ids["server_id"], test_ids["client_id"])
-# assert request_message.type == HANDSHAKE
-
-# request_message = RequestMessage.craft_with_message_type(
-# receiver_id=test_ids["server_id"], sender_id=test_ids["client_id"], message_type=EXIT
-# )
-# validate_request_message(request_message, test_ids["server_id"], test_ids["client_id"])
-# assert request_message.type == EXIT
-
-# def test_2_response_message(self, test_ids):
-# response_message = ResponseMessage.craft_from_arguments(
-# receiver_id=test_ids["client_id"],
-# sender_id=test_ids["server_id"],
-# message_type=HANDSHAKE,
-# message_id=uuid4(),
-# )
-# validate_response_message(response_message, test_ids["server_id"], test_ids["client_id"])
-# assert response_message.type == HANDSHAKE
-
-# response_message = ResponseMessage.craft_from_arguments(
-# receiver_id=test_ids["client_id"],
-# sender_id=test_ids["server_id"],
-# message_type=TIMEOUT,
-# message_id=uuid4(),
-# )
-# validate_response_message(response_message, test_ids["server_id"], test_ids["client_id"])
-# assert response_message.type == TIMEOUT
-
-# response_message = ResponseMessage.craft_from_arguments(
-# receiver_id=test_ids["client_id"],
-# sender_id=test_ids["server_id"],
-# message_type=INVALID_MESSAGE,
-# message_id=uuid4(),
-# )
-# validate_response_message(response_message, test_ids["server_id"], test_ids["client_id"])
-# assert response_message.type == INVALID_MESSAGE
-
-# response_message = ResponseMessage.craft_from_arguments(
-# receiver_id=test_ids["client_id"],
-# sender_id=test_ids["server_id"],
-# message_type=ERROR,
-# message_id=uuid4(),
-# payload=SerializableData(Exception("test")),
-# )
-# validate_response_message(response_message, test_ids["server_id"], test_ids["client_id"])
-# assert response_message.type == ERROR
-# assert isinstance(Serializers.json.loads(response_message._bytes[2]), dict)
-
-# request_message = RequestMessage.craft_from_arguments(
-# sender_id=test_ids["client_id"],
-# receiver_id=test_ids["server_id"],
-# thing_id=test_ids["thing_id"],
-# objekt="some_prop",
-# operation="readProperty",
-# )
-# request_message._sender_id = test_ids["client_id"]
-# response_message = ResponseMessage.craft_reply_from_request(
-# request_message=request_message,
-# )
-# validate_response_message(response_message, test_ids["server_id"], test_ids["client_id"])
-# assert response_message.type == REPLY
-# assert Serializers.json.loads(response_message._bytes[3]) is None
-# assert request_message.id == response_message.id
-
-# def test_3_event_message(self, test_ids):
-# event_message = EventMessage.craft_from_arguments(
-# event_id="test-event",
-# sender_id=test_ids["server_id"],
-# payload=SerializableData("test"),
-# preserialized_payload=PreserializedData(b"test"),
-# )
-# validate_event_message(event_message, test_ids["server_id"])
diff --git a/tests/pytests-new/test_02_socket.py b/tests/pytests-new/test_02_socket.py
new file mode 100644
index 00000000..cef853a5
--- /dev/null
+++ b/tests/pytests-new/test_02_socket.py
@@ -0,0 +1,240 @@
+import pytest
+import zmq.asyncio
+
+from hololinked.core.zmq.brokers import BaseZMQ
+from hololinked.constants import ZMQ_TRANSPORTS
+
+
+def test_1_socket_creation_defaults(zmq_context):
+ """check the default settings of socket creation - an IPC socket which is a ROUTER and async"""
+ socket, socket_address = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=zmq_context,
+ )
+ assert isinstance(socket, zmq.asyncio.Socket)
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert socket.socket_type == zmq.ROUTER
+ assert socket_address.startswith("ipc://")
+ assert socket_address.endswith(".ipc")
+ socket.close()
+
+
+def test_2_context_options():
+ """
+ Check that context and socket type are as expected.
+ Async context should be used for async socket and sync context for sync socket.
+ """
+ context = zmq.Context()
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ )
+ assert isinstance(socket, zmq.Socket)
+ assert not isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+ context.term()
+
+ context = zmq.asyncio.Context()
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ )
+ assert isinstance(socket, zmq.Socket)
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+ context.term()
+
+
+def test_3_transport_options():
+ """check only three transport options are supported"""
+ context = zmq.asyncio.Context()
+ socket, socket_address = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ access_point="tcp://*:5555",
+ )
+ for sock_addr in [socket_address, socket.getsockopt_string(zmq.LAST_ENDPOINT)]:
+ assert sock_addr.startswith("tcp://")
+ assert sock_addr.endswith(":5555")
+ socket.close()
+
+ socket, socket_address = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ access_point="IPC",
+ )
+
+ assert socket_address == socket.getsockopt_string(zmq.LAST_ENDPOINT)
+ assert socket_address.startswith("ipc://")
+ assert socket_address.endswith(".ipc")
+ socket.close()
+
+ socket, socket_address = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ access_point="INPROC",
+ )
+ assert socket_address == socket.getsockopt_string(zmq.LAST_ENDPOINT)
+ assert socket_address.startswith("inproc://")
+ assert socket_address.endswith("test-server")
+ socket.close()
+ context.term()
+
+ # Specify transport as enum and do the same tests
+ context = zmq.Context()
+ socket, socket_address = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ access_point=ZMQ_TRANSPORTS.INPROC,
+ )
+ assert socket_address.startswith("inproc://")
+ assert socket_address.endswith("test-server")
+ socket.close()
+
+ socket, socket_address = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ access_point=ZMQ_TRANSPORTS.IPC,
+ )
+ assert socket_address.startswith("ipc://")
+ assert socket_address.endswith(".ipc")
+ socket.close()
+
+ socket, socket_address = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ access_point=ZMQ_TRANSPORTS.TCP,
+ )
+ assert socket_address.startswith("tcp://")
+ # Strip the port number from TCP address and check if it's a valid port integer
+ host, port_str = socket_address.rsplit(":", 1)
+ assert port_str.isdigit()
+ assert 0 < int(port_str) < 65536
+ socket.close()
+ context.term()
+
+ # check that other transport options raise error
+ context = zmq.asyncio.Context()
+ with pytest.raises(NotImplementedError):
+ BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ access_point="PUB",
+ )
+
+
+def test_4_socket_options():
+ """check that socket options are as expected"""
+ context = zmq.asyncio.Context()
+
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ socket_type=zmq.ROUTER,
+ )
+ assert socket.socket_type == zmq.ROUTER
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ socket_type=zmq.DEALER,
+ )
+ assert socket.socket_type == zmq.DEALER
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ socket_type=zmq.PUB,
+ )
+ assert socket.socket_type == zmq.PUB
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ socket_type=zmq.SUB,
+ )
+ assert socket.socket_type == zmq.SUB
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ socket_type=zmq.PAIR,
+ )
+ assert socket.socket_type == zmq.PAIR
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ socket_type=zmq.PUSH,
+ )
+ assert socket.socket_type == zmq.PUSH
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+
+ socket, _ = BaseZMQ.get_socket(
+ server_id="test-server",
+ socket_id="test-server",
+ node_type="server",
+ context=context,
+ socket_type=zmq.PULL,
+ )
+ assert socket.socket_type == zmq.PULL
+ assert socket.getsockopt_string(zmq.IDENTITY) == "test-server"
+ assert isinstance(socket, zmq.asyncio.Socket)
+ socket.close()
+ context.term()
+
+
+"""
+TODO:
+1. check node_type values
+2. check if TCP socket search happens
+"""
diff --git a/tests/pytests-new/test_03_serializers.py b/tests/pytests-new/test_03_serializers.py
new file mode 100644
index 00000000..eb994fc4
--- /dev/null
+++ b/tests/pytests-new/test_03_serializers.py
@@ -0,0 +1,145 @@
+import pytest
+
+from hololinked.serializers import Serializers
+from hololinked.serializers.serializers import BaseSerializer
+
+from things import TestThing
+
+
+class YAMLSerializer(BaseSerializer):
+ """just a dummy, does not really serialize to YAML"""
+
+ @property
+ def content_type(self):
+ return "application/yaml"
+
+
+@pytest.fixture()
+def yaml_serializer() -> BaseSerializer:
+ # test register a new serializer with content type
+ return YAMLSerializer()
+
+
+def test_1_singleton():
+ """Test the singleton nature of the Serializers class."""
+
+ serializers = Serializers()
+ assert serializers == Serializers()
+ assert Serializers != Serializers()
+ assert isinstance(serializers, Serializers)
+ # all are class attributes
+ assert serializers.json == Serializers.json
+ assert serializers.pickle == Serializers.pickle
+ assert serializers.msgpack == Serializers.msgpack
+ assert serializers.content_types == Serializers.content_types
+ assert serializers.object_content_type_map == Serializers.object_content_type_map
+ assert serializers.object_serializer_map == Serializers.object_serializer_map
+ assert serializers.protocol_serializer_map == Serializers.protocol_serializer_map
+ # check existing serializers are all instances of BaseSerializer
+ for name, serializer in Serializers.content_types.items():
+ assert isinstance(serializer, BaseSerializer)
+ # check default serializer, given that we know its JSON at least for the current test
+ assert serializers.default == Serializers.json
+ assert serializers.default == Serializers.default
+ assert serializers.default == Serializers().json
+ assert serializers.default == Serializers().default
+ # check default content type, given that we know its JSON at least for the current test
+ assert serializers.default_content_type == Serializers.json.content_type
+ # change default to pickle and check if it is set correctly
+ # serializers.default = serializers.pickle
+ # self.assertEqual(serializers.default, Serializers.pickle)
+ # self.assertEqual(Serializers().default, Serializers.pickle)
+
+
+def test_2_protocol_registration(yaml_serializer: BaseSerializer):
+ """i.e. test if a new serializer (protocol) can be registered"""
+
+ # get existing number of serializers
+ num_serializers = len(Serializers.content_types)
+
+ # test register a new serializer
+ base_serializer = BaseSerializer()
+ # register with name
+ with pytest.warns(UserWarning):
+ Serializers.register(base_serializer, "base")
+ # user warning because content type property is not defined
+ # above is same as Serializers.register(base_serializer, 'base')
+
+ # check if name became a class attribute and name can be accessed as an attribute
+ assert "base" in Serializers
+ assert Serializers.base == base_serializer
+ assert Serializers().base == base_serializer
+ # we dont support getitem at instance level yet so we cannot test assertIn
+
+ # since a content type is not set, it should not be in the content types
+ assert base_serializer not in Serializers.content_types.values()
+ # so the length of content types should be the same
+ assert len(Serializers.content_types) == num_serializers
+
+ # register with name
+ Serializers.register(yaml_serializer, "yaml")
+ # check if name became a class attribute and name can be accessed as an attribute
+ assert "yaml" in Serializers
+ assert Serializers.yaml == yaml_serializer
+ assert Serializers().yaml == yaml_serializer
+ # we dont support getitem at instance level yet
+
+ # since a content type is set, it should be in the content types
+ assert yaml_serializer.content_type in Serializers.content_types.keys()
+ assert yaml_serializer in Serializers.content_types.values()
+ # so the length of content types should have increased by 1
+ assert len(Serializers.content_types) == num_serializers + 1
+
+
+def test_3_registration_for_objects():
+ """i.e. test if a new serializer can be registered for a specific property, action or event"""
+ Serializers.register_content_type_for_object(TestThing.base_property, "application/x-pickle")
+ Serializers.register_content_type_for_object(TestThing.action_echo, "application/msgpack")
+ Serializers.register_content_type_for_object(TestThing.test_event, "application/yaml")
+
+ assert Serializers.for_object(None, "TestThing", "action_echo") == Serializers.msgpack
+ assert Serializers.for_object(None, "TestThing", "base_property") == Serializers.pickle
+ assert Serializers.for_object(None, "TestThing", "test_event") == Serializers.yaml
+ assert Serializers.for_object(None, "TestThing", "test_unknown_property") == Serializers.default
+
+
+def test_4_registration_for_objects_by_name():
+ Serializers.register_content_type_for_object_per_thing_instance("test_thing", "base_property", "application/yaml")
+ assert isinstance(Serializers.for_object("test_thing", None, "base_property"), YAMLSerializer)
+
+
+def test_5_registration_dict():
+ """test the dictionary where all serializers are stored"""
+ # depends on test 3
+ assert "test_thing" in Serializers.object_content_type_map
+ assert "base_property" in Serializers.object_content_type_map["test_thing"]
+ assert Serializers.object_content_type_map["test_thing"]["base_property"] == "application/yaml"
+ assert Serializers.object_content_type_map["test_thing"]["base_property"] == "application/yaml"
+
+ assert "action_echo" in Serializers.object_content_type_map["TestThing"]
+ assert Serializers.object_content_type_map["TestThing"]["action_echo"] == "application/msgpack"
+ assert "test_event" in Serializers.object_content_type_map["TestThing"]
+ assert Serializers.object_content_type_map["TestThing"]["test_event"] == "application/yaml"
+
+
+def test_6_retrieval():
+ # added in previous tests
+ assert isinstance(Serializers.for_object("test_thing", None, "base_property"), YAMLSerializer)
+ # unknown object should retrieve the default serializer
+ assert Serializers.for_object("test_thing", None, "test_unknown_property") == Serializers.default
+ # unknown thing should retrieve the default serializer
+ assert Serializers.for_object("test_unknown_thing", None, "base_property") == Serializers.default
+
+
+def test_7_set_default():
+ """test setting the default serializer"""
+ # get existing default
+ old_default = Serializers.default
+ # set new default and check if default is set
+ Serializers.default = Serializers.yaml
+ assert Serializers.default == Serializers.yaml
+ test_6_retrieval() # check if retrieval is consistent with default
+ # reset default and check if default is reset
+ Serializers.default = old_default
+ assert Serializers.default == old_default
+ assert Serializers.default == Serializers.json # because we know its JSON
diff --git a/tests/pytests-new/test_04_thing_init.py b/tests/pytests-new/test_04_thing_init.py
new file mode 100644
index 00000000..c2c395f3
--- /dev/null
+++ b/tests/pytests-new/test_04_thing_init.py
@@ -0,0 +1,778 @@
+import typing
+import unittest
+import pytest
+import logging
+
+from hololinked.core.actions import BoundAction
+from hololinked.core.events import EventDispatcher
+from hololinked.core.zmq.brokers import EventPublisher
+from hololinked.core import Thing, ThingMeta, Action, Event, Property
+from hololinked.core.meta import (
+ DescriptorRegistry,
+ PropertiesRegistry,
+ ActionsRegistry,
+ EventsRegistry,
+)
+from hololinked.core.zmq.rpc_server import RPCServer, prepare_rpc_server
+from hololinked.core.properties import Parameter
+from hololinked.core.state_machine import BoundFSM
+from hololinked.utils import get_default_logger
+from hololinked.core.logger import RemoteAccessHandler
+
+
+from things import OceanOpticsSpectrometer
+
+
+"""
+The tests in this file are for the initialization of the Thing class and its subclasses.
+1. Test Thing class
+2. Test Thing subclass
+3. Test ThingMeta metaclass
+4. Test ActionRegistry class
+5. Test EventRegistry class
+6. Test PropertiesRegistry class
+"""
+
+
+"""
+Test sequence is as follows:
+1. Test id requirements
+2. Test logger setup
+3. Test state and state_machine setup
+4. Test composition of subthings
+5. Test servers init
+6. Test thing model generation
+"""
+
+
+@pytest.mark.parametrize("thing_cls", [Thing, OceanOpticsSpectrometer])
+def test_1_id(thing_cls: ThingMeta):
+ """Test id property of Thing class"""
+ # req. 1. instance name must be a string and cannot be changed after set
+ thing = thing_cls(id="test_id", log_level=logging.WARN) # type: Thing
+ assert thing.id == "test_id"
+ with pytest.raises(ValueError):
+ thing.id = "new_instance"
+ with pytest.raises(NotImplementedError):
+ del thing.id
+ # req. 2. regex is r'[A-Za-z]+[A-Za-z_0-9\-\/]*', simple URI like
+ valid_ids = ["test_id", "A123", "valid_id-123", "another/valid-id"]
+ invalid_ids = ["123_invalid", "invalid id", "invalid@id", ""]
+ for valid_id in valid_ids:
+ thing.properties.descriptors["id"].validate_and_adapt(valid_id)
+ for invalid_id in invalid_ids:
+ with pytest.raises(ValueError):
+ thing.properties.descriptors["id"].validate_and_adapt(invalid_id)
+
+
+@pytest.mark.parametrize("thing_cls", [Thing, OceanOpticsSpectrometer])
+def test_2_logger(thing_cls: ThingMeta):
+ """Test logger setup"""
+ # req. 1. logger must have remote access handler if remote_accessible_logger is True
+ logger = get_default_logger("test_logger", log_level=logging.WARN)
+ thing = thing_cls(
+ id="test_remote_accessible_logger",
+ logger=logger,
+ remote_accessible_logger=True,
+ ) # type: Thing
+ assert thing.logger == logger
+ assert any(isinstance(handler, RemoteAccessHandler) for handler in thing.logger.handlers)
+ # Therefore also check the false condition
+ logger = get_default_logger("test_logger_2", log_level=logging.WARN)
+ thing = thing_cls(
+ id="test_logger_without_remote_access",
+ logger=logger,
+ remote_accessible_logger=False,
+ ) # type: Thing
+ assert not any(isinstance(handler, RemoteAccessHandler) for handler in thing.logger.handlers)
+ # NOTE - logger is modifiable after instantiation
+
+ # req. 2. logger is created automatically if not provided
+ thing = thing_cls(id="test_logger_auto_creation", log_level=logging.WARN)
+ assert thing.logger is not None
+ assert not any(isinstance(handler, RemoteAccessHandler) for handler in thing.logger.handlers)
+ assert thing.logger != logger # not the above logger that we used.
+ # remote accessible only when we ask for it
+ thing = thing_cls(
+ id="test_logger_auto_creation_2",
+ log_level=logging.WARN,
+ remote_accessible_logger=True,
+ ) # type: Thing
+ assert thing.logger is not None
+ assert any(isinstance(handler, RemoteAccessHandler) for handler in thing.logger.handlers)
+ assert thing.logger != logger
+
+
+@pytest.mark.parametrize("thing_cls", [Thing])
+def test_3_has_no_fsm(thing_cls: ThingMeta):
+ """Test state and state_machine setup"""
+ # req. 1. state property must be None when no state machine is present
+ thing = thing_cls(id="test_no_state_machine", log_level=logging.WARN) # type: Thing
+ if thing.state_machine is None:
+ assert thing.state is None
+ assert thing.state_machine is None
+
+
+@pytest.mark.parametrize("thing_cls", [OceanOpticsSpectrometer])
+def test_4_bound_fsm(thing_cls: ThingMeta):
+ """Test state and state_machine setup"""
+ thing1 = thing_cls(id="test_state_machine", log_level=logging.WARN) # type: Thing
+ # req. 1. state and state machine must be present because we create this subclass with a state machine
+ assert thing1.state is not None
+ assert isinstance(thing1.state_machine, BoundFSM)
+ # req. 2. state and state machine must be different for different instances
+ thing2 = thing_cls(id="test_state_machine_2", log_level=logging.WARN) # type: Thing
+ # first check if state machine exists
+ assert thing2.state is not None
+ assert isinstance(thing2.state_machine, BoundFSM)
+ # then check if they are different
+ assert thing1.state_machine != thing2.state_machine
+ # until state is set, initial state is equal
+ assert thing1.state == thing2.state
+ assert thing1.state_machine.initial_state == thing2.state_machine.initial_state
+ # after state is set, they are different
+ thing1.state_machine.set_state(thing1.states.ALARM)
+ assert thing1.state != thing2.state
+ assert thing1.state_machine != thing2.state_machine
+ # initial state is still same
+ assert thing1.state_machine.initial_state == thing2.state_machine.initial_state
+ # detailed checks in another file
+
+
+@pytest.mark.parametrize("thing_cls", [Thing, OceanOpticsSpectrometer])
+def test_5_subthings(thing_cls: ThingMeta):
+ """Test object composition"""
+ thing = thing_cls(id="test_subthings", log_level=logging.WARN, remote_accessible_logger=True) # type: Thing
+ # req. 1. subthings must be a dictionary
+ assert isinstance(thing.sub_things, dict)
+ assert len(thing.sub_things) == 1 # logger
+ # req. 2. subthings are always recomputed when accessed (at least thats the way it is right now),
+ # so we can add new subthings anytime
+ thing.another_thing = OceanOpticsSpectrometer(id="another_thing", log_level=logging.WARN)
+ assert isinstance(thing.sub_things, dict)
+ assert len(thing.sub_things) == 2 # logger + another_thing
+ # req. 3. subthings must be instances of Thing and have the parent as owner
+ for name, subthing in thing.sub_things.items():
+ assert thing in subthing._owners # type: ignore[attr-defined]
+ assert isinstance(subthing, Thing)
+ # req. 4. name of subthing must match name of the attribute
+ assert hasattr(thing, name)
+
+
+@pytest.mark.parametrize("thing_cls", [Thing, OceanOpticsSpectrometer])
+def test_5_servers_init(thing_cls: ThingMeta):
+ """Test if servers can be initialized/instantiated"""
+ # req. 1. rpc_server and event_publisher must be None when not run()
+ thing = thing_cls(id="test_servers_init", log_level=logging.ERROR) # type: Thing
+ assert thing.rpc_server is None
+ assert thing.event_publisher is None
+ # req. 2. rpc_server and event_publisher must be instances of their respective classes when run()
+ prepare_rpc_server(thing, "IPC")
+ assert isinstance(thing.rpc_server, RPCServer)
+ assert isinstance(thing.event_publisher, EventPublisher)
+ # exit to quit nicely
+ thing.rpc_server.exit()
+ thing.event_publisher.exit()
+
+
+"""
+Test sequence is as follows:
+1. Test metaclass of Thing class
+2. Test registry creation and access which is currently the main purpose of the metaclass
+"""
+
+@pytest.mark.parametrize("thing_cls", [Thing, OceanOpticsSpectrometer])
+def test_6_metaclass_assigned(thing_cls: ThingMeta):
+ """test metaclass of Thing class"""
+ # req. 1 metaclass must be ThingMeta of any Thing class
+ assert thing_cls.__class__ == ThingMeta
+ assert OceanOpticsSpectrometer.__class__ == ThingMeta
+ assert Thing.__class__ == OceanOpticsSpectrometer.__class__
+
+
+def test_7_registry_creation():
+ """test registry creation and access which is currently the main purpose of the metaclass"""
+ # req. 1. registry attributes must be instances of their respective classes
+ assert isinstance(Thing.properties, PropertiesRegistry)
+ assert isinstance(Thing.actions, ActionsRegistry)
+ assert isinstance(Thing.events, EventsRegistry)
+
+ # req. 2. new registries are not created on the fly and are same between accesses
+ assert Thing.properties == Thing.properties
+ assert Thing.actions == Thing.actions
+ assert Thing.events == Thing.events
+ # This test is done as the implementation deviates from `param`
+
+ # req. 3. different subclasses have different registries
+ assert Thing.properties != OceanOpticsSpectrometer.properties
+ assert Thing.actions != OceanOpticsSpectrometer.actions
+ assert Thing.events != OceanOpticsSpectrometer.events
+
+ # create instances for further tests
+ thing = Thing(id="test_registry_creation", log_level=logging.WARN)
+ spectrometer = OceanOpticsSpectrometer(id="test_registry_creation_2", log_level=logging.WARN)
+
+ # req. 4. registry attributes must be instances of their respective classes also for instances
+ assert isinstance(thing.properties, PropertiesRegistry)
+ assert isinstance(thing.actions, ActionsRegistry)
+ assert isinstance(thing.events, EventsRegistry)
+
+ # req. 5. registries are not created on the fly and are same between accesses also for instances
+ assert thing.properties == thing.properties
+ assert thing.actions == thing.actions
+ assert thing.events == thing.events
+
+ # req. 6. registries are not shared between instances
+ assert thing.properties != spectrometer.properties
+ assert thing.actions != spectrometer.actions
+ assert thing.events != spectrometer.events
+
+ # req. 7. registries are not shared between instances and their classes
+ assert thing.properties != Thing.properties
+ assert thing.actions != Thing.actions
+ assert thing.events != Thing.events
+ assert spectrometer.properties != OceanOpticsSpectrometer.properties
+ assert spectrometer.actions != OceanOpticsSpectrometer.actions
+ assert spectrometer.events != OceanOpticsSpectrometer.events
+
+
+# # Uncomment the following for type hints while coding registry tests,
+# # comment it before testing, otherwise tests will fail due to overriding Thing object
+# # class Thing(Thing):
+# # class_registry: PropertiesRegistry | ActionsRegistry | EventsRegistry
+# # instance_registry: PropertiesRegistry | ActionsRegistry | EventsRegistry | None
+# # descriptor_object: type[Property | Action | Event]
+
+# # class OceanOpticsSpectrometer(OceanOpticsSpectrometer):
+# # class_registry: PropertiesRegistry | ActionsRegistry | EventsRegistry
+# # instance_registry: PropertiesRegistry | ActionsRegistry | EventsRegistry | None
+# # descriptor_object: type[Property | Action | Event]
+
+"""
+Test action registry first because actions are the easiest to test.
+1. Test owner attribute
+2. Test descriptors access
+3. Test dunders
+"""
+
+def setup_registry_tests():
+
+ # create instances for further tests
+ cls.thing = Thing(id=f"test_{cls.registry_object.__name__}_registry", log_level=logging.WARN)
+ cls.spectrometer = OceanOpticsSpectrometer(
+ id=f"test_{cls.registry_object.__name__}_registry", log_level=logging.WARN
+ )
+ if cls.registry_cls == ActionsRegistry:
+ Thing.class_registry = Thing.actions
+ OceanOpticsSpectrometer.class_registry = OceanOpticsSpectrometer.actions
+ cls.thing.instance_registry = cls.thing.actions
+ cls.spectrometer.instance_registry = cls.spectrometer.actions
+ cls.bound_object = BoundAction
+ elif cls.registry_cls == PropertiesRegistry:
+ Thing.class_registry = Thing.properties
+ OceanOpticsSpectrometer.class_registry = OceanOpticsSpectrometer.properties
+ cls.thing.instance_registry = cls.thing.properties
+ cls.spectrometer.instance_registry = cls.spectrometer.properties
+ cls.bound_object = typing.Any
+ elif cls.registry_cls == EventsRegistry:
+ Thing.class_registry = Thing.events
+ OceanOpticsSpectrometer.class_registry = OceanOpticsSpectrometer.events
+ cls.thing.instance_registry = cls.thing.events
+ cls.spectrometer.instance_registry = cls.spectrometer.events
+ cls.bound_object = EventDispatcher
+ else:
+ raise NotImplementedError("This registry class is not implemented")
+ yield
+
+
+
+def test_8_registry_owner():
+ """Test owner attribute of DescriptorRegistry"""
+ # See comment above TestRegistry class to enable type definitions
+ # req. 1. owner attribute must be the class itself when accessed as class attribute
+ assert Thing.class_registry.owner == Thing
+ assert OceanOpticsSpectrometer.class_registry.owner == OceanOpticsSpectrometer
+ # therefore owner instance must be None
+ assert Thing.class_registry.owner_inst is None
+ assert OceanOpticsSpectrometer.class_registry.owner_inst is None
+
+ # req. 2. owner attribute must be the instance for instance registries (i.e. when accessed as instance attribute)
+ assert self.thing.instance_registry.owner == self.thing
+ assert self.spectrometer.instance_registry.owner == self.spectrometer
+ assert self.thing.instance_registry.owner_cls == Thing
+ assert self.spectrometer.instance_registry.owner_cls == OceanOpticsSpectrometer
+
+ # req. 3. descriptor_object must be defined correctly and is a class
+ assert Thing.class_registry.descriptor_object == self.registry_object
+ assert OceanOpticsSpectrometer.class_registry.descriptor_object == self.registry_object
+ assert self.thing.instance_registry.descriptor_object == self.registry_object
+ assert self.spectrometer.instance_registry.descriptor_object == self.registry_object
+ self.thing.instance_registry.descriptor_object,
+ Thing.class_registry.descriptor_object,
+ )
+
+
+# def test_2_descriptors(self):
+# """Test descriptors access"""
+# if self.is_abstract_test_class:
+# return
+
+# # req. 1. descriptors are instances of the descriptor object - Property | Action | Event
+# for name, value in Thing.class_registry.descriptors.items():
+# self.assertIsInstance(value, self.registry_object)
+# self.assertIsInstance(name, str)
+# for name, value in OceanOpticsSpectrometer.class_registry.descriptors.items():
+# self.assertIsInstance(value, self.registry_object)
+# self.assertIsInstance(name, str)
+# # subclass have more descriptors than parent class because our example Thing OceanOpticsSpectrometer
+# # has defined its own actions, properties and events
+# self.assertTrue(len(OceanOpticsSpectrometer.class_registry.descriptors) > len(Thing.class_registry.descriptors))
+# # req. 2. either class level or instance level descriptors are same - not a strict requirement for different
+# # use cases, one can always add instance level descriptors
+# for name, value in self.thing.instance_registry.descriptors.items():
+# self.assertIsInstance(value, self.registry_object)
+# self.assertIsInstance(name, str)
+# for name, value in self.spectrometer.instance_registry.descriptors.items():
+# self.assertIsInstance(value, self.registry_object)
+# self.assertIsInstance(name, str)
+# # req. 3. because class level and instance level descriptors are same, they are equal
+# for (name, value), (name2, value2) in zip(
+# Thing.class_registry.descriptors.items(),
+# self.thing.instance_registry.descriptors.items(),
+# ):
+# self.assertEqual(name, name2)
+# self.assertEqual(value, value2)
+# for (name, value), (name2, value2) in zip(
+# OceanOpticsSpectrometer.class_registry.descriptors.items(),
+# self.spectrometer.instance_registry.descriptors.items(),
+# ):
+# self.assertEqual(name, name2)
+# self.assertEqual(value, value2)
+# # req. 4. descriptors can be cleared
+# self.assertTrue(
+# hasattr(
+# self.thing.instance_registry,
+# f"_{self.thing.instance_registry._qualified_prefix}_{self.registry_cls.__name__.lower()}",
+# )
+# )
+# self.thing.instance_registry.clear()
+# self.assertTrue(
+# not hasattr(
+# self.thing.instance_registry,
+# f"_{self.thing.instance_registry._qualified_prefix}_{self.registry_cls.__name__.lower()}",
+# )
+# )
+# # clearing again any number of times should not raise error
+# self.thing.instance_registry.clear()
+# self.thing.instance_registry.clear()
+# self.assertTrue(
+# not hasattr(
+# self.thing.instance_registry,
+# f"_{self.thing.instance_registry._qualified_prefix}_{self.registry_cls.__name__.lower()}",
+# )
+# )
+
+# def test_3_dunders(self):
+# """Test dunders of DescriptorRegistry"""
+# if self.is_abstract_test_class:
+# return
+
+# # req. 1. __getitem__ must return the descriptor object
+# for name, value in Thing.class_registry.descriptors.items():
+# self.assertEqual(Thing.class_registry[name], value)
+# # req. 2. __contains__ must return True if the descriptor is present
+# self.assertIn(value, Thing.class_registry)
+# self.assertIn(name, Thing.class_registry.descriptors.keys())
+
+# # req. 2. __iter__ must return an iterator over the descriptors dictionary
+# # which in turn iterates over the keys
+# self.assertTrue(all(isinstance(descriptor_name, str) for descriptor_name in Thing.class_registry))
+# self.assertTrue(
+# all(isinstance(descriptor_name, str) for descriptor_name in OceanOpticsSpectrometer.class_registry)
+# )
+# # __iter__ can also be casted as other iterators like lists
+# thing_descriptors = list(self.thing.instance_registry)
+# spectrometer_descriptors = list(self.spectrometer.instance_registry)
+# self.assertIsInstance(thing_descriptors, list)
+# self.assertIsInstance(spectrometer_descriptors, list)
+# self.assertTrue(all(isinstance(descriptor_name, str) for descriptor_name in thing_descriptors))
+# self.assertTrue(all(isinstance(descriptor_name, str) for descriptor_name in spectrometer_descriptors))
+
+# # req. 3. __len__ must return the number of descriptors
+# self.assertTrue(len(Thing.class_registry) == len(Thing.class_registry.descriptors))
+# self.assertTrue(
+# len(OceanOpticsSpectrometer.class_registry) == len(OceanOpticsSpectrometer.class_registry.descriptors)
+# )
+# self.assertTrue(len(self.thing.instance_registry) == len(self.thing.instance_registry.descriptors))
+# self.assertTrue(
+# len(self.spectrometer.instance_registry) == len(self.spectrometer.instance_registry.descriptors)
+# )
+# self.assertTrue(len(self.thing.instance_registry) == len(Thing.class_registry))
+# self.assertTrue(len(self.spectrometer.instance_registry) == len(OceanOpticsSpectrometer.class_registry))
+
+# # req. 4. registries have their unique hashes
+# # NOTE - not sure if this is really a useful feature or just plain stupid
+# # The requirement was to be able to generate unique hashes for each registry like foodict[