| # | |
| # Copyright (c) 2023 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. | |
| # | |
| import logging | |
| from mobly import asserts | |
| import matter.clusters as Clusters | |
| from matter import ChipDeviceCtrl | |
| from matter.clusters.Types import NullValue | |
| from matter.interaction_model import InteractionModelError, Status | |
| from matter.testing.decorators import async_test_body | |
| from matter.testing.matter_testing import MatterTestCommissionedDevice | |
| from matter.testing.runner import default_matter_test_main | |
| from matter.testing.timeoperations import utc_time_in_matter_epoch | |
| log = logging.getLogger(__name__) | |
| # We don't have a good pipe between the c++ enums in CommissioningDelegate and python | |
| # so this is hardcoded. | |
| # I realize this is dodgy, not sure how to cross the enum from c++ to python cleanly | |
| kCheckForMatchingFabric = 3 | |
| kConfigureUTCTime = 6 | |
| kConfigureTimeZone = 7 | |
| kConfigureDSTOffset = 8 | |
| kConfigureDefaultNTP = 9 | |
| kConfigureTrustedTimeSource = 19 | |
| class TestCommissioningTimeSync(MatterTestCommissionedDevice): | |
| disable_wildcard_subscription = True | |
| def setup_class(self): | |
| self.commissioner = None | |
| self.commissioned = False | |
| return super().setup_class() | |
| async def destroy_current_commissioner(self): | |
| if self.commissioner: | |
| if self.commissioned: | |
| fabricidx = await self.read_single_attribute_check_success(dev_ctrl=self.commissioner, cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.CurrentFabricIndex) | |
| cmd = Clusters.OperationalCredentials.Commands.RemoveFabric(fabricIndex=fabricidx) | |
| await self.send_single_cmd(cmd=cmd) | |
| self.commissioner.Shutdown() | |
| self.commissioner = None | |
| self.commissioned = False | |
| def teardown_test(self): | |
| self.event_loop.run_until_complete(self.destroy_current_commissioner()) | |
| super().teardown_test() | |
| async def commission_and_base_checks(self): | |
| params = await self.default_controller.OpenCommissioningWindow( | |
| nodeId=self.dut_node_id, timeout=600, iteration=10000, discriminator=1234, option=1) | |
| await self.commissioner.CommissionOnNetwork( | |
| nodeId=self.dut_node_id, setupPinCode=params.setupPinCode, | |
| filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=1234) | |
| self.commissioned = True | |
| # Check the feature map - if we have a time cluster, we want UTC time to be set | |
| features = await self.read_single_attribute(dev_ctrl=self.default_controller, node_id=self.dut_node_id, | |
| endpoint=0, attribute=Clusters.TimeSynchronization.Attributes.FeatureMap) | |
| self.supports_time = True | |
| if isinstance(features, Clusters.Attribute.ValueDecodeFailure): | |
| asserts.assert_true(isinstance(features.Reason, InteractionModelError), InteractionModelError, | |
| f'Unexpected exception from reading time cluster feature map {features.Reason}') | |
| asserts.assert_equal(features.Reason.status, Status.UnsupportedCluster, | |
| f'Unexpected error response from reading time cluster feature map {features.Reason}') | |
| self.supports_time = False | |
| asserts.assert_equal(self.commissioner.CheckStageSuccessful( | |
| kConfigureUTCTime), self.supports_time, 'UTC time stage incorrect') | |
| if self.supports_time: | |
| self.supports_time_zone = bool(features & Clusters.TimeSynchronization.Bitmaps.Feature.kTimeZone) | |
| self.supports_default_ntp = bool(features & Clusters.TimeSynchronization.Bitmaps.Feature.kNTPClient) | |
| self.supports_trusted_time_source = bool(features & Clusters.TimeSynchronization.Bitmaps.Feature.kTimeSyncClient) | |
| else: | |
| self.supports_time_zone = False | |
| self.supports_default_ntp = False | |
| self.supports_trusted_time_source = False | |
| async def create_commissioner(self): | |
| if self.commissioner: | |
| await self.destroy_current_commissioner() | |
| new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority() | |
| new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2) | |
| self.commissioner = new_fabric_admin.NewController(nodeId=112233, useTestCommissioner=True) | |
| self.commissioner.ResetCommissioningParameters() | |
| self.commissioner.ResetTestCommissioner() | |
| # If the app we're testing against doesn't have a time cluster, we still want to run this | |
| # tests and we want it to succeed, so catch the unsupported cluster error here and ignore | |
| try: | |
| cmd = Clusters.TimeSynchronization.Commands.SetDefaultNTP(defaultNTP=NullValue) | |
| await self.send_single_cmd(cmd=cmd) | |
| cmd = Clusters.TimeSynchronization.Commands.SetTrustedTimeSource(NullValue) | |
| await self.send_single_cmd(cmd=cmd) | |
| except InteractionModelError as e: | |
| if e.status == Status.UnsupportedCluster: | |
| pass | |
| async def commission_stages(self, time_zone: bool, dst: bool, default_ntp: bool, trusted_time_source: bool): | |
| await self.create_commissioner() | |
| log.info( | |
| 'Running Commissioning test - time_zone: %s, dst: %s, default_ntp: %s, trusted_time_source: %s', time_zone, dst, default_ntp, trusted_time_source) | |
| if time_zone: | |
| self.commissioner.SetTimeZone(offset=3600, validAt=0) | |
| if dst: | |
| six_months = 1.577e+13 # in us | |
| dst_valid_until = utc_time_in_matter_epoch() + int(six_months) | |
| self.commissioner.SetDSTOffset(offset=3600, validStarting=0, validUntil=dst_valid_until) | |
| if default_ntp: | |
| self.commissioner.SetDefaultNTP("fe80::1") | |
| if trusted_time_source: | |
| self.commissioner.SetTrustedTimeSource(self.commissioner.nodeId, 0) | |
| await self.commission_and_base_checks() | |
| should_set_time_zone = bool(self.supports_time_zone and time_zone) | |
| should_set_dst = bool(self.supports_time_zone and time_zone and dst) | |
| should_set_default_ntp = bool(self.supports_default_ntp and default_ntp) | |
| should_set_trusted_time = bool(self.supports_trusted_time_source and trusted_time_source) | |
| asserts.assert_equal(self.commissioner.CheckStageSuccessful(kConfigureTimeZone), | |
| should_set_time_zone, 'Incorrect value for time zone stage check') | |
| asserts.assert_equal(self.commissioner.CheckStageSuccessful(kConfigureDSTOffset), | |
| should_set_dst, 'Incorrect value for kConfigureDSTOffset stage') | |
| asserts.assert_equal(self.commissioner.CheckStageSuccessful(kConfigureDefaultNTP), | |
| should_set_default_ntp, 'Incorrect value for kConfigureDefaultNTP stage') | |
| asserts.assert_equal(self.commissioner.CheckStageSuccessful(kConfigureTrustedTimeSource), | |
| should_set_trusted_time, 'Incorrect value for kConfigureTrustedTimeSource stage') | |
| if should_set_time_zone: | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.TimeZone) | |
| expected = [Clusters.TimeSynchronization.Structs.TimeZoneStruct(offset=3600, validAt=0)] | |
| asserts.assert_equal(received, expected, "Time zone was not correctly set by commissioner") | |
| if should_set_dst: | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.DSTOffset) | |
| expected = [Clusters.TimeSynchronization.Structs.DSTOffsetStruct( | |
| offset=3600, validStarting=0, validUntil=dst_valid_until)] | |
| asserts.assert_equal(received, expected, "DST was not set correctly by the commissioner") | |
| if should_set_trusted_time: | |
| fabric_idx = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.CurrentFabricIndex, dev_ctrl=self.commissioner) | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.TrustedTimeSource) | |
| expected = Clusters.TimeSynchronization.Structs.TrustedTimeSourceStruct( | |
| fabricIndex=fabric_idx, nodeID=self.commissioner.nodeId, endpoint=0) | |
| asserts.assert_equal(received, expected, "Trusted Time source was not set properly") | |
| if should_set_default_ntp: | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.DefaultNTP) | |
| expected = "fe80::1" | |
| asserts.assert_equal(received, expected, "Default NTP was not set properly") | |
| @async_test_body | |
| async def test_CommissioningAllBasic(self): | |
| # We want to assess all combos (ie, all flags in the range of 0b0000 to 0b1111) | |
| for i in range(0, 0xF): | |
| time_zone = bool(i & 0x1) | |
| dst = bool(i & 0x2) | |
| default_ntp = bool(i & 0x4) | |
| trusted_time_source = bool(i & 0x8) | |
| await self.commission_stages(time_zone, dst, default_ntp, trusted_time_source) | |
| @async_test_body | |
| async def test_CommissioningPreSetValues(self): | |
| await self.create_commissioner() | |
| # If we're running this against a node that doesn't have a time cluster, this test doesn't apply | |
| # and the remaining cases are covered in the base case above. | |
| try: | |
| trusted_time_source = Clusters.TimeSynchronization.Structs.FabricScopedTrustedTimeSourceStruct( | |
| nodeID=0x5555, endpoint=0) | |
| cmd = Clusters.TimeSynchronization.Commands.SetTrustedTimeSource(trusted_time_source) | |
| await self.send_single_cmd(cmd) | |
| cmd = Clusters.TimeSynchronization.Commands.SetDefaultNTP("fe80::02") | |
| await self.send_single_cmd(cmd) | |
| except InteractionModelError as e: | |
| if e.status == Status.UnsupportedCluster: | |
| await self.destroy_current_commissioner() | |
| return | |
| self.commissioner.SetTimeZone(offset=3600, validAt=0) | |
| six_months = 1.577e+13 # in us | |
| self.commissioner.SetDSTOffset(offset=3600, validStarting=0, | |
| validUntil=utc_time_in_matter_epoch() + int(six_months)) | |
| self.commissioner.SetDefaultNTP("fe80::1") | |
| self.commissioner.SetTrustedTimeSource(self.commissioner.nodeId, 0) | |
| await self.commission_and_base_checks() | |
| asserts.assert_equal(self.commissioner.CheckStageSuccessful(kConfigureTimeZone), | |
| self.supports_time_zone, 'Incorrect value for time zone stage check') | |
| asserts.assert_equal(self.commissioner.CheckStageSuccessful(kConfigureDSTOffset), | |
| self.supports_time_zone, 'Incorrect value for kConfigureDSTOffset stage') | |
| asserts.assert_false(self.commissioner.CheckStageSuccessful(kConfigureDefaultNTP), 'kConfigureDefaultNTP incorrectly set') | |
| asserts.assert_false(self.commissioner.CheckStageSuccessful( | |
| kConfigureTrustedTimeSource), 'kConfigureTrustedTimeSource incorrectly set') | |
| @async_test_body | |
| async def test_FabricCheckStage(self): | |
| await self.create_commissioner() | |
| # This was moved into a different stage when the time sync stuff was added | |
| asserts.assert_equal(self.commissioner.GetFabricCheckResult(), -1, "Fabric check result is already set") | |
| self.commissioner.SetCheckMatchingFabric(True) | |
| await self.commission_and_base_checks() | |
| asserts.assert_true(self.commissioner.CheckStageSuccessful( | |
| kCheckForMatchingFabric), "Did not run check for matching fabric stage") | |
| asserts.assert_equal(self.commissioner.GetFabricCheckResult(), 0, "Fabric check result did not get set by pairing delegate") | |
| # Let's try it again with no check | |
| await self.create_commissioner() | |
| asserts.assert_equal(self.commissioner.GetFabricCheckResult(), -1, "Fabric check result is already set") | |
| self.commissioner.SetCheckMatchingFabric(False) | |
| await self.commission_and_base_checks() | |
| asserts.assert_false(self.commissioner.CheckStageSuccessful( | |
| kCheckForMatchingFabric), "Incorrectly ran check for matching fabric stage") | |
| asserts.assert_equal(self.commissioner.GetFabricCheckResult(), -1, "Fabric check result incorrectly set") | |
| @async_test_body | |
| async def test_TimeZoneName(self): | |
| await self.create_commissioner() | |
| self.commissioner.SetTimeZone(offset=3600, validAt=0, name="test") | |
| await self.commission_and_base_checks() | |
| asserts.assert_true(self.commissioner.CheckStageSuccessful(kConfigureTimeZone), 'Time zone was not successfully set') | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.TimeZone) | |
| expected = [Clusters.TimeSynchronization.Structs.TimeZoneStruct(offset=3600, validAt=0, name="test")] | |
| asserts.assert_equal(received, expected, "Time zone was not correctly set by commissioner") | |
| await self.create_commissioner() | |
| # name is max 64 per the spec | |
| sixty_five_byte_string = "x" * 65 | |
| self.commissioner.SetTimeZone(offset=3600, validAt=0, name=sixty_five_byte_string) | |
| await self.commission_and_base_checks() | |
| asserts.assert_true(self.commissioner.CheckStageSuccessful(kConfigureTimeZone), 'Time zone was not successfully set') | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.TimeZone) | |
| expected = [Clusters.TimeSynchronization.Structs.TimeZoneStruct(offset=3600, validAt=0, name=None)] | |
| asserts.assert_equal(received, expected, "Commissioner did not ignore too-long name") | |
| await self.create_commissioner() | |
| # name is max 64 per the spec | |
| sixty_four_byte_string = "x" * 64 | |
| self.commissioner.SetTimeZone(offset=3600, validAt=0, name=sixty_four_byte_string) | |
| await self.commission_and_base_checks() | |
| asserts.assert_true(self.commissioner.CheckStageSuccessful(kConfigureTimeZone), 'Time zone was not successfully set') | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.TimeZone) | |
| expected = [Clusters.TimeSynchronization.Structs.TimeZoneStruct(offset=3600, validAt=0, name=sixty_four_byte_string)] | |
| asserts.assert_equal(received, expected, "Time zone 64 byte name was not correctly set") | |
| @async_test_body | |
| async def test_DefaultNtpSize(self): | |
| await self.create_commissioner() | |
| too_long_name = "x." + "x" * 127 | |
| self.commissioner.SetDefaultNTP(too_long_name) | |
| await self.commission_and_base_checks() | |
| asserts.assert_false(self.commissioner.CheckStageSuccessful(kConfigureDefaultNTP), | |
| 'Commissioner attempted to set default NTP to a too long value') | |
| await self.create_commissioner() | |
| just_fits_name = "x." + "x" * 126 | |
| self.commissioner.SetDefaultNTP(just_fits_name) | |
| await self.commission_and_base_checks() | |
| asserts.assert_true(self.commissioner.CheckStageSuccessful(kConfigureDefaultNTP), | |
| 'Commissioner did not correctly set default NTP') | |
| received = await self.read_single_attribute_check_success(cluster=Clusters.TimeSynchronization, attribute=Clusters.TimeSynchronization.Attributes.DefaultNTP) | |
| asserts.assert_equal(received, just_fits_name, 'Commissioner incorrectly set default NTP name') | |
| # TODO(cecille): Test - Add hooks to change the time zone response to indicate no DST is needed | |
| # TODO(cecille): Test - Set commissioningParameters TimeZone and DST list size to > node list size to ensure they get truncated | |
| if __name__ == "__main__": | |
| default_matter_test_main() |