Finally found a way to make tests work with pytests. Using context ma…

…nager and the right usage of the AsyncGenerator in the tests themselves. Now I have to validate th etests and fix bugs

tests/__init__.py

@@ -0,0 +1,3 @@
+import pytest
+
+pytestmark = pytest.mark.asyncio(scope="package")

tests/conftest.py

@@ -6,10 +6,13 @@
 """
 
 import asyncio
+import time
 from collections import namedtuple
+from signal import SIGINT, SIGTERM, signal
 
 import pytest
 import pytest_asyncio
+from pytest_asyncio import is_async_test
 
 import BAC0
 from BAC0.core.devices.local.factory import (
@@ -29,6 +32,20 @@
     multistate_value,
 )
 
+bacnet = None
+device_app = None
+device30_app = None
+test_device = None
+test_device_30 = None
+# loop = asyncio.get_running_loop()
+
+
+def pytest_collection_modifyitems(items):
+    pytest_asyncio_tests = (item for item in items if is_async_test(item))
+    session_scope_marker = pytest.mark.asyncio(scope="session")
+    for async_test in pytest_asyncio_tests:
+        async_test.add_marker(session_scope_marker, append=False)
+
 
 def add_points(qty_per_type, device):
     # Start from fresh
@@ -72,64 +89,77 @@ def add_points(qty_per_type, device):
     _new_objects.add_objects_to_application(device)
 
 
-# @pytest.mark.asyncio
-# @pytest_asyncio.fixture(scope="session")
-@pytest.fixture(scope="session")
-def network_and_devices():
-    loop = asyncio.get_event_loop()
-    # assert event_loop is asyncio.get_running_loop()
-    # This is the BACnet network and the "client" instance used to interact with
-    # devices that will be created.
-    loop.run_until_complete(BAC0.lite())
-
-
-"""   bacnet = BAC0.lite()
-    #await bacnet.iam()
-
-    # We'll use 3 devices with our first instance
-    device_app = BAC0.lite(port=47809, deviceId=101)
-    #await device_app.iam()
-    device30_app = BAC0.lite(port=47810, deviceId=102)
-    #await device30_app.iam()
-    # device300_app = BAC0.lite(port=47811, deviceId=103)
-    # time.sleep(0.01)
-
-    add_points(2, device_app)
-    add_points(5, device30_app)
-    # add_points(10, device300_app)
-
-    ip = device_app.localIPAddr.addrTuple[0]
-    boid = device_app.Boid
-
-    ip_30 = device30_app.localIPAddr.addrTuple[0]
-    boid_30 = device30_app.Boid
-
-    # ip_300 = device300_app.localIPAddr.addrTuple[0]
-    # boid_300 = device300_app.Boid
-
-    # Connect to test device using main network
-    test_device = BAC0.device("{}:47809".format(ip), boid, bacnet, poll=10)
-    test_device_30 = BAC0.device("{}:47810".format(ip_30), boid_30, bacnet, poll=0)
-    # test_device_300 = BAC0.device("{}:47811".format(ip_300), boid_300, bacnet, poll=0)
-
-    params = namedtuple(
-        "devices",
-        ["bacnet", "device_app", "test_device", "test_device_30", "test_device_300"],
-    )
-    params.bacnet = bacnet
-    params.device_app = device_app
-    params.test_device = test_device
-    params.test_device_30 = test_device_30
-    params.test_device_300 = None
-
-    yield params
+class NetworkAndDevices:
+    def __init__(
+        self, loop, bacnet, device_app, device30_app, test_device, test_device_30
+    ):
+        self.loop = loop
+        self.bacnet = bacnet
+        self.device_app = device_app
+        self.device30_app = device30_app
+        self.test_device = test_device
+        self.test_device_30 = test_device_30
+
+    def __repr__(self):
+        return "NetworkAndDevices({!r}, {!r}, {!r}, {!r}, {!r}, {!r})".format(
+            self.loop,
+            self.bacnet,
+            self.device_app,
+            self.device30_app,
+            self.test_device,
+            self.test_device_30,
+        )
 
-    # Close when done
-    params.test_device.disconnect()
-    params.test_device_30.disconnect()
-    # params.test_device_300.disconnect()
 
-    params.bacnet.disconnect()
-    # If too quick, we may encounter socket issues...
+@pytest.fixture(scope="session")
+async def network_and_devices():
+    global loop
+    global bacnet
+    global device_app
+    global device30_app
+    global test_device
+    global test_device_30
+
+    loop = asyncio.get_running_loop()
+    # await create_applications()
+
+    # Create BAC0.lite() instances
+    # bacnet = BAC0.lite()
+    # device_app = BAC0.lite(port=47809)
+    # device30_app = BAC0.lite(port=47810)
+
+    # Wait for the instances to be initialized
+    # while not bacnet._initialized or not device_app._initialized or not device30_app._initialized:
+    #    #await asyncio.sleep(0.1)  # Sleep for a short time to prevent busy waiting
+    #    time.sleep(0.1)
+
+    # Once the instances are initialized, yield them to the test functions
+    async with BAC0.lite(localObjName="bacnet") as bacnet:
+        async with BAC0.lite(port=47809, localObjName="device_app") as device_app:
+            async with BAC0.lite(port=47810,localObjName="device30_app") as device30_app:
+                #await asyncio.sleep(5) # let all the objects be valid before continuing
+                add_points(2, device_app)
+                add_points(5, device30_app)
+                # add_points(10, device300_app)
+
+                ip = device_app.localIPAddr.addrTuple[0]
+                boid = device_app.Boid
+
+                ip_30 = device30_app.localIPAddr.addrTuple[0]
+                boid_30 = device30_app.Boid
+                # Connect to test device using main network
+                test_device = BAC0.device("{}:47809".format(ip), boid, bacnet, poll=10)
+                test_device_30 = BAC0.device(
+                    "{}:47810".format(ip_30), boid_30, bacnet, poll=0
+                )
+                t1 = test_device.creation_task
+                t2 = test_device_30.creation_task
+                await asyncio.gather(t1, t2)
+                # Wait for the instances to be initialized
+                net_and_dev = (
+                    loop, bacnet, device_app, device30_app, test_device, test_device_30
+                )
+                yield net_and_dev
+                await test_device._disconnect(save_on_disconnect=False)
+                await test_device_30._disconnect(save_on_disconnect=False)
 
-"""

tests/test_Read.py

@@ -1,7 +1,8 @@
 #!/usr/BIn/env python
 # -*- coding utf-8 -*-
+from typing import AsyncGenerator
 import pytest
-
+import asyncio
 """
 Test Bacnet communication with another device
 """
@@ -10,54 +11,42 @@
 CHANGE_DELTA_AO = 89.90
 CHANGE_DELTA_AV = 79.90
 TOLERANCE = 0.01
-BINARY_TEST_STATE = "inactive"
+BINARY_TEST_STATE_STR1 = "inactive"
+BINARY_TEST_STATE_STR2 = "0:inactive"
+BINARY_TEST_STATE_BOOL = False
 CHARACTERSTRINGVALUE = "test"
 
 
 @pytest.mark.asyncio
-def test_ReadAV(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert (test_device["AV"] - CHANGE_DELTA_AV) < TOLERANCE
-
-
-@pytest.mark.asyncio
-def test_ReadMV(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert test_device["MSV"].value == 1
-    assert test_device["BIG-ALARM"] == "Normal"
-
-
-@pytest.mark.asyncio
-def test_ReadBV(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert test_device["BV"].value == BINARY_TEST_STATE
+async def test_ReadAnalog(network_and_devices:AsyncGenerator):
+    async for resources in network_and_devices:
+        loop, bacnet, device_app, device30_app, test_device, test_device_30 = resources
 
+        await test_device["AV"].value
+        assert (test_device["AV"].lastValue - CHANGE_DELTA_AV) < TOLERANCE
+
+        #assert not test_device["MSV"] == 1
 
-@pytest.mark.asyncio
-def test_ReadAI(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert (test_device["AI"] - CHANGE_DELTA_AI) < TOLERANCE
-
+        assert test_device["BIG-ALARM"] == "Normal"
+        await test_device["AI"].value
+        assert (test_device["AI"].lastValue - CHANGE_DELTA_AI) < TOLERANCE
+        await test_device["AO"].value
+        assert (test_device["AO"].lastValue - CHANGE_DELTA_AO) < TOLERANCE
 
-@pytest.mark.asyncio
-def test_ReadAO(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert (test_device["AO"] - CHANGE_DELTA_AO) < TOLERANCE
 
+        #assert test_device["CS_VALUE"] == CHARACTERSTRINGVALUE
 
 @pytest.mark.asyncio
-def test_ReadBI(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert test_device["BI"].value == BINARY_TEST_STATE
+async def test_ReadBinary(network_and_devices:AsyncGenerator):
+    async for resources in network_and_devices:
+        loop, bacnet, device_app, device30_app, test_device, test_device_30 = resources
+        await test_device["BV-1"].value
+        #assert test_device["BV-1"] is False
+        print(test_device["BV-1"])
+        #assert test_device["BV-1"] == BINARY_TEST_STATE
+        #assert test_device["CS_VALUE"] == CHARACTERSTRINGVALUE
+        assert test_device["BI"] == BINARY_TEST_STATE_STR1
 
+        assert test_device["BO"] == BINARY_TEST_STATE_STR2
+        assert test_device["BO-1"] == BINARY_TEST_STATE_BOOL
 
-@pytest.mark.asyncio
-def test_ReadBO(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert test_device["BO"].value == BINARY_TEST_STATE
-
-
-@pytest.mark.asyncio
-def test_ReadCharacterstringValue(network_and_devices):
-    test_device = network_and_devices.test_device
-    assert test_device["CS_VALUE"].value == CHARACTERSTRINGVALUE

tests/test_Write.py

@@ -5,40 +5,47 @@
 Test Bacnet communication with another device
 """
 import time
-
+import asyncio
 import pytest
 
 NEWCSVALUE = "New_Test"
 
+@pytest.mark.asyncio
+async def test_WriteAV(network_and_devices):
+    async for resources in network_and_devices:
+        loop, bacnet, device_app, device30_app, test_device, test_device_30 = resources
+        # Write to an object and validate new value is correct
+        old_value = await test_device["AV"].value
+        test_device["AV"] = 11.2
+        await asyncio.sleep(1.5)  # or cache will play a trick on you
+        new_value = await test_device["AV"].value
+        assert (new_value - 11.2) < 0.01
 
-def test_WriteAV(network_and_devices):
-    # Write to an object and validate new value is correct
-    test_device = network_and_devices.test_device
-    old_value = test_device["AV"].value
-    test_device["AV"] = 11.2
-    time.sleep(1.5)  # or cache will play a trick on you
-    new_value = test_device["AV"].value
-    assert (new_value - 11.2) < 0.01
+@pytest.mark.asyncio
+async def test_RelinquishDefault(network_and_devices):
+    async for resources in network_and_devices:
+        loop, bacnet, device_app, device30_app, test_device, test_device_30 = resources
+        test_device = test_device
+        # Write to an object and validate new value is correct
+        old_value = await test_device["AV"].value
+        test_device["AV"].default(90)
+        # time.sleep(1)
+        new_value = await test_device["AV"].value
+        assert (new_value - 90) < 0.01
 
 
-def test_RelinquishDefault(network_and_devices):
+@pytest.mark.asyncio
+async def test_WriteCharStr(network_and_devices):
     # Write to an object and validate new value is correct
-    test_device = network_and_devices.test_device
-    old_value = test_device["AV"].value
-    test_device["AV"].default(90)
-    # time.sleep(1)
-    new_value = test_device["AV"].value
-    assert (new_value - 90) < 0.01
+    async for resources in network_and_devices:
+        loop, bacnet, device_app, device30_app, test_device, test_device_30 = resources
+        test_device = test_device
+        test_device["CS_VALUE"] = NEWCSVALUE
+        # time.sleep(1)
+        new_value = await test_device["CS_VALUE"].value
+        assert new_value == NEWCSVALUE
 
 
-def test_WriteCharStr(network_and_devices):
-    # Write to an object and validate new value is correct
-    test_device = network_and_devices.test_device
-    test_device["CS_VALUE"] = NEWCSVALUE
-    # time.sleep(1)
-    new_value = test_device["CS_VALUE"].value
-    assert new_value == NEWCSVALUE
-
 
 @pytest.mark.skip(
     "Not ready yet as BAC0 do not support out_of_service write -> unlocking PV"