#
#    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 ipaddress
from datetime import timedelta

import chip.clusters as Clusters
from chip.clusters.Types import NullValue
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, utc_time_in_matter_epoch
from mobly import asserts


class TC_TIMESYNC_2_1(MatterBaseTest):
    async def read_ts_attribute_expect_success(self, endpoint, attribute):
        cluster = Clusters.Objects.TimeSynchronization
        return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)

    def pics_TC_TIMESYNC_2_1(self) -> list[str]:
        return ["TIMESYNC.S"]

    @async_test_body
    async def test_TC_TIMESYNC_2_1(self):

        endpoint = self.user_params.get("endpoint", 0)

        self.print_step(1, "Commissioning, already done")
        attributes = Clusters.TimeSynchronization.Attributes

        self.print_step(2, "Read Granularity attribute")
        if self.check_pics("TIMESYNC.S.A0001"):
            granularity_dut = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.Granularity)
            asserts.assert_less(granularity_dut, Clusters.TimeSynchronization.Enums.GranularityEnum.kUnknownEnumValue,
                                "Granularity is not in valid range")
        else:
            asserts.assert_true(False, "Granularity is a mandatory attribute and must be present in the PICS file")

        self.print_step(3, "Read TimeSource")
        if self.check_pics("TIMESYNC.S.A0002"):
            time_source = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.TimeSource)
            asserts.assert_less(time_source, Clusters.TimeSynchronization.Enums.TimeSourceEnum.kUnknownEnumValue,
                                "TimeSource is not in valid range")

        self.print_step(4, "Read TrustedTimeSource")
        if self.check_pics("TIMESYNC.S.A0003"):
            trusted_time_source = await self.read_ts_attribute_expect_success(endpoint=endpoint,
                                                                              attribute=attributes.TrustedTimeSource)
            if trusted_time_source is not NullValue:
                asserts.assert_less_equal(trusted_time_source.fabricIndex, 0xFE,
                                          "FabricIndex for the TrustedTimeSource is out of range")
                asserts.assert_greater_equal(trusted_time_source.fabricIndex, 1,
                                             "FabricIndex for the TrustedTimeSource is out of range")
        elif self.check_pics("TIMESYNC.S.F03"):
            asserts.assert_true(False, "TrustedTimeSource is mandatory if the TSC feature (TIMESYNC.S.F03) is supported")

        self.print_step(5, "Read DefaultNTP")
        if self.check_pics("TIMESYNC.S.A0004"):
            default_ntp = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.DefaultNTP)
            if default_ntp is not NullValue:
                asserts.assert_less_equal(len(default_ntp), 128, "DefaultNTP length must be less than 128")
                # Assume this is a valid web address if it has at least one . in the name
                is_web_addr = default_ntp.find('.') != -1
                try:
                    ipaddress.IPv6Address(default_ntp)
                    is_ip_addr = True
                except ipaddress.AddressValueError:
                    is_ip_addr = False
                    pass
                asserts.assert_true(is_web_addr or is_ip_addr, "Returned DefaultNTP value is not a IP address or web address")
        elif self.check_pics("TIMESYNC.S.F01"):
            asserts.assert_true(False, "DefaultNTP is mandatory if the NTPC (TIMESYNC.S.F01) feature is supported")

        self.print_step(6, "Read TimeZone")
        if self.check_pics("TIMESYNC.S.A0005"):
            tz_dut = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.TimeZone)
            asserts.assert_greater_equal(len(tz_dut), 1, "TimeZone must have at least one entry in the list")
            asserts.assert_less_equal(len(tz_dut), 2, "TimeZone may have a maximum of two entries in the list")
            for entry in tz_dut:
                asserts.assert_greater_equal(entry.offset, -43200, "TimeZone offset is out of range")
                asserts.assert_less_equal(entry.offset, 50400, "TimeZone offset is out of range")
                if entry.name:
                    asserts.assert_less_equal(len(entry.name), 64, "TimeZone name is too long")

            asserts.assert_equal(tz_dut[0].validAt, 0, "TimeZone list first entry must have a 0 ValidAt time")
            if len(tz_dut) > 1:
                asserts.assert_not_equal(tz_dut[1].validAt, 0, "TimeZone list second entry must have a non-zero ValidAt time")
        elif self.check_pics("TIMESYNC.S.F00"):
            asserts.assert_true(False, "TimeZone is mandatory if the TZ (TIMESYNC.S.F00) feature is supported")

        self.print_step(7, "Read DSTOffset")
        if self.check_pics("TIMESYNC.S.A0006"):
            dst_dut = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.DSTOffset)
            last_valid_until = -1
            last_valid_starting = -1
            for dst in dst_dut:
                asserts.assert_greater(dst.validStarting, last_valid_starting,
                                       "DSTOffset list must be sorted by ValidStarting time")
                last_valid_starting = dst.validStarting
                asserts.assert_greater_equal(dst.validStarting, last_valid_until,
                                             "DSTOffset list must have every ValidStarting > ValidUntil of the previous entry")
                last_valid_until = dst.validUntil
                if dst.validUntil is NullValue or dst.validUntil is None:
                    asserts.assert_equal(dst, dst_dut[-1], "DSTOffset list must have Null ValidUntil at the end")
        elif self.check_pics("TIMESYNC.S.F00"):
            asserts.assert_true(False, "DSTOffset is mandatory if the TZ (TIMESYNC.S.F00) feature is supported")

        self.print_step(8, "Read UTCTime")
        if self.check_pics("TIMESYNC.S.A0000"):
            utc_dut = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.UTCTime)
            if utc_dut is NullValue:
                asserts.assert_equal(granularity_dut, Clusters.TimeSynchronization.Enums.GranularityEnum.kNoTimeGranularity)
            else:
                asserts.assert_not_equal(granularity_dut, Clusters.TimeSynchronization.Enums.GranularityEnum.kNoTimeGranularity)
                if granularity_dut is Clusters.TimeSynchronization.Enums.GranularityEnum.kMinutesGranularity:
                    toleranace = timedelta(minutes=10)
                else:
                    toleranace = timedelta(minutes=1)
                delta_us = abs(utc_dut - utc_time_in_matter_epoch())
                delta = timedelta(microseconds=delta_us)
                asserts.assert_less_equal(delta, toleranace, "UTC time is not within tolerance of TH")
        else:
            asserts.assert_true(False, "UTCTime is a mandatory attribute and must be present in the PICS file")

        self.print_step(9, "Read LocalTime")
        if self.check_pics("TIMESYNC.S.A0007"):
            utc_dut = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.UTCTime)
            local_dut = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.LocalTime)
            if utc_dut is NullValue:
                asserts.assert_true(local_dut is NullValue, "LocalTime must be Null if UTC time is Null")
            elif len(dst_dut) == 0:
                asserts.assert_true(local_dut is NullValue, "LocalTime must be Null if the DST table is empty")
            else:
                local_calculated = utc_dut + dst_dut[0].offset + tz_dut[0].offset
                delta_us = abs(local_dut, local_calculated)
                delta = timedelta(microseconds=delta_us)
                toleranace = timedelta(minutes=1)
                asserts.assert_less_equal(delta, toleranace, "Local time caluclation is not within tolerance of calculated value")
        elif self.check_pics("TIMESYNC.S.F00"):
            asserts.assert_true(False, "LocalTime is mandatory if the TZ (TIMESYNC.S.F00) feature is supported")

        self.print_step(10, "Read TimeZoneDatabase")
        if self.check_pics("TIMESYNC.S.A0008"):
            tz_db_dut = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.TimeZoneDatabase)
            asserts.assert_less(tz_db_dut, Clusters.TimeSynchronization.Enums.TimeZoneDatabaseEnum.kUnknownEnumValue,
                                "TimeZoneDatabase is not in valid range")
        elif self.check_pics("TIMESYNC.S.F00"):
            asserts.assert_true(False, "TimeZoneDatabase is mandatory if the TZ (TIMESYNC.S.F00) feature is supported")

        self.print_step(11, "Read NTPServerAvailable")
        if self.check_pics("TIMESYNC.S.A0009"):
            # bool typechecking happens in the test read functions, so all we need to do here is do the read
            await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.NTPServerAvailable)
        elif self.check_pics("TIMESYNC.S.F02"):
            asserts.assert_true(False, "NTPServerAvailable is mandatory if the NTPS (TIMESYNC.S.F02) feature is supported")

        self.print_step(12, "Read TimeZoneListMaxSize")
        if self.check_pics("TIMESYNC.S.A000a"):
            size = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.TimeZoneListMaxSize)
            asserts.assert_greater_equal(size, 1, "TimeZoneListMaxSize must be at least 1")
            asserts.assert_less_equal(size, 2, "TimeZoneListMaxSize must be max 2")
        elif self.check_pics("TIMESYNC.S.F00"):
            asserts.assert_true(False, "TimeZoneListMaxSize is mandatory if the TZ (TIMESYNC.S.F00) feature is supported")

        self.print_step(13, "Read DSTOffsetListMaxSize")
        if self.check_pics("TIMESYNC.S.A000b"):
            size = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.DSTOffsetListMaxSize)
            asserts.assert_greater_equal(size, 1, "DSTOffsetListMaxSize must be at least 1")
        elif self.check_pics("TIMESYNC.S.F00"):
            asserts.assert_true(False, "DSTOffsetListMaxSize is mandatory if the TZ (TIMESYNC.S.F00) feature is supported")

        self.print_step(14, "Read SupportsDNSResolve")
        if self.check_pics("TIMESYNC.S.A0004"):
            # bool typechecking happens in the test read functions, so all we need to do here is do the read
            await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.SupportsDNSResolve)
        elif self.check_pics("TIMESYNC.S.F01"):
            asserts.assert_true(False, "SupportsDNSResolve is mandatory if the NTPC (TIMESYNC.S.F01) feature is supported")


if __name__ == "__main__":
    default_matter_test_main()