Tweaks to evse test plans based on review(issue #31460)

.github/workflows/tests.yaml

@@ -451,6 +451,7 @@ jobs:
                       --target linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test \
                       --target linux-x64-lock-ipv6only-no-ble-no-wifi-tsan-clang-test \
                       --target linux-x64-lit-icd-ipv6only-no-ble-no-wifi-tsan-clang-test \
+                      --target linux-x64-energy-management-ipv6only-no-ble-no-wifi-tsan-clang-test \
                       --target linux-x64-python-bindings \
                       build \
                       --copy-artifacts-to objdir-clone \
@@ -503,6 +504,9 @@ jobs:
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-lock-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-lock-app  --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_DRLK_2_12.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app  --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_AccessChecker.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-lock-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-lock-app  --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_DeviceConformance.py" --script-args "--storage-path admin_storage.json --manual-code 10054912339 --bool-arg ignore_in_progress:True allow_provisional:True --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto  --tests test_TC_IDM_10_2"'
+                  scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-energy-management-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-energy-management-app  --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json --enable-key 000102030405060708090a0b0c0d0e0f" --script "src/python_testing/TC_EEVSE_2_2.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 3840 --passcode 20202021 --hex-arg enableKey:000102030405060708090a0b0c0d0e0f --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
+                  scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-energy-management-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-energy-management-app  --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json --enable-key 000102030405060708090a0b0c0d0e0f" --script "src/python_testing/TC_EEVSE_2_4.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 3840 --passcode 20202021 --hex-arg enableKey:000102030405060708090a0b0c0d0e0f --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
+                  scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-energy-management-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-energy-management-app  --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json --enable-key 000102030405060708090a0b0c0d0e0f" --script "src/python_testing/TC_EEVSE_2_5.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 3840 --passcode 20202021 --hex-arg enableKey:000102030405060708090a0b0c0d0e0f --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --script "src/python_testing/TestMatterTestingSupport.py" --script-args "--trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --script "src/python_testing/TestConformanceSupport.py" --script-args "--trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --script "src/python_testing/TestSpecParsingSupport.py" --script-args "--trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'

examples/energy-management-app/energy-management-common/src/EVSEManufacturerImpl.cpp

@@ -232,8 +232,8 @@ void EVSEManufacturer::FakeReadingsUpdate()
 
     // Update meter values
     // Avoid using floats - so we will do a basic rand() call which will generate a integer value between 0 and RAND_MAX
-    // first compute power as a mean + some random value in range 0 to mPowerRandomness_mW
-    int64_t power = (rand() % gFakeReadingsData.mPowerRandomness_mW);
+    // first compute power as a mean + some random value in range +/- mPowerRandomness_mW
+    int64_t power = ((int64_t) rand() % (2 * gFakeReadingsData.mPowerRandomness_mW)) - gFakeReadingsData.mPowerRandomness_mW;
     power += gFakeReadingsData.mPower_mW; // add in the base power
 
     // TODO call the EPM cluster to send a power reading

src/python_testing/TC_EEVSE_2_2.py

@@ -21,9 +21,9 @@
 
 import chip.clusters as Clusters
 from chip.clusters.Types import NullValue
-from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from matter_testing_support import EventChangeCallback, MatterBaseTest, TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
-from TC_EEVSE_Utils import EEVSEBaseTestHelper, EventChangeCallback
+from TC_EEVSE_Utils import EEVSEBaseTestHelper
 
 logger = logging.getLogger(__name__)
 

src/python_testing/TC_EEVSE_2_4.py

@@ -20,8 +20,8 @@
 
 import chip.clusters as Clusters
 from chip.clusters.Types import NullValue
-from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
-from TC_EEVSE_Utils import EEVSEBaseTestHelper, EventChangeCallback
+from matter_testing_support import EventChangeCallback, MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from TC_EEVSE_Utils import EEVSEBaseTestHelper
 
 logger = logging.getLogger(__name__)
 

src/python_testing/TC_EEVSE_2_5.py

@@ -20,8 +20,8 @@
 import chip.clusters as Clusters
 from chip.clusters.Types import NullValue
 from chip.interaction_model import Status
-from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
-from TC_EEVSE_Utils import EEVSEBaseTestHelper, EventChangeCallback
+from matter_testing_support import EventChangeCallback, MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from TC_EEVSE_Utils import EEVSEBaseTestHelper
 
 logger = logging.getLogger(__name__)
 

src/python_testing/TC_EEVSE_Utils.py

@@ -16,45 +16,14 @@
 
 
 import logging
-import queue
 
 import chip.clusters as Clusters
-from chip.clusters import ClusterObjects as ClusterObjects
-from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction
 from chip.interaction_model import InteractionModelError, Status
 from mobly import asserts
 
 logger = logging.getLogger(__name__)
 
 
-class EventChangeCallback:
-    def __init__(self, expected_cluster: ClusterObjects):
-        self._q = queue.Queue()
-        self._expected_cluster = expected_cluster
-
-    async def start(self, dev_ctrl, node_id: int, endpoint: int):
-        self._subscription = await dev_ctrl.ReadEvent(node_id,
-                                                      events=[(endpoint, self._expected_cluster, True)], reportInterval=(1, 5),
-                                                      fabricFiltered=False, keepSubscriptions=True, autoResubscribe=False)
-        self._subscription.SetEventUpdateCallback(self.__call__)
-
-    def __call__(self, res: EventReadResult, transaction: SubscriptionTransaction):
-        if res.Status == Status.Success and res.Header.ClusterId == self._expected_cluster.id:
-            logging.info(
-                f'Got subscription report for event on cluster {self._expected_cluster}: {res.Data}')
-            self._q.put(res)
-
-    def WaitForEventReport(self, expected_event: ClusterObjects.ClusterEvent):
-        try:
-            res = self._q.get(block=True, timeout=10)
-        except queue.Empty:
-            asserts.fail("Failed to receive a report for the event {}".format(expected_event))
-
-        asserts.assert_equal(res.Header.ClusterId, expected_event.cluster_id, "Expected cluster ID not found in event report")
-        asserts.assert_equal(res.Header.EventId, expected_event.event_id, "Expected event ID not found in event report")
-        return res.Data
-
-
 class EEVSEBaseTestHelper:
 
     async def read_evse_attribute_expect_success(self, endpoint: int = None, attribute: str = ""):
@@ -79,7 +48,7 @@ async def write_user_max_charge(self, endpoint: int = None, user_max_charge: int
         asserts.assert_equal(result[0].Status, Status.Success, "UserMaximumChargeCurrent write failed")
 
     async def send_enable_charge_command(self, endpoint: int = None, charge_until: int = None, timedRequestTimeoutMs: int = 3000,
-                                         min_charge: int = None, max_charge: int = None, expected_status: Status = Status.Success):
+                                         min_charge: int = 6000, max_charge: int = 32000, expected_status: Status = Status.Success):
         try:
             await self.send_single_cmd(cmd=Clusters.EnergyEvse.Commands.EnableCharging(
                 chargingEnabledUntil=charge_until,
@@ -110,34 +79,6 @@ async def send_start_diagnostics_command(self, endpoint: int = None, timedReques
         except InteractionModelError as e:
             asserts.assert_equal(e.status, expected_status, "Unexpected error returned")
 
-    async def send_test_event_triggers(self, enableKey: bytes = None, eventTrigger=0x0099000000000000):
-        # get the test event enable key or assume the default
-        # This can be passed in on command line using
-        #    --hex-arg enableKey:000102030405060708090a0b0c0d0e0f
-        if enableKey is None:
-            if 'enableKey' not in self.matter_test_config.global_test_params:
-                enableKey = bytes([b for b in range(16)])
-            else:
-                enableKey = self.matter_test_config.global_test_params['enableKey']
-
-        try:
-            # GeneralDiagnosics cluster is meant to be on Endpoint 0 (Root)
-            await self.send_single_cmd(endpoint=0,
-                                       cmd=Clusters.GeneralDiagnostics.Commands.TestEventTrigger(
-                                           enableKey,
-                                           eventTrigger)
-                                       )
-
-        except InteractionModelError as e:
-            asserts.fail(f"Unexpected error returned - {e.status}")
-
-    async def check_test_event_triggers_enabled(self):
-        full_attr = Clusters.GeneralDiagnostics.Attributes.TestEventTriggersEnabled
-        cluster = Clusters.Objects.GeneralDiagnostics
-        # GeneralDiagnosics cluster is meant to be on Endpoint 0 (Root)
-        test_event_enabled = await self.read_single_attribute_check_success(endpoint=0, cluster=cluster, attribute=full_attr)
-        asserts.assert_equal(test_event_enabled, True, "TestEventTriggersEnabled is False")
-
     async def send_test_event_trigger_basic(self):
         await self.send_test_event_triggers(eventTrigger=0x0099000000000000)
 

src/python_testing/matter_testing_support.py

@@ -51,6 +51,7 @@
 import chip.native
 from chip import discovery
 from chip.ChipStack import ChipStack
+from chip.clusters import ClusterObjects as ClusterObjects
 from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction
 from chip.exceptions import ChipStackError
 from chip.interaction_model import InteractionModelError, Status
@@ -252,6 +253,34 @@ def name(self) -> str:
         return self._name
 
 
+class EventChangeCallback:
+    def __init__(self, expected_cluster: ClusterObjects):
+        self._q = queue.Queue()
+        self._expected_cluster = expected_cluster
+
+    async def start(self, dev_ctrl, node_id: int, endpoint: int):
+        self._subscription = await dev_ctrl.ReadEvent(node_id,
+                                                      events=[(endpoint, self._expected_cluster, True)], reportInterval=(1, 5),
+                                                      fabricFiltered=False, keepSubscriptions=True, autoResubscribe=False)
+        self._subscription.SetEventUpdateCallback(self.__call__)
+
+    def __call__(self, res: EventReadResult, transaction: SubscriptionTransaction):
+        if res.Status == Status.Success and res.Header.ClusterId == self._expected_cluster.id:
+            logging.info(
+                f'Got subscription report for event on cluster {self._expected_cluster}: {res.Data}')
+            self._q.put(res)
+
+    def WaitForEventReport(self, expected_event: ClusterObjects.ClusterEvent, timeout: int = 10):
+        try:
+            res = self._q.get(block=True, timeout=timeout)
+        except queue.Empty:
+            asserts.fail("Failed to receive a report for the event {}".format(expected_event))
+
+        asserts.assert_equal(res.Header.ClusterId, expected_event.cluster_id, "Expected cluster ID not found in event report")
+        asserts.assert_equal(res.Header.EventId, expected_event.event_id, "Expected event ID not found in event report")
+        return res.Data
+
+
 class InternalTestRunnerHooks(TestRunnerHooks):
 
     def start(self, count: int):
@@ -825,6 +854,35 @@ async def send_single_cmd(
         result = await dev_ctrl.SendCommand(nodeid=node_id, endpoint=endpoint, payload=cmd, timedRequestTimeoutMs=timedRequestTimeoutMs)
         return result
 
+    async def send_test_event_triggers(self, enableKey: bytes = None, eventTrigger: int = 0x0099000000000000):
+        # get the test event enable key or assume the default
+        # This can be passed in on command line using
+        #    --hex-arg enableKey:000102030405060708090a0b0c0d0e0f
+        if enableKey is None:
+            if 'enableKey' not in self.matter_test_config.global_test_params:
+                enableKey = bytes([b for b in range(16)])
+            else:
+                enableKey = self.matter_test_config.global_test_params['enableKey']
+
+        try:
+            # GeneralDiagnostics cluster is meant to be on Endpoint 0 (Root)
+            await self.send_single_cmd(endpoint=0,
+                                       cmd=Clusters.GeneralDiagnostics.Commands.TestEventTrigger(
+                                           enableKey,
+                                           eventTrigger)
+                                       )
+
+        except InteractionModelError as e:
+            asserts.fail(
+                f"Sending TestEventTrigger resulted in Unexpected error. Are they enabled in DUT? Command returned - {e.status}")
+
+    async def check_test_event_triggers_enabled(self):
+        full_attr = Clusters.GeneralDiagnostics.Attributes.TestEventTriggersEnabled
+        cluster = Clusters.Objects.GeneralDiagnostics
+        # GeneralDiagnostics cluster is meant to be on Endpoint 0 (Root)
+        test_event_enabled = await self.read_single_attribute_check_success(endpoint=0, cluster=cluster, attribute=full_attr)
+        asserts.assert_equal(test_event_enabled, True, "TestEventTriggersEnabled is False")
+
     def print_step(self, stepnum: typing.Union[int, str], title: str) -> None:
         logging.info(f'***** Test Step {stepnum} : {title}')