# Copyright 2015 Google Inc. 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 unittest


class Test_not_null(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _not_null
        return _not_null(value, field)

    def test_w_none_nullable(self):
        self.assertFalse(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        self.assertTrue(self._callFUT(None, _Field('REQUIRED')))

    def test_w_value(self):
        self.assertTrue(self._callFUT(object(), object()))


class Test_int_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _int_from_json
        return _int_from_json(value, field)

    def test_w_none_nullable(self):
        self.assertIsNone(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        with self.assertRaises(TypeError):
            self._callFUT(None, _Field('REQUIRED'))

    def test_w_string_value(self):
        coerced = self._callFUT('42', object())
        self.assertEqual(coerced, 42)

    def test_w_float_value(self):
        coerced = self._callFUT(42, object())
        self.assertEqual(coerced, 42)


class Test_float_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _float_from_json
        return _float_from_json(value, field)

    def test_w_none_nullable(self):
        self.assertIsNone(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        with self.assertRaises(TypeError):
            self._callFUT(None, _Field('REQUIRED'))

    def test_w_string_value(self):
        coerced = self._callFUT('3.1415', object())
        self.assertEqual(coerced, 3.1415)

    def test_w_float_value(self):
        coerced = self._callFUT(3.1415, object())
        self.assertEqual(coerced, 3.1415)


class Test_bool_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _bool_from_json
        return _bool_from_json(value, field)

    def test_w_none_nullable(self):
        self.assertIsNone(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        with self.assertRaises(AttributeError):
            self._callFUT(None, _Field('REQUIRED'))

    def test_w_value_t(self):
        coerced = self._callFUT('T', object())
        self.assertTrue(coerced)

    def test_w_value_true(self):
        coerced = self._callFUT('True', object())
        self.assertTrue(coerced)

    def test_w_value_1(self):
        coerced = self._callFUT('1', object())
        self.assertTrue(coerced)

    def test_w_value_other(self):
        coerced = self._callFUT('f', object())
        self.assertFalse(coerced)


class Test_datetime_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _datetime_from_json
        return _datetime_from_json(value, field)

    def test_w_none_nullable(self):
        self.assertIsNone(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        with self.assertRaises(TypeError):
            self._callFUT(None, _Field('REQUIRED'))

    def test_w_string_value(self):
        import datetime
        from google.cloud._helpers import _EPOCH
        coerced = self._callFUT('1.234567', object())
        self.assertEqual(
            coerced,
            _EPOCH + datetime.timedelta(seconds=1, microseconds=234567))

    def test_w_float_value(self):
        import datetime
        from google.cloud._helpers import _EPOCH
        coerced = self._callFUT(1.234567, object())
        self.assertEqual(
            coerced,
            _EPOCH + datetime.timedelta(seconds=1, microseconds=234567))


class Test_date_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _date_from_json
        return _date_from_json(value, field)

    def test_w_none_nullable(self):
        self.assertIsNone(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        with self.assertRaises(TypeError):
            self._callFUT(None, _Field('REQUIRED'))

    def test_w_string_value(self):
        import datetime
        coerced = self._callFUT('1987-09-22', object())
        self.assertEqual(
            coerced,
            datetime.date(1987, 9, 22))


class Test_record_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _record_from_json
        return _record_from_json(value, field)

    def test_w_none_nullable(self):
        self.assertIsNone(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        with self.assertRaises(TypeError):
            self._callFUT(None, _Field('REQUIRED'))

    def test_w_nullable_subfield_none(self):
        subfield = _Field('NULLABLE', 'age', 'INTEGER')
        field = _Field('REQUIRED', fields=[subfield])
        value = {'f': [{'v': None}]}
        coerced = self._callFUT(value, field)
        self.assertEqual(coerced, {'age': None})

    def test_w_scalar_subfield(self):
        subfield = _Field('REQUIRED', 'age', 'INTEGER')
        field = _Field('REQUIRED', fields=[subfield])
        value = {'f': [{'v': 42}]}
        coerced = self._callFUT(value, field)
        self.assertEqual(coerced, {'age': 42})

    def test_w_repeated_subfield(self):
        subfield = _Field('REPEATED', 'color', 'STRING')
        field = _Field('REQUIRED', fields=[subfield])
        value = {'f': [{'v': ['red', 'yellow', 'blue']}]}
        coerced = self._callFUT(value, field)
        self.assertEqual(coerced, {'color': ['red', 'yellow', 'blue']})

    def test_w_record_subfield(self):
        full_name = _Field('REQUIRED', 'full_name', 'STRING')
        area_code = _Field('REQUIRED', 'area_code', 'STRING')
        local_number = _Field('REQUIRED', 'local_number', 'STRING')
        rank = _Field('REQUIRED', 'rank', 'INTEGER')
        phone = _Field('NULLABLE', 'phone', 'RECORD',
                       fields=[area_code, local_number, rank])
        person = _Field('REQUIRED', 'person', 'RECORD',
                        fields=[full_name, phone])
        value = {
            'f': [
                {'v': 'Phred Phlyntstone'},
                {'v': {'f': [{'v': '800'}, {'v': '555-1212'}, {'v': 1}]}},
            ],
        }
        expected = {
            'full_name': 'Phred Phlyntstone',
            'phone': {
                'area_code': '800',
                'local_number': '555-1212',
                'rank': 1,
            }
        }
        coerced = self._callFUT(value, person)
        self.assertEqual(coerced, expected)


class Test_string_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _string_from_json
        return _string_from_json(value, field)

    def test_w_none_nullable(self):
        self.assertIsNone(self._callFUT(None, _Field('NULLABLE')))

    def test_w_none_required(self):
        self.assertIsNone(self._callFUT(None, _Field('RECORD')))

    def test_w_string_value(self):
        coerced = self._callFUT('Wonderful!', object())
        self.assertEqual(coerced, 'Wonderful!')


class Test_rows_from_json(unittest.TestCase):

    def _callFUT(self, value, field):
        from google.cloud.bigquery._helpers import _rows_from_json
        return _rows_from_json(value, field)

    def test_w_record_subfield(self):
        full_name = _Field('REQUIRED', 'full_name', 'STRING')
        area_code = _Field('REQUIRED', 'area_code', 'STRING')
        local_number = _Field('REQUIRED', 'local_number', 'STRING')
        rank = _Field('REQUIRED', 'rank', 'INTEGER')
        phone = _Field('NULLABLE', 'phone', 'RECORD',
                       fields=[area_code, local_number, rank])
        color = _Field('REPEATED', 'color', 'STRING')
        schema = [full_name, phone, color]
        rows = [
            {'f': [
                {'v': 'Phred Phlyntstone'},
                {'v': {'f': [{'v': '800'}, {'v': '555-1212'}, {'v': 1}]}},
                {'v': ['orange', 'black']},
            ]},
            {'f': [
                {'v': 'Bharney Rhubble'},
                {'v': {'f': [{'v': '877'}, {'v': '768-5309'}, {'v': 2}]}},
                {'v': ['brown']},
            ]},
            {'f': [
                {'v': 'Wylma Phlyntstone'},
                {'v': None},
                {'v': []},
            ]},
        ]
        phred_phone = {
            'area_code': '800',
            'local_number': '555-1212',
            'rank': 1,
        }
        bharney_phone = {
            'area_code': '877',
            'local_number': '768-5309',
            'rank': 2,
        }
        expected = [
            ('Phred Phlyntstone', phred_phone, ['orange', 'black']),
            ('Bharney Rhubble', bharney_phone, ['brown']),
            ('Wylma Phlyntstone', None, []),
        ]
        coerced = self._callFUT(rows, schema)
        self.assertEqual(coerced, expected)

    def test_w_int64_float64(self):
        # "Standard" SQL dialect uses 'INT64', 'FLOAT64'.
        candidate = _Field('REQUIRED', 'candidate', 'STRING')
        votes = _Field('REQUIRED', 'votes', 'INT64')
        percentage = _Field('REQUIRED', 'percentage', 'FLOAT64')
        schema = [candidate, votes, percentage]
        rows = [
            {'f': [
                {'v': 'Phred Phlyntstone'},
                {'v': 8},
                {'v': 0.25},
            ]},
            {'f': [
                {'v': 'Bharney Rhubble'},
                {'v': 4},
                {'v': 0.125},
            ]},
            {'f': [
                {'v': 'Wylma Phlyntstone'},
                {'v': 20},
                {'v': 0.625},
            ]},
        ]
        expected = [
            ('Phred Phlyntstone', 8, 0.25),
            ('Bharney Rhubble', 4, 0.125),
            ('Wylma Phlyntstone', 20, 0.625),
        ]
        coerced = self._callFUT(rows, schema)
        self.assertEqual(coerced, expected)


class Test_ConfigurationProperty(unittest.TestCase):

    def _getTargetClass(self):
        from google.cloud.bigquery._helpers import _ConfigurationProperty
        return _ConfigurationProperty

    def _makeOne(self, *args, **kw):
        return self._getTargetClass()(*args, **kw)

    def test_it(self):

        class Configuration(object):
            _attr = None

        class Wrapper(object):
            attr = self._makeOne('attr')

            def __init__(self):
                self._configuration = Configuration()

        self.assertEqual(Wrapper.attr.name, 'attr')

        wrapper = Wrapper()
        self.assertEqual(wrapper.attr, None)

        value = object()
        wrapper.attr = value
        self.assertTrue(wrapper.attr is value)
        self.assertTrue(wrapper._configuration._attr is value)

        del wrapper.attr
        self.assertEqual(wrapper.attr, None)
        self.assertEqual(wrapper._configuration._attr, None)


class Test_TypedProperty(unittest.TestCase):

    def _getTargetClass(self):
        from google.cloud.bigquery._helpers import _TypedProperty
        return _TypedProperty

    def _makeOne(self, *args, **kw):
        return self._getTargetClass()(*args, **kw)

    def test_it(self):

        class Configuration(object):
            _attr = None

        class Wrapper(object):
            attr = self._makeOne('attr', int)

            def __init__(self):
                self._configuration = Configuration()

        wrapper = Wrapper()
        with self.assertRaises(ValueError):
            wrapper.attr = 'BOGUS'

        wrapper.attr = 42
        self.assertEqual(wrapper.attr, 42)
        self.assertEqual(wrapper._configuration._attr, 42)

        del wrapper.attr
        self.assertEqual(wrapper.attr, None)
        self.assertEqual(wrapper._configuration._attr, None)


class Test_EnumProperty(unittest.TestCase):

    def _getTargetClass(self):
        from google.cloud.bigquery._helpers import _EnumProperty
        return _EnumProperty

    def test_it(self):

        class Sub(self._getTargetClass()):
            ALLOWED = ('FOO', 'BAR', 'BAZ')

        class Configuration(object):
            _attr = None

        class Wrapper(object):
            attr = Sub('attr')

            def __init__(self):
                self._configuration = Configuration()

        wrapper = Wrapper()
        with self.assertRaises(ValueError):
            wrapper.attr = 'BOGUS'

        wrapper.attr = 'FOO'
        self.assertEqual(wrapper.attr, 'FOO')
        self.assertEqual(wrapper._configuration._attr, 'FOO')

        del wrapper.attr
        self.assertEqual(wrapper.attr, None)
        self.assertEqual(wrapper._configuration._attr, None)


class Test_UDFResourcesProperty(unittest.TestCase):

    def _getTargetClass(self):
        from google.cloud.bigquery._helpers import UDFResourcesProperty
        return UDFResourcesProperty

    def _makeOne(self, *args, **kw):
        return self._getTargetClass()(*args, **kw)

    def _descriptor_and_klass(self):
        descriptor = self._makeOne()

        class _Test(object):
            _udf_resources = ()
            udf_resources = descriptor

        return descriptor, _Test

    def test_class_getter(self):
        descriptor, klass = self._descriptor_and_klass()
        self.assertTrue(klass.udf_resources is descriptor)

    def test_instance_getter_empty(self):
        _, klass = self._descriptor_and_klass()
        instance = klass()
        self.assertEqual(instance.udf_resources, [])

    def test_instance_getter_w_non_empty_list(self):
        from google.cloud.bigquery._helpers import UDFResource
        RESOURCE_URI = 'gs://some-bucket/js/lib.js'
        udf_resources = [UDFResource("resourceUri", RESOURCE_URI)]
        _, klass = self._descriptor_and_klass()
        instance = klass()
        instance._udf_resources = tuple(udf_resources)

        self.assertEqual(instance.udf_resources, udf_resources)

    def test_instance_setter_w_empty_list(self):
        from google.cloud.bigquery._helpers import UDFResource
        RESOURCE_URI = 'gs://some-bucket/js/lib.js'
        udf_resources = [UDFResource("resourceUri", RESOURCE_URI)]
        _, klass = self._descriptor_and_klass()
        instance = klass()
        instance._udf_resources = udf_resources

        instance.udf_resources = []

        self.assertEqual(instance.udf_resources, [])

    def test_instance_setter_w_valid_udf(self):
        from google.cloud.bigquery._helpers import UDFResource
        RESOURCE_URI = 'gs://some-bucket/js/lib.js'
        udf_resources = [UDFResource("resourceUri", RESOURCE_URI)]
        _, klass = self._descriptor_and_klass()
        instance = klass()

        instance.udf_resources = udf_resources

        self.assertEqual(instance.udf_resources, udf_resources)

    def test_instance_setter_w_bad_udfs(self):
        _, klass = self._descriptor_and_klass()
        instance = klass()

        with self.assertRaises(ValueError):
            instance.udf_resources = ["foo"]

        self.assertEqual(instance.udf_resources, [])


class _Field(object):

    def __init__(self, mode, name='unknown', field_type='UNKNOWN', fields=()):
        self.mode = mode
        self.name = name
        self.field_type = field_type
        self.fields = fields
