#!/usr/bin/env python3

##
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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
#
# https://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.

"""Test the name parsing logic."""

import json
import unittest

import avro.errors
import avro.protocol
import avro.schema


class TestName(unittest.TestCase):
    """Test name parsing"""

    def test_name_is_none(self):
        """When a name is None its namespace is None."""
        self.assertIsNone(avro.schema.Name(None, None, None).fullname)
        self.assertIsNone(avro.schema.Name(None, None, None).space)

    def test_name_not_empty_string(self):
        """A name cannot be the empty string."""
        self.assertRaises(avro.errors.SchemaParseException, avro.schema.Name, "", None, None)

    def test_name_space_specified(self):
        """Space combines with a name to become the fullname."""
        # name and namespace specified
        fullname = avro.schema.Name("a", "o.a.h", None).fullname
        self.assertEqual(fullname, "o.a.h.a")

    def test_name_inlined_space(self):
        """Space inlined with name is correctly splitted out."""
        name = avro.schema.Name("o.a", None)
        self.assertEqual(name.fullname, "o.a")
        self.assertEqual(name.name, "a")
        self.assertEqual(name.space, "o")

        name = avro.schema.Name("o.a.h.a", None)
        self.assertEqual(name.fullname, "o.a.h.a")
        self.assertEqual(name.name, "a")
        self.assertEqual(name.space, "o.a.h")

    def test_fullname_space_specified(self):
        """When name contains dots, namespace should be ignored."""
        fullname = avro.schema.Name("a.b.c.d", "o.a.h", None).fullname
        self.assertEqual(fullname, "a.b.c.d")

    def test_name_default_specified(self):
        """Default space becomes the namespace when the namespace is None."""
        fullname = avro.schema.Name("a", None, "b.c.d").fullname
        self.assertEqual(fullname, "b.c.d.a")

    def test_fullname_default_specified(self):
        """When a name contains dots, default space should be ignored."""
        fullname = avro.schema.Name("a.b.c.d", None, "o.a.h").fullname
        self.assertEqual(fullname, "a.b.c.d")

    def test_fullname_space_default_specified(self):
        """When a name contains dots, namespace and default space should be ignored."""
        fullname = avro.schema.Name("a.b.c.d", "o.a.a", "o.a.h").fullname
        self.assertEqual(fullname, "a.b.c.d")

    def test_name_space_default_specified(self):
        """When name and space are specified, default space should be ignored."""
        fullname = avro.schema.Name("a", "o.a.a", "o.a.h").fullname
        self.assertEqual(fullname, "o.a.a.a")

    def test_equal_names(self):
        """Equality of names is defined on the fullname and is case-sensitive."""
        self.assertEqual(
            avro.schema.Name("a.b.c.d", None, None),
            avro.schema.Name("d", "a.b.c", None),
        )
        self.assertNotEqual(avro.schema.Name("C.d", None, None), avro.schema.Name("c.d", None, None))

    def test_invalid_name(self):
        """The name portion of a fullname, record field names, and enum symbols must:
        start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"""
        self.assertRaises(
            avro.errors.InvalidName,
            avro.schema.Name,
            "an especially spacey cowboy",
            None,
            None,
        )
        self.assertRaises(
            avro.errors.InvalidName,
            avro.schema.Name,
            "99 problems but a name aint one",
            None,
            None,
        )
        # A name cannot start with dot.
        self.assertRaises(avro.errors.InvalidName, avro.schema.Name, ".a", None, None)
        self.assertRaises(avro.errors.InvalidName, avro.schema.Name, "o..a", None, None)
        self.assertRaises(avro.errors.InvalidName, avro.schema.Name, "a.", None, None)

    def test_null_namespace(self):
        """The empty string may be used as a namespace to indicate the null namespace."""
        name = avro.schema.Name("name", "", None)
        self.assertEqual(name.fullname, "name")
        self.assertIsNone(name.space)

    def test_disable_name_validation(self):
        """Test name validation disable."""
        # Test name class
        avro.schema.Name(name_attr="an especially spacey cowboy", space_attr=None, default_space=None, validate_name=False)
        avro.schema.Name(name_attr="cowboy", space_attr="an especially spacey ", default_space=None, validate_name=False)
        avro.schema.Name(name_attr="cowboy", space_attr=None, default_space="an especially spacey ", validate_name=False)
        avro.schema.Name(name_attr="name-space.with-dash.cowboy", space_attr=None, default_space=None, validate_name=False)
        avro.schema.Name(name_attr="cowboy", space_attr="name-space.with-dash", default_space=None, validate_name=False)

        # Test record schema
        cowboy_record_1 = avro.schema.RecordSchema(
            name="an especially spacey cowboy", namespace=None, fields=[{"name": "value", "type": "long"}], validate_names=False
        )
        cowboy_record_2 = avro.schema.RecordSchema(
            name="cowboy", namespace="an especially spacey ", fields=[{"name": "value", "type": "int"}], validate_names=False
        )

        # Test Names container class
        names = avro.schema.Names(default_namespace=None, validate_names=False)
        names.add_name(name_attr=cowboy_record_1.name, space_attr=cowboy_record_1.namespace, new_schema=cowboy_record_1)
        names.add_name(name_attr=cowboy_record_2.name, space_attr=cowboy_record_2.namespace, new_schema=cowboy_record_2)

        # Test fixed schema
        avro.schema.FixedSchema(name="an especially spacey cowboy", namespace=None, size=16, validate_names=False)
        avro.schema.FixedSchema(name="cowboy", namespace="an especially spacey", size=16, validate_names=False)

        # Test fixed decimal schema
        avro.schema.FixedDecimalSchema(name="an especially spacey cowboy", namespace=None, size=16, precision=2, validate_names=False)
        avro.schema.FixedDecimalSchema(name="cowboy", namespace="an especially spacey", size=16, precision=2, validate_names=False)

        # Test enum schema
        avro.schema.EnumSchema(name="an especially spacey cowboy", namespace=None, symbols=["A", "B"], validate_names=False)
        avro.schema.EnumSchema(name="cowboy", namespace="an especially spacey", symbols=["A", "B"], validate_names=False)


EXAMPLES = [
    # Enum
    {"type": "enum", "name": "invalid-name", "symbols": ["A", "B"]},
    {"type": "enum", "name": "invalid-ns.ab", "symbols": ["A", "B"]},
    {"type": "enum", "name": "ab", "namespace": "invalid-ns", "symbols": ["A", "B"]},
    # Record
    {"type": "record", "name": "invalid-name", "fields": [{"name": "distance", "type": "long"}]},
    {"type": "record", "name": "invalid-ns.journey", "fields": [{"name": "distance", "type": "long"}]},
    {"type": "record", "name": "journey", "namespace": "invalid-ns", "fields": [{"name": "distance", "type": "long"}]},
    # FixedSchema
    {"type": "fixed", "name": "invalid-name", "size": 10, "precision": 2},
    {"type": "fixed", "name": "invalid-ns.irrational", "size": 10, "precision": 2},
    {"type": "fixed", "name": "irrational", "namespace": "invalid-ns", "size": 10, "precision": 2},
    # FixedDecimalSchema / logical type
    {"type": "fixed", "logicalType": "decimal", "name": "invalid-name", "size": 10, "precision": 2},
    {"type": "fixed", "logicalType": "decimal", "name": "invalid-ns.irrational", "size": 10, "precision": 2},
    {"type": "fixed", "logicalType": "decimal", "name": "irrational", "namespace": "invalid-ns", "size": 10, "precision": 2},
    # In fields
    {
        "type": "record",
        "name": "world",
        "fields": [
            {
                "type": {"type": "record", "name": "invalid-name", "fields": [{"name": "distance", "type": "long"}]},
                "name": "cup",
            },
        ],
    },
    # In union
    [{"type": "string"}, {"type": "record", "name": "invalid-name", "fields": [{"name": "distance", "type": "long"}]}],
    # In array
    {
        "type": "record",
        "name": "world",
        "fields": [
            {
                "name": "journeys",
                "type": {
                    "type": "array",
                    "items": {
                        "type": "record",
                        "name": "invalid-name",
                        "fields": [{"name": "distance", "type": "long"}],
                    },
                },
            },
        ],
    },
    # In map
    {
        "type": "record",
        "name": "world",
        "fields": [
            {
                "name": "journeys",
                "type": {
                    "type": "map",
                    "values": {
                        "type": "record",
                        "name": "invalid-name",
                        "fields": [{"name": "distance", "type": "long"}],
                    },
                },
            },
        ],
    },
]


class ParseSchemaNameValidationDisabledTestCase(unittest.TestCase):
    """Enable generating parse test cases over all the valid and invalid example schema."""

    def __init__(self, test_schema_string):
        """Ignore the normal signature for unittest.TestCase because we are generating
        many test cases from this one class. This is safe as long as the autoloader
        ignores this class. The autoloader will ignore this class as long as it has
        no methods starting with `test_`.
        """
        super().__init__("parse_invalid_name")
        self.test_schema_string = test_schema_string

    def parse_invalid_name(self) -> None:
        """Parsing a schema with invalid name should not error"""
        schema_string = json.dumps(self.test_schema_string)
        # Parse with validation to ensure that correct exception is raised when validation enabled.
        with self.assertRaises(
            (avro.errors.AvroException, avro.errors.SchemaParseException), msg=f"Invalid schema should not have parsed: {self.test_schema_string!s}"
        ):
            avro.schema.parse(schema_string, validate_names=True)

        # The actual test with validation disabled.
        avro.schema.parse(schema_string, validate_names=False)


PROTOCOL_EXAMPLES = [
    # In record
    {
        "namespace": "lightyear",
        "protocol": "lightspeed",
        "types": [
            {"name": "current-speed", "type": "record", "fields": [{"name": "speed", "type": "int"}, {"name": "unit", "type": "string"}]},
            {"name": "over_c", "type": "error", "fields": [{"name": "message", "type": "string"}]},
        ],
        "messages": {
            "speedmessage": {"request": [{"name": "current_speed", "type": "current-speed"}], "response": "current-speed", "errors": ["over_c"]}
        },
    },
    # Error union
    {
        "namespace": "lightyear",
        "protocol": "lightspeed",
        "types": [
            {"name": "current_speed", "type": "record", "fields": [{"name": "speed", "type": "int"}, {"name": "unit", "type": "string"}]},
            {"name": "over-c", "type": "error", "fields": [{"name": "message", "type": "string"}]},
        ],
        "messages": {
            "speedmessage": {"request": [{"name": "current_speed", "type": "current_speed"}], "response": "current_speed", "errors": ["over-c"]}
        },
    },
    {
        "namespace": "lightyear",
        "protocol": "lightspeed",
        "types": [
            {"name": "current_speed", "type": "record", "fields": [{"name": "speed", "type": "int"}, {"name": "unit", "type": "string"}]},
            {"name": "over_c", "namespace": "error-speed", "type": "error", "fields": [{"name": "message", "type": "string"}]},
        ],
        "messages": {
            "speedmessage": {
                "request": [{"name": "current_speed", "type": "current_speed"}],
                "response": "current_speed",
                "errors": ["error-speed.over_c"],
            }
        },
    },
]


class ParseProtocolNameValidationDisabledTestCase(unittest.TestCase):
    """Test parsing of protocol when name validation is disabled."""

    def __init__(self, test_protocol_string):
        """Ignore the normal signature for unittest.TestCase because we are generating
        many test cases from this one class. This is safe as long as the autoloader
        ignores this class. The autoloader will ignore this class as long as it has
        no methods starting with `test_`.
        """
        super().__init__("parse_protocol_with_invalid_name")
        self.test_protocol_string = test_protocol_string

    def parse_protocol_with_invalid_name(self):
        """Test error union"""
        protocol = json.dumps(self.test_protocol_string)

        with self.assertRaises(
            (avro.errors.AvroException, avro.errors.SchemaParseException), msg=f"Invalid schema should not have parsed: {self.test_protocol_string!s}"
        ):
            avro.protocol.parse(protocol, validate_names=True)

        avro.protocol.parse(protocol, validate_names=False)


def load_tests(loader, default_tests, pattern):
    """Generate test cases across many test schema."""
    suite = unittest.TestSuite()
    suite.addTests(loader.loadTestsFromTestCase(TestName))
    suite.addTests(ParseSchemaNameValidationDisabledTestCase(ex) for ex in EXAMPLES)
    suite.addTests(ParseProtocolNameValidationDisabledTestCase(ex) for ex in PROTOCOL_EXAMPLES)
    return suite


if __name__ == "__main__":  # pragma: no coverage
    unittest.main()