#!/usr/bin/env python3

# Copyright (c) 2022 Project CHIP Authors
#
# 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.

try:
    from matter_idl.matter_idl_parser import CreateParser
except ModuleNotFoundError:
    import os
    import sys
    sys.path.append(os.path.dirname(
        os.path.dirname(os.path.abspath(__file__))))

    from matter_idl.matter_idl_parser import CreateParser

import unittest

from matter_idl.matter_idl_types import (AccessPrivilege, Attribute, AttributeInstantiation, AttributeQuality, AttributeStorage,
                                         Bitmap, Cluster, ClusterSide, Command, CommandQuality, ConstantEntry, DataType, DeviceType,
                                         Endpoint, Enum, Event, EventPriority, EventQuality, Field, FieldQuality, Idl,
                                         ParseMetaData, ServerClusterInstantiation, Struct, StructQuality, StructTag)


def parseText(txt, skip_meta=True):
    return CreateParser(skip_meta=skip_meta).parse(txt)


class TestParser(unittest.TestCase):

    def test_skips_comments(self):
        actual = parseText("""
            // this is a single line comment
            // repeated

            /* This is a C++ comment
               and also whitespace should be ignored
             */
        """)
        expected = Idl()

        self.assertEqual(actual, expected)

    def test_global_enum(self):
        actual = parseText("""
            enum GlobalEnum : ENUM8 {
               kValue1 = 1;
               kOther = 0x12; /* hex numbers tested sporadically */
            }
        """)

        expected = Idl(enums=[
            Enum(name='GlobalEnum', base_type='ENUM8',
                 entries=[
                     ConstantEntry(name="kValue1", code=1),
                     ConstantEntry(name="kOther", code=0x12),
                 ])]
        )
        self.assertEqual(actual, expected)

    def test_global_struct(self):
        actual = parseText("""
            struct Something {
                CHAR_STRING astring = 1;
                optional CLUSTER_ID idlist[] = 2;
                nullable int valueThatIsNullable = 0x123;
                char_string<123> sized_string = 222;
            }
        """)

        expected = Idl(structs=[
            Struct(name='Something',
                   fields=[
                        Field(
                            data_type=DataType(name="CHAR_STRING"), code=1, name="astring", ),
                        Field(data_type=DataType(name="CLUSTER_ID"), code=2, name="idlist",
                              is_list=True, qualities=FieldQuality.OPTIONAL),
                        Field(data_type=DataType(name="int"), code=0x123,
                              name="valueThatIsNullable", qualities=FieldQuality.NULLABLE),
                        Field(data_type=DataType(name="char_string", max_length=123),
                              code=222, name="sized_string"),
                   ])]
        )
        self.assertEqual(actual, expected)

    def test_fabric_scoped_struct(self):
        actual = parseText("""
            fabric_scoped struct FabricStruct {
                CHAR_STRING astring = 1;
                optional CLUSTER_ID idlist[] = 2;
                nullable fabric_sensitive int nullablesensitive = 0x123;
            }
        """)

        expected = Idl(structs=[
            Struct(name='FabricStruct',
                   qualities=StructQuality.FABRIC_SCOPED,
                   fields=[
                        Field(
                            data_type=DataType(name="CHAR_STRING"), code=1, name="astring", ),
                        Field(data_type=DataType(name="CLUSTER_ID"), code=2, name="idlist",
                              is_list=True, qualities=FieldQuality.OPTIONAL),
                        Field(data_type=DataType(name="int"), code=0x123, name="nullablesensitive",
                              qualities=FieldQuality.NULLABLE | FieldQuality.FABRIC_SENSITIVE),
                   ])]
        )
        self.assertEqual(actual, expected)

    def test_cluster_attribute(self):
        actual = parseText("""
            server cluster MyCluster = 0x321 {
                readonly attribute int8u roAttr = 1;
                attribute int32u rwAttr[] = 123;
                readonly nosubscribe attribute int8s nosub[] = 0xaa;
                readonly attribute nullable int8s isNullable = 0xab;
            }
        """)

        expected = Idl(clusters=[
            Cluster(side=ClusterSide.SERVER,
                    name="MyCluster",
                    code=0x321,
                    attributes=[
                        Attribute(qualities=AttributeQuality.READABLE, definition=Field(
                            data_type=DataType(name="int8u"), code=1, name="roAttr")),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int32u"), code=123, name="rwAttr", is_list=True)),
                        Attribute(qualities=AttributeQuality.NOSUBSCRIBE | AttributeQuality.READABLE, definition=Field(
                            data_type=DataType(name="int8s"), code=0xAA, name="nosub", is_list=True)),
                        Attribute(qualities=AttributeQuality.READABLE, definition=Field(
                            data_type=DataType(name="int8s"), code=0xAB, name="isNullable", qualities=FieldQuality.NULLABLE)),
                    ]
                    )])
        self.assertEqual(actual, expected)

    def test_doc_comments(self):
        actual = parseText("""
            /** Documentation for MyCluster */
            server cluster MyCluster = 0x321 {
            }

            /** Documentation for MyCluster #2 */
            client cluster MyCluster = 0x321 {
                /* NOT a doc comment */
                command WithoutArg(): DefaultSuccess = 123;

                /** Some command doc comment */
                command InOutStuff(InParam): OutParam = 222;
            }
        """, skip_meta=False)

        # meta_data may not match but is required for doc comments. Clean it up

        # Metadata parsing varies line/column, so only check doc comments
        self.assertEqual(
            actual.clusters[0].description, "Documentation for MyCluster")
        self.assertEqual(
            actual.clusters[1].description, "Documentation for MyCluster #2")
        self.assertIsNone(actual.clusters[1].commands[0].description)
        self.assertEqual(
            actual.clusters[1].commands[1].description, "Some command doc comment")

    def test_sized_attribute(self):
        actual = parseText("""
            server cluster MyCluster = 1 {
                attribute char_string<11> attr1 = 1;
                attribute octet_string<33> attr2[] = 2;
            }
        """)

        expected = Idl(clusters=[
            Cluster(side=ClusterSide.SERVER,
                    name="MyCluster",
                    code=1,
                    attributes=[
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="char_string", max_length=11), code=1, name="attr1")),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="octet_string", max_length=33), code=2, name="attr2", is_list=True)),
                    ]
                    )])
        self.assertEqual(actual, expected)

    def test_timed_attributes(self):
        actual = parseText("""
            server cluster MyCluster = 1 {
                attribute int32u attr1 = 1;
                timedwrite attribute int32u attr2 = 2;
                attribute int32u attr3 = 3;
                timedwrite attribute octet_string<44> attr4[] = 4;
            }
        """)

        expected = Idl(clusters=[
            Cluster(side=ClusterSide.SERVER,
                    name="MyCluster",
                    code=1,
                    attributes=[
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int32u"), code=1, name="attr1")),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE | AttributeQuality.TIMED_WRITE, definition=Field(
                            data_type=DataType(name="int32u"), code=2, name="attr2")),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int32u"), code=3, name="attr3")),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE | AttributeQuality.TIMED_WRITE, definition=Field(
                            data_type=DataType(name="octet_string", max_length=44), code=4, name="attr4", is_list=True)),
                    ]
                    )])
        self.assertEqual(actual, expected)

    def test_attribute_access(self):
        actual = parseText("""
            server cluster MyCluster = 1 {
                attribute                                      int8s attr1 = 1;
                attribute access()                             int8s attr2 = 2;
                attribute access(read: manage)                 int8s attr3 = 3;
                attribute access(write: administer)            int8s attr4 = 4;
                attribute access(read: operate, write: manage) int8s attr5 = 5;
            }
        """)

        expected = Idl(clusters=[
            Cluster(side=ClusterSide.SERVER,
                    name="MyCluster",
                    code=1,
                    attributes=[
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int8s"), code=1, name="attr1"),
                            readacl=AccessPrivilege.VIEW,
                            writeacl=AccessPrivilege.OPERATE
                        ),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int8s"), code=2, name="attr2"),
                            readacl=AccessPrivilege.VIEW,
                            writeacl=AccessPrivilege.OPERATE
                        ),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int8s"), code=3, name="attr3"),
                            readacl=AccessPrivilege.MANAGE
                        ),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int8s"), code=4, name="attr4"),
                            writeacl=AccessPrivilege.ADMINISTER
                        ),
                        Attribute(qualities=AttributeQuality.READABLE | AttributeQuality.WRITABLE, definition=Field(
                            data_type=DataType(name="int8s"), code=5, name="attr5"),
                            readacl=AccessPrivilege.OPERATE,
                            writeacl=AccessPrivilege.MANAGE
                        ),
                    ]
                    )])
        self.assertEqual(actual, expected)

    def test_cluster_commands(self):
        actual = parseText("""
            server cluster WithCommands = 1 {
                struct FreeStruct {}
                request struct InParam {}
                response struct OutParam = 223 {}

                command WithoutArg(): DefaultSuccess = 123;
                command InOutStuff(InParam): OutParam = 222;
                timed command TimedCommand(InParam): DefaultSuccess = 0xab;
                fabric command FabricScopedCommand(InParam): DefaultSuccess = 0xac;
                fabric Timed command FabricScopedTimedCommand(InParam): DefaultSuccess = 0xad;
            }
        """)
        expected = Idl(clusters=[
            Cluster(side=ClusterSide.SERVER,
                    name="WithCommands",
                    code=1,
                    structs=[
                        Struct(name="FreeStruct", fields=[]),
                        Struct(name="InParam", fields=[],
                               tag=StructTag.REQUEST),
                        Struct(name="OutParam", fields=[],
                               tag=StructTag.RESPONSE, code=223),
                    ],
                    commands=[
                        Command(name="WithoutArg", code=123,
                                input_param=None, output_param="DefaultSuccess"),
                        Command(name="InOutStuff", code=222,
                                input_param="InParam", output_param="OutParam"),
                        Command(name="TimedCommand", code=0xab,
                                input_param="InParam", output_param="DefaultSuccess",
                                qualities=CommandQuality.TIMED_INVOKE),
                        Command(name="FabricScopedCommand", code=0xac,
                                input_param="InParam", output_param="DefaultSuccess",
                                qualities=CommandQuality.FABRIC_SCOPED),
                        Command(name="FabricScopedTimedCommand", code=0xad,
                                input_param="InParam", output_param="DefaultSuccess",
                                qualities=CommandQuality.TIMED_INVOKE | CommandQuality.FABRIC_SCOPED),
                    ],
                    )])
        self.assertEqual(actual, expected)

    def test_cluster_command_access(self):
        actual = parseText("""
            server cluster WithCommands = 1 {
                request struct InParam {}
                response struct OutParam = 4 {}

                command WithoutArg(): DefaultSuccess = 1;
                timed command access(invoke: manage) TimedCommand(InParam): OutParam = 2;
                command access(invoke: administer) OutOnly(): OutParam = 3;
            }
        """)
        expected = Idl(clusters=[
            Cluster(side=ClusterSide.SERVER,
                    name="WithCommands",
                    code=1,
                    structs=[
                        Struct(name="InParam", fields=[],
                               tag=StructTag.REQUEST),
                        Struct(name="OutParam", fields=[],
                               tag=StructTag.RESPONSE, code=4),
                    ],
                    commands=[
                        Command(name="WithoutArg", code=1,
                                invokeacl=AccessPrivilege.OPERATE,
                                input_param=None, output_param="DefaultSuccess"),
                        Command(name="TimedCommand", code=2,
                                input_param="InParam", output_param="OutParam",
                                invokeacl=AccessPrivilege.MANAGE,
                                qualities=CommandQuality.TIMED_INVOKE),
                        Command(name="OutOnly", code=3,
                                input_param=None, output_param="OutParam",
                                invokeacl=AccessPrivilege.ADMINISTER,
                                ),
                    ],
                    )])
        self.assertEqual(actual, expected)

    def test_cluster_enum(self):
        actual = parseText("""
            client cluster WithEnums = 0xab {
                enum TestEnum : ENUM16 {
                    A = 0x123;
                    B = 0x234;
                }
            }
        """)
        expected = Idl(clusters=[
            Cluster(side=ClusterSide.CLIENT,
                    name="WithEnums",
                    code=0xab,
                    enums=[
                        Enum(name="TestEnum", base_type="ENUM16",
                             entries=[
                                 ConstantEntry(name="A", code=0x123),
                                 ConstantEntry(name="B", code=0x234),
                             ])],
                    )])
        self.assertEqual(actual, expected)

    def test_cluster_bitmap(self):
        actual = parseText("""
            client cluster Test = 0xab {
                bitmap TestBitmap : BITMAP32 {
                    kFirst = 0x1;
                    kSecond = 0x2;
                }
            }
        """)
        expected = Idl(clusters=[
            Cluster(side=ClusterSide.CLIENT,
                    name="Test",
                    code=0xab,
                    bitmaps=[
                        Bitmap(name="TestBitmap", base_type="BITMAP32",
                               entries=[
                                   ConstantEntry(name="kFirst", code=0x1),
                                   ConstantEntry(name="kSecond", code=0x2),
                               ])],
                    )])
        self.assertEqual(actual, expected)

    def test_cluster_events(self):
        actual = parseText("""
            client cluster EventTester = 0x123 {
               critical event StartUp = 0 {
                 INT32U softwareVersion = 0;
               }
               info event Hello = 1 {}
               debug event GoodBye = 2 {}
            }
        """)
        expected = Idl(clusters=[
            Cluster(side=ClusterSide.CLIENT,
                    name="EventTester",
                    code=0x123,
                    events=[
                        Event(priority=EventPriority.CRITICAL, name="StartUp", code=0, fields=[
                            Field(data_type=DataType(name="INT32U"),
                                  code=0, name="softwareVersion"),
                        ]),
                        Event(priority=EventPriority.INFO,
                              name="Hello", code=1, fields=[]),
                        Event(priority=EventPriority.DEBUG,
                              name="GoodBye", code=2, fields=[]),
                    ])])
        self.assertEqual(actual, expected)

    def test_cluster_event_acl(self):
        actual = parseText("""
            client cluster EventTester = 0x123 {
               info event Hello = 1 {}
               debug event access(read: manage) GoodBye = 2 {}
               debug event access(read: administer) AdminEvent = 3 {}
            }
        """)
        expected = Idl(clusters=[
            Cluster(side=ClusterSide.CLIENT,
                    name="EventTester",
                    code=0x123,
                    events=[
                        Event(priority=EventPriority.INFO, readacl=AccessPrivilege.VIEW,
                              name="Hello", code=1, fields=[]),
                        Event(priority=EventPriority.DEBUG, readacl=AccessPrivilege.MANAGE,
                              name="GoodBye", code=2, fields=[]),
                        Event(priority=EventPriority.DEBUG, readacl=AccessPrivilege.ADMINISTER,
                              name="AdminEvent", code=3, fields=[]),
                    ])])
        self.assertEqual(actual, expected)

    def test_fabric_sensitive_event(self):
        actual = parseText("""
            client cluster EventTester = 0x123 {
               fabric_sensitive info event Hello = 1 {}
               fabric_sensitive debug event access(read: manage) GoodBye = 2 {}
               fabric_sensitive debug event access(read: administer) AdminEvent = 3 {}
            }
        """)
        expected = Idl(clusters=[
            Cluster(side=ClusterSide.CLIENT,
                    name="EventTester",
                    code=0x123,
                    events=[
                        Event(priority=EventPriority.INFO, readacl=AccessPrivilege.VIEW,
                              name="Hello", code=1, fields=[], qualities=EventQuality.FABRIC_SENSITIVE),
                        Event(priority=EventPriority.DEBUG, readacl=AccessPrivilege.MANAGE,
                              name="GoodBye", code=2, fields=[], qualities=EventQuality.FABRIC_SENSITIVE),
                        Event(priority=EventPriority.DEBUG, readacl=AccessPrivilege.ADMINISTER,
                              name="AdminEvent", code=3, fields=[], qualities=EventQuality.FABRIC_SENSITIVE),
                    ])])
        self.assertEqual(actual, expected)

    def test_parsing_metadata_for_cluster(self):
        actual = CreateParser(skip_meta=False).parse("""
server cluster A = 1 { /* Test comment */ }

// some empty lines and then indented
   client cluster B = 2 { }
        """)

        expected = Idl(clusters=[
            Cluster(parse_meta=ParseMetaData(line=2, column=1, start_pos=1),
                    side=ClusterSide.SERVER, name="A", code=1),
            Cluster(parse_meta=ParseMetaData(line=5, column=4, start_pos=87),
                    side=ClusterSide.CLIENT, name="B", code=2),
        ])
        self.assertEqual(actual, expected)

    def test_multiple_clusters(self):
        actual = parseText("""
            server cluster A = 1 { /* Test comment */ }
            client cluster B = 2 { }
            client cluster C = 3 { }
        """)

        expected = Idl(clusters=[
            Cluster(side=ClusterSide.SERVER, name="A", code=1),
            Cluster(side=ClusterSide.CLIENT, name="B", code=2),
            Cluster(side=ClusterSide.CLIENT, name="C", code=3),
        ])
        self.assertEqual(actual, expected)

    def test_endpoints(self):
        actual = parseText("""
            endpoint 12 {
                device type foo = 123, version 1;
                device type bar = 0xFF, version 2;

                server cluster Foo { }
                server cluster Bar { }
                binding cluster Bar;
                binding cluster Test;
            }
        """)

        expected = Idl(endpoints=[Endpoint(number=12,
                                           device_types=[
                                               DeviceType(
                                                   name="foo", code=123, version=1),
                                               DeviceType(
                                                   name="bar", code=0xFF, version=2),
                                           ],
                                           server_clusters=[
                                               ServerClusterInstantiation(
                                                   name="Foo"),
                                               ServerClusterInstantiation(
                                                   name="Bar"),
                                           ],
                                           client_bindings=["Bar", "Test"],)
                                  ])
        self.assertEqual(actual, expected)

    def test_cluster_instantiation(self):
        actual = parseText("""
            endpoint 3 {
                server cluster Example {
                    ram attribute inRamZero;
                    ram attribute inRamWithDefault default=123;
                    persist attribute inNVMNoDef;
                    persist attribute inNVMStr default="abc";
                    persist attribute inNVMWithDefault default = -33;
                    callback attribute hasCallbackBool default = true;
                }
            }
        """)

        expected = Idl(endpoints=[Endpoint(number=3,
                                           server_clusters=[
                                               ServerClusterInstantiation(name="Example", attributes=[
                                                   AttributeInstantiation(
                                                       name='inRamZero', storage=AttributeStorage.RAM),
                                                   AttributeInstantiation(name='inRamWithDefault',
                                                                          storage=AttributeStorage.RAM, default=123),
                                                   AttributeInstantiation(
                                                       name='inNVMNoDef', storage=AttributeStorage.PERSIST),
                                                   AttributeInstantiation(
                                                       name='inNVMStr', storage=AttributeStorage.PERSIST, default="abc"),
                                                   AttributeInstantiation(name='inNVMWithDefault',
                                                                          storage=AttributeStorage.PERSIST, default=-33),
                                                   AttributeInstantiation(name='hasCallbackBool',
                                                                          storage=AttributeStorage.CALLBACK, default=True),
                                               ]),
                                           ],
                                           client_bindings=[],)
                                  ])
        self.assertEqual(actual, expected)

    def test_multi_endpoints(self):
        actual = parseText("""
            endpoint 1 {}
            endpoint 2 {}
            endpoint 0xa {}
            endpoint 100 {}
        """)

        expected = Idl(endpoints=[
            Endpoint(number=1),
            Endpoint(number=2),
            Endpoint(number=10),
            Endpoint(number=100),
        ])
        self.assertEqual(actual, expected)

    def test_emits_events(self):
        actual = parseText("""
            endpoint 1 {
                server cluster Example {}
            }
            endpoint 2 {
              server cluster Example {
                emits event FooBar;
                emits event SomeNewEvent;
              }
              server cluster AnotherExample {
                emits event StartUp;
                emits event ShutDown;
              }
            }
        """)

        expected = Idl(endpoints=[
            Endpoint(number=1, server_clusters=[
                     ServerClusterInstantiation(name="Example")]),
            Endpoint(number=2, server_clusters=[
                ServerClusterInstantiation(name="Example", events_emitted={
                                           "FooBar", "SomeNewEvent"}),
                ServerClusterInstantiation(name="AnotherExample", events_emitted={
                    "StartUp", "ShutDown"}),
            ])
        ])

        self.assertEqual(actual, expected)


if __name__ == '__main__':
    unittest.main()