test_schema.py

# Copyright 2015 Google LLC
#
# 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

import mock


class TestSchemaField(unittest.TestCase):

    @staticmethod
    def _get_target_class():
        from google.cloud.bigquery.schema import SchemaField

        return SchemaField

    def _make_one(self, *args, **kw):
        return self._get_target_class()(*args, **kw)

    def test_constructor_defaults(self):
        field = self._make_one('test', 'STRING')
        self.assertEqual(field._name, 'test')
        self.assertEqual(field._field_type, 'STRING')
        self.assertEqual(field._mode, 'NULLABLE')
        self.assertIsNone(field._description)
        self.assertEqual(field._fields, ())

    def test_constructor_explicit(self):
        field = self._make_one('test', 'STRING', mode='REQUIRED',
                               description='Testing')
        self.assertEqual(field._name, 'test')
        self.assertEqual(field._field_type, 'STRING')
        self.assertEqual(field._mode, 'REQUIRED')
        self.assertEqual(field._description, 'Testing')
        self.assertEqual(field._fields, ())

    def test_constructor_subfields(self):
        sub_field1 = self._make_one('area_code', 'STRING')
        sub_field2 = self._make_one('local_number', 'STRING')
        field = self._make_one(
            'phone_number',
            'RECORD',
            fields=[sub_field1, sub_field2],
        )
        self.assertEqual(field._name, 'phone_number')
        self.assertEqual(field._field_type, 'RECORD')
        self.assertEqual(field._mode, 'NULLABLE')
        self.assertIsNone(field._description)
        self.assertEqual(len(field._fields), 2)
        self.assertIs(field._fields[0], sub_field1)
        self.assertIs(field._fields[1], sub_field2)

    def test_to_api_repr(self):
        field = self._make_one('foo', 'INTEGER', 'NULLABLE')
        self.assertEqual(field.to_api_repr(), {
            'mode': 'NULLABLE',
            'name': 'foo',
            'type': 'INTEGER',
            'description': None,
        })

    def test_to_api_repr_with_subfield(self):
        subfield = self._make_one('bar', 'INTEGER', 'NULLABLE')
        field = self._make_one('foo', 'RECORD', 'REQUIRED', fields=(subfield,))
        self.assertEqual(field.to_api_repr(), {
            'fields': [{
                'mode': 'NULLABLE',
                'name': 'bar',
                'type': 'INTEGER',
                'description': None,
            }],
            'mode': 'REQUIRED',
            'name': 'foo',
            'type': 'RECORD',
            'description': None,
        })

    def test_from_api_repr(self):
        field = self._get_target_class().from_api_repr({
            'fields': [{
                'mode': 'nullable',
                'name': 'bar',
                'type': 'integer',
            }],
            'mode': 'required',
            'description': 'test_description',
            'name': 'foo',
            'type': 'record',
        })
        self.assertEqual(field.name, 'foo')
        self.assertEqual(field.field_type, 'RECORD')
        self.assertEqual(field.mode, 'REQUIRED')
        self.assertEqual(field.description, 'test_description')
        self.assertEqual(len(field.fields), 1)
        self.assertEqual(field.fields[0].name, 'bar')
        self.assertEqual(field.fields[0].field_type, 'INTEGER')
        self.assertEqual(field.fields[0].mode, 'NULLABLE')

    def test_from_api_repr_defaults(self):
        field = self._get_target_class().from_api_repr({
            'name': 'foo',
            'type': 'record',
        })
        self.assertEqual(field.name, 'foo')
        self.assertEqual(field.field_type, 'RECORD')
        self.assertEqual(field.mode, 'NULLABLE')
        self.assertEqual(field.description, None)
        self.assertEqual(len(field.fields), 0)

    def test_name_property(self):
        name = 'lemon-ness'
        schema_field = self._make_one(name, 'INTEGER')
        self.assertIs(schema_field.name, name)

    def test_field_type_property(self):
        field_type = 'BOOLEAN'
        schema_field = self._make_one('whether', field_type)
        self.assertIs(schema_field.field_type, field_type)

    def test_mode_property(self):
        mode = 'REPEATED'
        schema_field = self._make_one('again', 'FLOAT', mode=mode)
        self.assertIs(schema_field.mode, mode)

    def test_is_nullable(self):
        mode = 'NULLABLE'
        schema_field = self._make_one('test', 'FLOAT', mode=mode)
        self.assertTrue(schema_field.is_nullable)

    def test_is_not_nullable(self):
        mode = 'REPEATED'
        schema_field = self._make_one('test', 'FLOAT', mode=mode)
        self.assertFalse(schema_field.is_nullable)

    def test_description_property(self):
        description = 'It holds some data.'
        schema_field = self._make_one(
            'do', 'TIMESTAMP', description=description)
        self.assertIs(schema_field.description, description)

    def test_fields_property(self):
        sub_field1 = self._make_one('one', 'STRING')
        sub_field2 = self._make_one('fish', 'INTEGER')
        fields = (sub_field1, sub_field2)
        schema_field = self._make_one('boat', 'RECORD', fields=fields)
        self.assertIs(schema_field.fields, fields)

    def test___eq___wrong_type(self):
        field = self._make_one('test', 'STRING')
        other = object()
        self.assertNotEqual(field, other)
        self.assertEqual(field, mock.ANY)

    def test___eq___name_mismatch(self):
        field = self._make_one('test', 'STRING')
        other = self._make_one('other', 'STRING')
        self.assertNotEqual(field, other)

    def test___eq___field_type_mismatch(self):
        field = self._make_one('test', 'STRING')
        other = self._make_one('test', 'INTEGER')
        self.assertNotEqual(field, other)

    def test___eq___mode_mismatch(self):
        field = self._make_one('test', 'STRING', mode='REQUIRED')
        other = self._make_one('test', 'STRING', mode='NULLABLE')
        self.assertNotEqual(field, other)

    def test___eq___description_mismatch(self):
        field = self._make_one('test', 'STRING', description='Testing')
        other = self._make_one('test', 'STRING', description='Other')
        self.assertNotEqual(field, other)

    def test___eq___fields_mismatch(self):
        sub1 = self._make_one('sub1', 'STRING')
        sub2 = self._make_one('sub2', 'STRING')
        field = self._make_one('test', 'RECORD', fields=[sub1])
        other = self._make_one('test', 'RECORD', fields=[sub2])
        self.assertNotEqual(field, other)

    def test___eq___hit(self):
        field = self._make_one('test', 'STRING', mode='REQUIRED',
                               description='Testing')
        other = self._make_one('test', 'STRING', mode='REQUIRED',
                               description='Testing')
        self.assertEqual(field, other)

    def test___eq___hit_case_diff_on_type(self):
        field = self._make_one('test', 'STRING', mode='REQUIRED',
                               description='Testing')
        other = self._make_one('test', 'string', mode='REQUIRED',
                               description='Testing')
        self.assertEqual(field, other)

    def test___eq___hit_w_fields(self):
        sub1 = self._make_one('sub1', 'STRING')
        sub2 = self._make_one('sub2', 'STRING')
        field = self._make_one('test', 'RECORD', fields=[sub1, sub2])
        other = self._make_one('test', 'RECORD', fields=[sub1, sub2])
        self.assertEqual(field, other)

    def test___ne___wrong_type(self):
        field = self._make_one('toast', 'INTEGER')
        other = object()
        self.assertNotEqual(field, other)
        self.assertEqual(field, mock.ANY)

    def test___ne___same_value(self):
        field1 = self._make_one('test', 'TIMESTAMP', mode='REPEATED')
        field2 = self._make_one('test', 'TIMESTAMP', mode='REPEATED')
        # unittest ``assertEqual`` uses ``==`` not ``!=``.
        comparison_val = (field1 != field2)
        self.assertFalse(comparison_val)

    def test___ne___different_values(self):
        field1 = self._make_one(
            'test1', 'FLOAT', mode='REPEATED', description='Not same')
        field2 = self._make_one(
            'test2', 'FLOAT', mode='NULLABLE', description='Knot saym')
        self.assertNotEqual(field1, field2)

    def test___hash__set_equality(self):
        sub1 = self._make_one('sub1', 'STRING')
        sub2 = self._make_one('sub2', 'STRING')
        field1 = self._make_one('test', 'RECORD', fields=[sub1])
        field2 = self._make_one('test', 'RECORD', fields=[sub2])
        set_one = {field1, field2}
        set_two = {field1, field2}
        self.assertEqual(set_one, set_two)

    def test___hash__not_equals(self):
        sub1 = self._make_one('sub1', 'STRING')
        sub2 = self._make_one('sub2', 'STRING')
        field1 = self._make_one('test', 'RECORD', fields=[sub1])
        field2 = self._make_one('test', 'RECORD', fields=[sub2])
        set_one = {field1}
        set_two = {field2}
        self.assertNotEqual(set_one, set_two)

    def test___repr__(self):
        field1 = self._make_one('field1', 'STRING')
        expected = "SchemaField('field1', 'STRING', 'NULLABLE', None, ())"
        self.assertEqual(repr(field1), expected)


# TODO: dedup with the same class in test_table.py.
class _SchemaBase(object):

    def _verify_field(self, field, r_field):
        self.assertEqual(field.name, r_field['name'])
        self.assertEqual(field.field_type, r_field['type'])
        self.assertEqual(field.mode, r_field.get('mode', 'NULLABLE'))

    def _verifySchema(self, schema, resource):
        r_fields = resource['schema']['fields']
        self.assertEqual(len(schema), len(r_fields))

        for field, r_field in zip(schema, r_fields):
            self._verify_field(field, r_field)


class Test_parse_schema_resource(unittest.TestCase, _SchemaBase):

    def _call_fut(self, resource):
        from google.cloud.bigquery.schema import _parse_schema_resource

        return _parse_schema_resource(resource)

    def _make_resource(self):
        return {
            'schema': {'fields': [
                {'name': 'full_name', 'type': 'STRING', 'mode': 'REQUIRED'},
                {'name': 'age', 'type': 'INTEGER', 'mode': 'REQUIRED'},
            ]},
        }

    def test__parse_schema_resource_defaults(self):
        RESOURCE = self._make_resource()
        schema = self._call_fut(RESOURCE['schema'])
        self._verifySchema(schema, RESOURCE)

    def test__parse_schema_resource_subfields(self):
        RESOURCE = self._make_resource()
        RESOURCE['schema']['fields'].append(
            {'name': 'phone',
             'type': 'RECORD',
             'mode': 'REPEATED',
             'fields': [{'name': 'type',
                         'type': 'STRING',
                         'mode': 'REQUIRED'},
                        {'name': 'number',
                         'type': 'STRING',
                         'mode': 'REQUIRED'}]})
        schema = self._call_fut(RESOURCE['schema'])
        self._verifySchema(schema, RESOURCE)

    def test__parse_schema_resource_fields_without_mode(self):
        RESOURCE = self._make_resource()
        RESOURCE['schema']['fields'].append(
            {'name': 'phone',
             'type': 'STRING'})

        schema = self._call_fut(RESOURCE['schema'])
        self._verifySchema(schema, RESOURCE)


class Test_build_schema_resource(unittest.TestCase, _SchemaBase):

    def _call_fut(self, resource):
        from google.cloud.bigquery.schema import _build_schema_resource

        return _build_schema_resource(resource)

    def test_defaults(self):
        from google.cloud.bigquery.schema import SchemaField

        full_name = SchemaField('full_name', 'STRING', mode='REQUIRED')
        age = SchemaField('age', 'INTEGER', mode='REQUIRED')
        resource = self._call_fut([full_name, age])
        self.assertEqual(len(resource), 2)
        self.assertEqual(resource[0],
                         {'name': 'full_name',
                          'type': 'STRING',
                          'mode': 'REQUIRED',
                          'description': None})
        self.assertEqual(resource[1],
                         {'name': 'age',
                          'type': 'INTEGER',
                          'mode': 'REQUIRED',
                          'description': None})

    def test_w_description(self):
        from google.cloud.bigquery.schema import SchemaField

        DESCRIPTION = 'DESCRIPTION'
        full_name = SchemaField('full_name', 'STRING', mode='REQUIRED',
                                description=DESCRIPTION)
        age = SchemaField('age', 'INTEGER', mode='REQUIRED')
        resource = self._call_fut([full_name, age])
        self.assertEqual(len(resource), 2)
        self.assertEqual(resource[0],
                         {'name': 'full_name',
                          'type': 'STRING',
                          'mode': 'REQUIRED',
                          'description': DESCRIPTION})
        self.assertEqual(resource[1],
                         {'name': 'age',
                          'type': 'INTEGER',
                          'mode': 'REQUIRED',
                          'description': None})

    def test_w_subfields(self):
        from google.cloud.bigquery.schema import SchemaField

        full_name = SchemaField('full_name', 'STRING', mode='REQUIRED')
        ph_type = SchemaField('type', 'STRING', 'REQUIRED')
        ph_num = SchemaField('number', 'STRING', 'REQUIRED')
        phone = SchemaField('phone', 'RECORD', mode='REPEATED',
                            fields=[ph_type, ph_num])
        resource = self._call_fut([full_name, phone])
        self.assertEqual(len(resource), 2)
        self.assertEqual(resource[0],
                         {'name': 'full_name',
                          'type': 'STRING',
                          'mode': 'REQUIRED',
                          'description': None})
        self.assertEqual(resource[1],
                         {'name': 'phone',
                          'type': 'RECORD',
                          'mode': 'REPEATED',
                          'description': None,
                          'fields': [{'name': 'type',
                                      'type': 'STRING',
                                      'mode': 'REQUIRED',
                                      'description': None},
                                     {'name': 'number',
                                      'type': 'STRING',
                                      'mode': 'REQUIRED',
                                      'description': None}]})