diff --git a/src/python_testing/TC_DA_1_2.py b/src/python_testing/TC_DA_1_2.py index 5c0b6dbf154d7c..dde370293eb514 100644 --- a/src/python_testing/TC_DA_1_2.py +++ b/src/python_testing/TC_DA_1_2.py @@ -43,8 +43,8 @@ import chip.clusters as Clusters from chip.interaction_model import InteractionModelError, Status from chip.testing.basic_composition import BasicCompositionTests -from chip.testing.matter_testing import (MatterBaseTest, TestStep, async_test_body, default_matter_test_main, hex_from_bytes, - type_matches) +from chip.testing.conversions import hex_from_bytes +from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches from chip.tlv import TLVReader from cryptography import x509 from cryptography.exceptions import InvalidSignature diff --git a/src/python_testing/TC_DA_1_5.py b/src/python_testing/TC_DA_1_5.py index e8e6ce773c6fec..6ee3c81ac370ad 100644 --- a/src/python_testing/TC_DA_1_5.py +++ b/src/python_testing/TC_DA_1_5.py @@ -40,7 +40,8 @@ import chip.clusters as Clusters from chip import ChipDeviceCtrl from chip.interaction_model import InteractionModelError, Status -from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, hex_from_bytes, type_matches +from chip.testing.conversions import hex_from_bytes +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, type_matches from chip.tlv import TLVReader from cryptography import x509 from cryptography.hazmat.primitives import hashes diff --git a/src/python_testing/TC_DA_1_7.py b/src/python_testing/TC_DA_1_7.py index 6678080a600d6b..ce9399d8b7896c 100644 --- a/src/python_testing/TC_DA_1_7.py +++ b/src/python_testing/TC_DA_1_7.py @@ -41,8 +41,8 @@ from typing import List, Optional import chip.clusters as Clusters -from chip.testing.matter_testing import (MatterBaseTest, TestStep, async_test_body, bytes_from_hex, default_matter_test_main, - hex_from_bytes) +from chip.testing.conversions import bytes_from_hex, hex_from_bytes +from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec diff --git a/src/python_testing/TC_DEMTestBase.py b/src/python_testing/TC_DEMTestBase.py index 49c8c6a6375a3e..64c27916b420c4 100644 --- a/src/python_testing/TC_DEMTestBase.py +++ b/src/python_testing/TC_DEMTestBase.py @@ -19,7 +19,7 @@ import chip.clusters as Clusters from chip.interaction_model import InteractionModelError, Status -from chip.testing.matter_testing import utc_time_in_matter_epoch +from chip.testing.timeoperations import utc_time_in_matter_epoch from mobly import asserts logger = logging.getLogger(__name__) diff --git a/src/python_testing/TC_DGGEN_2_4.py b/src/python_testing/TC_DGGEN_2_4.py index 17b068452cdae6..12a7f01a55c0c4 100644 --- a/src/python_testing/TC_DGGEN_2_4.py +++ b/src/python_testing/TC_DGGEN_2_4.py @@ -40,9 +40,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, async_test_body, default_matter_test_main, - matter_epoch_us_from_utc_datetime, utc_datetime_from_matter_epoch_us, - utc_datetime_from_posix_time_ms) +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main +from chip.testing.timeoperations import utc_datetime_from_matter_epoch_us, utc_datetime_from_posix_time_ms, utc_time_in_matter_epoch from mobly import asserts logger = logging.getLogger(__name__) @@ -104,7 +103,7 @@ async def test_TC_DGGEN_2_4(self): self.print_step("1b", "Write current time to DUT") # Get current time in the correct format to set via command. - th_utc = matter_epoch_us_from_utc_datetime(desired_datetime=None) + th_utc = utc_time_in_matter_epoch(desired_datetime=None) await self.set_time_in_timesync(th_utc) diff --git a/src/python_testing/TC_TIMESYNC_2_1.py b/src/python_testing/TC_TIMESYNC_2_1.py index cb2df9d66a87e3..e272d2c0a592cd 100644 --- a/src/python_testing/TC_TIMESYNC_2_1.py +++ b/src/python_testing/TC_TIMESYNC_2_1.py @@ -42,7 +42,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.testing.matter_testing import (MatterBaseTest, default_matter_test_main, has_attribute, has_cluster, - run_if_endpoint_matches, utc_time_in_matter_epoch) + run_if_endpoint_matches) +from chip.testing.timeoperations import utc_time_in_matter_epoch from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_10.py b/src/python_testing/TC_TIMESYNC_2_10.py index 99ca9f3ac0194b..ee815b33a57fed 100644 --- a/src/python_testing/TC_TIMESYNC_2_10.py +++ b/src/python_testing/TC_TIMESYNC_2_10.py @@ -43,17 +43,12 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, SimpleEventCallback, async_test_body, default_matter_test_main, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, SimpleEventCallback, async_test_body, default_matter_test_main +from chip.testing.timeoperations import get_wait_seconds_from_set_time, utc_time_in_matter_epoch from chip.tlv import uint from mobly import asserts -def get_wait_seconds_from_set_time(set_time_matter_us: int, wait_seconds: int): - seconds_passed = int((utc_time_in_matter_epoch() - set_time_matter_us)/1000000) - return wait_seconds - seconds_passed - - class TC_TIMESYNC_2_10(MatterBaseTest): async def send_set_time_zone_cmd(self, tz: typing.List[Clusters.Objects.TimeSynchronization.Structs.TimeZoneStruct]) -> Clusters.Objects.TimeSynchronization.Commands.SetTimeZoneResponse: ret = await self.send_single_cmd(cmd=Clusters.Objects.TimeSynchronization.Commands.SetTimeZone(timeZone=tz), endpoint=self.endpoint) diff --git a/src/python_testing/TC_TIMESYNC_2_11.py b/src/python_testing/TC_TIMESYNC_2_11.py index 0865bc87f4d1cc..e0b8f4dc2eabed 100644 --- a/src/python_testing/TC_TIMESYNC_2_11.py +++ b/src/python_testing/TC_TIMESYNC_2_11.py @@ -43,8 +43,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, SimpleEventCallback, async_test_body, default_matter_test_main, - get_wait_seconds_from_set_time, type_matches, utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, SimpleEventCallback, async_test_body, default_matter_test_main, type_matches +from chip.testing.timeoperations import get_wait_seconds_from_set_time, utc_time_in_matter_epoch from chip.tlv import uint from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_12.py b/src/python_testing/TC_TIMESYNC_2_12.py index 7677dcf184c0b2..7a1ff03e72cb3f 100644 --- a/src/python_testing/TC_TIMESYNC_2_12.py +++ b/src/python_testing/TC_TIMESYNC_2_12.py @@ -43,8 +43,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, SimpleEventCallback, async_test_body, default_matter_test_main, - get_wait_seconds_from_set_time, type_matches, utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, SimpleEventCallback, async_test_body, default_matter_test_main, type_matches +from chip.testing.timeoperations import get_wait_seconds_from_set_time, utc_time_in_matter_epoch from chip.tlv import uint from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_2.py b/src/python_testing/TC_TIMESYNC_2_2.py index 74f935aac9714e..2053bf0e24e98c 100644 --- a/src/python_testing/TC_TIMESYNC_2_2.py +++ b/src/python_testing/TC_TIMESYNC_2_2.py @@ -40,8 +40,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, async_test_body, compare_time, default_matter_test_main, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main +from chip.testing.timeoperations import compare_time, utc_time_in_matter_epoch from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_4.py b/src/python_testing/TC_TIMESYNC_2_4.py index 76fe9074bec56f..a3b6bac020d54e 100644 --- a/src/python_testing/TC_TIMESYNC_2_4.py +++ b/src/python_testing/TC_TIMESYNC_2_4.py @@ -40,8 +40,8 @@ import chip.clusters as Clusters from chip.interaction_model import InteractionModelError, Status -from chip.testing.matter_testing import (MatterBaseTest, async_test_body, default_matter_test_main, type_matches, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, type_matches +from chip.testing.timeoperations import utc_time_in_matter_epoch from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_5.py b/src/python_testing/TC_TIMESYNC_2_5.py index f363f03ba688d8..9705511236d781 100644 --- a/src/python_testing/TC_TIMESYNC_2_5.py +++ b/src/python_testing/TC_TIMESYNC_2_5.py @@ -40,7 +40,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError, Status -from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, utc_time_in_matter_epoch +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main +from chip.testing.timeoperations import utc_time_in_matter_epoch from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_7.py b/src/python_testing/TC_TIMESYNC_2_7.py index 1c6cea2023d4de..a164cd64809f05 100644 --- a/src/python_testing/TC_TIMESYNC_2_7.py +++ b/src/python_testing/TC_TIMESYNC_2_7.py @@ -42,8 +42,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, async_test_body, compare_time, default_matter_test_main, type_matches, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, type_matches +from chip.testing.timeoperations import compare_time, utc_time_in_matter_epoch from chip.tlv import uint from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_8.py b/src/python_testing/TC_TIMESYNC_2_8.py index 82a77c4c886d17..1a227b579547ef 100644 --- a/src/python_testing/TC_TIMESYNC_2_8.py +++ b/src/python_testing/TC_TIMESYNC_2_8.py @@ -42,8 +42,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, async_test_body, compare_time, default_matter_test_main, type_matches, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, type_matches +from chip.testing.timeoperations import compare_time, utc_time_in_matter_epoch from chip.tlv import uint from mobly import asserts diff --git a/src/python_testing/TC_TIMESYNC_2_9.py b/src/python_testing/TC_TIMESYNC_2_9.py index 4ce83f81ddbd4b..703e3b9d58e664 100644 --- a/src/python_testing/TC_TIMESYNC_2_9.py +++ b/src/python_testing/TC_TIMESYNC_2_9.py @@ -41,8 +41,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError -from chip.testing.matter_testing import (MatterBaseTest, async_test_body, compare_time, default_matter_test_main, type_matches, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, type_matches +from chip.testing.timeoperations import compare_time, utc_time_in_matter_epoch from chip.tlv import uint from mobly import asserts diff --git a/src/python_testing/TC_VALCC_4_4.py b/src/python_testing/TC_VALCC_4_4.py index 8915bd9044734e..3dc1b93efd3e04 100644 --- a/src/python_testing/TC_VALCC_4_4.py +++ b/src/python_testing/TC_VALCC_4_4.py @@ -37,8 +37,8 @@ import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError, Status -from chip.testing.matter_testing import (MatterBaseTest, TestStep, async_test_body, default_matter_test_main, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main +from chip.testing.timeoperations import utc_time_in_matter_epoch from mobly import asserts diff --git a/src/python_testing/TestCommissioningTimeSync.py b/src/python_testing/TestCommissioningTimeSync.py index d4033ea58ba8b6..51d33f2f67344d 100644 --- a/src/python_testing/TestCommissioningTimeSync.py +++ b/src/python_testing/TestCommissioningTimeSync.py @@ -20,7 +20,8 @@ from chip import ChipDeviceCtrl from chip.clusters.Types import NullValue from chip.interaction_model import InteractionModelError, Status -from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main, utc_time_in_matter_epoch +from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main +from chip.testing.timeoperations import utc_time_in_matter_epoch from mobly import asserts # We don't have a good pipe between the c++ enums in CommissioningDelegate and python diff --git a/src/python_testing/TestMatterTestingSupport.py b/src/python_testing/TestMatterTestingSupport.py index f111250481032d..5632834a843ba3 100644 --- a/src/python_testing/TestMatterTestingSupport.py +++ b/src/python_testing/TestMatterTestingSupport.py @@ -22,13 +22,13 @@ import chip.clusters as Clusters from chip.clusters.Types import Nullable, NullValue -from chip.testing.matter_testing import (MatterBaseTest, async_test_body, compare_time, default_matter_test_main, - get_wait_seconds_from_set_time, parse_matter_test_args, type_matches, - utc_time_in_matter_epoch) +from chip.testing.matter_testing import (MatterBaseTest, async_test_body, default_matter_test_main, parse_matter_test_args, + type_matches) from chip.testing.pics import parse_pics, parse_pics_xml from chip.testing.taglist_and_topology_test import (TagProblem, create_device_type_list_for_root, create_device_type_lists, find_tag_list_problems, find_tree_roots, flat_list_ok, get_all_children, get_direct_children_of_root, parts_list_cycles, separate_endpoint_types) +from chip.testing.timeoperations import compare_time, get_wait_seconds_from_set_time, utc_time_in_matter_epoch from chip.tlv import uint from mobly import asserts, signals diff --git a/src/python_testing/matter_testing_infrastructure/BUILD.gn b/src/python_testing/matter_testing_infrastructure/BUILD.gn index f4d92e9496215c..cc803ca9a44d60 100644 --- a/src/python_testing/matter_testing_infrastructure/BUILD.gn +++ b/src/python_testing/matter_testing_infrastructure/BUILD.gn @@ -39,7 +39,9 @@ pw_python_package("chip-testing-module") { "chip/testing/basic_composition.py", "chip/testing/choice_conformance.py", "chip/testing/conformance.py", + "chip/testing/conversions.py", "chip/testing/global_attribute_ids.py", + "chip/testing/matchers.py", "chip/testing/matter_asserts.py", "chip/testing/matter_testing.py", "chip/testing/metadata.py", @@ -48,6 +50,7 @@ pw_python_package("chip-testing-module") { "chip/testing/spec_parsing.py", "chip/testing/taglist_and_topology_test.py", "chip/testing/tasks.py", + "chip/testing/timeoperations.py", ] tests = [ "chip/testing/test_metadata.py", diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/conversions.py b/src/python_testing/matter_testing_infrastructure/chip/testing/conversions.py new file mode 100644 index 00000000000000..20a5d760a1e68d --- /dev/null +++ b/src/python_testing/matter_testing_infrastructure/chip/testing/conversions.py @@ -0,0 +1,122 @@ +# +# Copyright (c) 2025 Project CHIP Authors +# All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Conversion utilities for Matter testing infrastructure. + +This module provides utility functions for converting between different data +representations commonly used in Matter testing. +""" + +from binascii import hexlify, unhexlify + +import chip.clusters as Clusters + + +def bytes_from_hex(hex: str) -> bytes: + """ Converts hex string to bytes, handling various formats (colons, spaces, newlines). + + Examples: + >>> bytes_from_hex("01:ab:cd") + b'\\x01\\xab\\xcd' + >>> bytes_from_hex("01 ab cd") + b'\\x01\\xab\\xcd' + >>> bytes_from_hex("01abcd") + b'\\x01\\xab\\xcd' + >>> bytes_from_hex("01\\nab\\ncd") + b'\\x01\\xab\\xcd' + """ + return unhexlify("".join(hex.replace(":", "").replace(" ", "").split())) + + +def hex_from_bytes(b: bytes) -> str: + """ Converts a bytes object to a hexadecimal string. + + This function performs the inverse operation of bytes_from_hex(). It converts + a bytes object into a continuous hexadecimal string without any separators. + + Args: + b: bytes, the bytes object to convert to hexadecimal + + Returns: + str: A string containing the hexadecimal representation of the bytes, + using lowercase letters a-f for hex digits + + Examples: + >>> hex_from_bytes(b'\\x01\\xab\\xcd') + '01abcd' + >>> hex_from_bytes(bytes([1, 171, 205])) # Same bytes, different notation + '01abcd' + """ + return hexlify(b).decode("utf-8") + + +def format_decimal_and_hex(number): + """ Formats a number showing both decimal and hexadecimal representations. + + Creates a string representation of a number showing both its decimal value + and its hex representation in parentheses. + + Args: + number: int, the number to format + + Returns: + str: A formatted string like "123 (0x7b)" + + Examples: + >>> format_decimal_and_hex(123) + '123 (0x7b)' + >>> format_decimal_and_hex(0) + '0 (0x00)' + >>> format_decimal_and_hex(255) + '255 (0xff)' + """ + return f'{number} (0x{number:02x})' + + +def cluster_id_with_name(id): + """ Formats a Matter cluster ID with its name and numeric representation. + + Uses format_decimal_and_hex() for numeric formatting and looks up cluster name from registry. + Falls back to "Unknown cluster" if ID not recognized. + + Args: + id: int, the Matter cluster identifier + + Returns: + str: A formatted string containing the ID and cluster name + + Examples: + >>> cluster_id_with_name(6) # OnOff cluster + '6 (0x06) OnOff' + >>> cluster_id_with_name(999999) # Unknown cluster + '999999 (0xf423f) Unknown cluster' + >>> cluster_id_with_name("invalid") # Invalid input + 'HERE IS THE PROBLEM' + """ + if id in Clusters.ClusterObjects.ALL_CLUSTERS.keys(): + s = Clusters.ClusterObjects.ALL_CLUSTERS[id].__name__ + else: + s = "Unknown cluster" + try: + return f'{format_decimal_and_hex(id)} {s}' + except (TypeError, ValueError): + return 'HERE IS THE PROBLEM' + + +if __name__ == "__main__": + import doctest + doctest.testmod() diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/matchers.py b/src/python_testing/matter_testing_infrastructure/chip/testing/matchers.py new file mode 100644 index 00000000000000..46e53ddbd2dce0 --- /dev/null +++ b/src/python_testing/matter_testing_infrastructure/chip/testing/matchers.py @@ -0,0 +1,70 @@ +# +# Copyright (c) 2025 Project CHIP Authors +# All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Type matching utilities for Matter testing infrastructure. + +This module provides functionality for validating type compatibility between +received values and expected type specifications. +""" + +import typing + +from chip.tlv import float32, uint + + +def is_type(received_value, desired_type): + """ Checks if a received value matches an expected type. + + Handles unpacking Nullable and Optional types and + compares list value types for non-empty lists. + + Args: + received_value: The value to type check + desired_type: The expected type specification (can be a basic type, Union, + Optional, or List type) + + Returns: + bool: True if the received_value matches the desired_type specification + + Examples: + >>> is_type(42, int) + True + >>> from typing import Optional + >>> is_type(None, Optional[str]) + True + >>> is_type([1,2,3], list[int]) + True + """ + if typing.get_origin(desired_type) == typing.Union: + return any(is_type(received_value, t) for t in typing.get_args(desired_type)) + elif typing.get_origin(desired_type) == list: + if isinstance(received_value, list): + # Assume an empty list is of the correct type + return True if received_value == [] else any(is_type(received_value[0], t) for t in typing.get_args(desired_type)) + else: + return False + elif desired_type == uint: + return isinstance(received_value, int) and received_value >= 0 + elif desired_type == float32: + return isinstance(received_value, float) + else: + return isinstance(received_value, desired_type) + + +if __name__ == "__main__": + import doctest + doctest.testmod() diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py b/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py index 5a6481d3d315a1..d47290a98c627f 100644 --- a/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py +++ b/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py @@ -32,7 +32,7 @@ import time import typing import uuid -from binascii import hexlify, unhexlify +from binascii import unhexlify from dataclasses import asdict as dataclass_asdict from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone @@ -41,7 +41,10 @@ from itertools import chain from typing import Any, Iterable, List, Optional, Tuple -from chip.tlv import float32, uint +import chip.testing.conversions as conversions +import chip.testing.matchers as matchers +import chip.testing.timeoperations as timeoperations +from chip.tlv import uint # isort: off @@ -149,78 +152,6 @@ def get_default_paa_trust_store(root_path: pathlib.Path) -> pathlib.Path: return pathlib.Path.cwd() -def type_matches(received_value, desired_type): - """ Checks if the value received matches the expected type. - - Handles unpacking Nullable and Optional types and - compares list value types for non-empty lists. - """ - if typing.get_origin(desired_type) == typing.Union: - return any(type_matches(received_value, t) for t in typing.get_args(desired_type)) - elif typing.get_origin(desired_type) == list: - if isinstance(received_value, list): - # Assume an empty list is of the correct type - return True if received_value == [] else any(type_matches(received_value[0], t) for t in typing.get_args(desired_type)) - else: - return False - elif desired_type == uint: - return isinstance(received_value, int) and received_value >= 0 - elif desired_type == float32: - return isinstance(received_value, float) - else: - return isinstance(received_value, desired_type) - -# TODO(#31177): Need to add unit tests for all time conversion methods. - - -def utc_time_in_matter_epoch(desired_datetime: Optional[datetime] = None): - """ Returns the time in matter epoch in us. - - If desired_datetime is None, it will return the current time. - """ - if desired_datetime is None: - utc_native = datetime.now(tz=timezone.utc) - else: - utc_native = desired_datetime - # Matter epoch is 0 hours, 0 minutes, 0 seconds on Jan 1, 2000 UTC - utc_th_delta = utc_native - datetime(2000, 1, 1, 0, 0, 0, 0, timezone.utc) - utc_th_us = int(utc_th_delta.total_seconds() * 1000000) - return utc_th_us - - -matter_epoch_us_from_utc_datetime = utc_time_in_matter_epoch - - -def utc_datetime_from_matter_epoch_us(matter_epoch_us: int) -> datetime: - """Returns the given Matter epoch time as a usable Python datetime in UTC.""" - delta_from_epoch = timedelta(microseconds=matter_epoch_us) - matter_epoch = datetime(2000, 1, 1, 0, 0, 0, 0, timezone.utc) - - return matter_epoch + delta_from_epoch - - -def utc_datetime_from_posix_time_ms(posix_time_ms: int) -> datetime: - millis = posix_time_ms % 1000 - seconds = posix_time_ms // 1000 - return datetime.fromtimestamp(seconds, timezone.utc) + timedelta(milliseconds=millis) - - -def compare_time(received: int, offset: timedelta = timedelta(), utc: Optional[int] = None, tolerance: timedelta = timedelta(seconds=5)) -> None: - if utc is None: - utc = utc_time_in_matter_epoch() - - # total seconds includes fractional for microseconds - expected = utc + offset.total_seconds() * 1000000 - delta_us = abs(expected - received) - delta = timedelta(microseconds=delta_us) - asserts.assert_less_equal(delta, tolerance, "Received time is out of tolerance") - - -def get_wait_seconds_from_set_time(set_time_matter_us: int, wait_seconds: int): - seconds_passed = (utc_time_in_matter_epoch() - set_time_matter_us) // 1000000 - return wait_seconds - seconds_passed - - @dataclass class SetupPayloadInfo: filter_type: discovery.FilterType = discovery.FilterType.LONG_DISCRIMINATOR @@ -804,21 +735,6 @@ def get_attribute_string(self, cluster_id: int, attribute_id) -> str: return f"Attribute {attribute_name} ({attribute_id}, 0x{attribute_id:04X})" -def id_str(id): - return f'{id} (0x{id:02x})' - - -def cluster_id_str(id): - if id in Clusters.ClusterObjects.ALL_CLUSTERS.keys(): - s = Clusters.ClusterObjects.ALL_CLUSTERS[id].__name__ - else: - s = "Unknown cluster" - try: - return f'{id_str(id)} {s}' - except TypeError: - return 'HERE IS THE PROBLEM' - - @dataclass class CustomCommissioningParameters: commissioningParameters: CommissioningParameters @@ -1024,19 +940,6 @@ def stack(self) -> ChipStack: return builtins.chipStack -def bytes_from_hex(hex: str) -> bytes: - """Converts any `hex` string representation including `01:ab:cd` to bytes - - Handles any whitespace including newlines, which are all stripped. - """ - return unhexlify("".join(hex.replace(":", "").replace(" ", "").split())) - - -def hex_from_bytes(b: bytes) -> str: - """Converts a bytes object `b` into a hex string (reverse of bytes_from_hex)""" - return hexlify(b).decode("utf-8") - - @dataclass class TestStep: test_plan_number: typing.Union[int, str] @@ -2616,3 +2519,16 @@ def run_tests(test_class: MatterBaseTest, matter_test_config: MatterTestConfig, if not run_tests_no_exit(test_class, matter_test_config, runner.get_loop(), hooks, default_controller, external_stack): sys.exit(1) + + +# TODO(#37537): Remove these temporary aliases after transition period +type_matches = matchers.is_type +utc_time_in_matter_epoch = timeoperations.utc_time_in_matter_epoch +utc_datetime_from_matter_epoch_us = timeoperations.utc_datetime_from_matter_epoch_us +utc_datetime_from_posix_time_ms = timeoperations.utc_datetime_from_posix_time_ms +compare_time = timeoperations.compare_time +get_wait_seconds_from_set_time = timeoperations.get_wait_seconds_from_set_time +bytes_from_hex = conversions.bytes_from_hex +hex_from_bytes = conversions.hex_from_bytes +id_str = conversions.format_decimal_and_hex +cluster_id_str = conversions.cluster_id_with_name diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/timeoperations.py b/src/python_testing/matter_testing_infrastructure/chip/testing/timeoperations.py new file mode 100644 index 00000000000000..7255e2254917fe --- /dev/null +++ b/src/python_testing/matter_testing_infrastructure/chip/testing/timeoperations.py @@ -0,0 +1,176 @@ +# +# Copyright (c) 2025 Project CHIP Authors +# All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Time conversion and validation utilities for Matter testing infrastructure. + +This module provides functionality for handling time-related operations in Matter +testing, particularly focusing on conversions between different time formats and +epochs. + +Examples: + To run the doctests, execute the module as a script: + $ python3 timeoperations.py +""" + +from datetime import datetime, timedelta, timezone +from typing import Optional + +from mobly import asserts + +# TODO(#31177): Need to add unit tests for all time conversion methods. + + +def utc_time_in_matter_epoch(desired_datetime: Optional[datetime] = None): + """ Converts a datetime to microseconds since Matter epoch. + + The Matter epoch is defined as January 1, 2000, 00:00:00 UTC. This function + calculates the number of microseconds beчtween the input datetime and the + Matter epoch. If no datetime is provided, the current UTC time is used. + + Args: + desired_datetime: Optional datetime to convert, uses current UTC if None. + Should be timezone-aware. + + Returns: + int: Microseconds since Matter epoch + + Examples: + >>> from datetime import datetime, timezone + >>> utc_time_in_matter_epoch(datetime(2000, 1, 1, 0, 0, 0, 0, tzinfo=timezone.utc)) + 0 + >>> utc_time_in_matter_epoch(datetime(2000, 1, 1, 0, 0, 0, 1, tzinfo=timezone.utc)) + 1 + """ + if desired_datetime is None: + utc_native = datetime.now(tz=timezone.utc) + else: + utc_native = desired_datetime + # Matter epoch is 0 hours, 0 minutes, 0 seconds on Jan 1, 2000 UTC + utc_th_delta = utc_native - datetime(2000, 1, 1, 0, 0, 0, 0, timezone.utc) + utc_th_us = int(utc_th_delta.total_seconds() * 1000000) + return utc_th_us + + +def utc_datetime_from_matter_epoch_us(matter_epoch_us: int) -> datetime: + """ Converts microseconds since Matter epoch to UTC datetime. + + Inverse of utc_time_in_matter_epoch(). + It converts a microsecond timestamp relative to the Matter epoch + (Jan 1, 2000, 00:00:00 UTC) into a timezone-aware Python datetime object. + + Args: + matter_epoch_us: int, microseconds since Matter epoch + + Returns: + datetime: UTC datetime + + Examples: + >>> utc_datetime_from_matter_epoch_us(0) + datetime.datetime(2000, 1, 1, 0, 0, tzinfo=datetime.timezone.utc) + >>> utc_datetime_from_matter_epoch_us(123456789) + datetime.datetime(2000, 1, 1, 0, 2, 3, 456789, tzinfo=datetime.timezone.utc) + """ + delta_from_epoch = timedelta(microseconds=matter_epoch_us) + matter_epoch = datetime(2000, 1, 1, 0, 0, 0, 0, timezone.utc) + + return matter_epoch + delta_from_epoch + + +def utc_datetime_from_posix_time_ms(posix_time_ms: int) -> datetime: + """ Converts POSIX timestamp in milliseconds to UTC datetime. + + This function converts a POSIX timestamp (milliseconds since January 1, 1970, 00:00:00 UTC) + to a timezone-aware Python datetime object. + + Args: + posix_time_ms: int, Unix timestamp in milliseconds since Jan 1, 1970 + + Returns: + datetime: UTC datetime + + Examples: + >>> utc_datetime_from_posix_time_ms(0) + datetime.datetime(1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc) + >>> utc_datetime_from_posix_time_ms(1609459200000) # 2021-01-01 00:00:00 UTC + datetime.datetime(2021, 1, 1, 0, 0, tzinfo=datetime.timezone.utc) + """ + millis = posix_time_ms % 1000 + seconds = posix_time_ms // 1000 + return datetime.fromtimestamp(seconds, timezone.utc) + timedelta(milliseconds=millis) + + +def compare_time(received: int, offset: timedelta = timedelta(), utc: Optional[int] = None, tolerance: timedelta = timedelta(seconds=5)) -> None: + """ Validates a Matter timestamp against expected time within tolerance. + + Args: + received: int, Matter timestamp in microseconds + offset: timedelta, Optional offset from reference time (default: 0s) + utc: Optional[int], Reference time in Matter microseconds (default: current time) + tolerance: timedelta, Maximum allowed time difference (default: 5s) + + Raises: + AssertionError: If the received time differs from the expected time by + more than the specified tolerance + + Examples: + >>> compare_time(5000000, utc=0, tolerance=timedelta(seconds=5)) # Passes (exactly 5s) + >>> from mobly import signals + >>> try: + ... compare_time(6000000, utc=0, tolerance=timedelta(seconds=5)) + ... except signals.TestFailure as e: + ... print("AssertionError: Received time is out of tolerance") + AssertionError: Received time is out of tolerance + """ + if utc is None: + utc = utc_time_in_matter_epoch() + + # total seconds includes fractional for microseconds + expected = utc + offset.total_seconds() * 1000000 + delta_us = abs(expected - received) + delta = timedelta(microseconds=delta_us) + asserts.assert_less_equal(delta, tolerance, "Received time is out of tolerance") + + +def get_wait_seconds_from_set_time(set_time_matter_us: int, wait_seconds: int): + """ Calculates remaining wait time based on a previously set timestamp. + + This function determines how many seconds remain from an original wait duration, + accounting for time that has already elapsed since a Matter timestamp. Useful + for implementing timeouts or delays that need to account for already elapsed time. + + Args: + set_time_matter_us: int, Start time in Matter microseconds + wait_seconds: int, Total seconds to wait from start time + + Returns: + int: Remaining seconds (negative if period expired) + + Examples: + >>> import unittest.mock + >>> start_time = utc_time_in_matter_epoch(datetime(2000, 1, 1, 0, 0, 0, tzinfo=timezone.utc)) # 0 + >>> with unittest.mock.patch(__name__ + '.utc_time_in_matter_epoch') as mock_time: + ... mock_time.return_value = 2000000 # Simulate current time 2 seconds after start + ... get_wait_seconds_from_set_time(start_time, 5) # 5 - (2000000 / 1e6) = 5 - 2 = 3 + 3 + """ + seconds_passed = (utc_time_in_matter_epoch() - set_time_matter_us) // 1000000 + return wait_seconds - seconds_passed + + +if __name__ == "__main__": + import doctest + doctest.testmod()