tests/common/test_datatypes.py

# 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
#
#      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.

from collections import OrderedDict
import ctypes
from datetime import datetime, timedelta
import decimal
import pytest
import uuid

from pyignite.datatypes import (
    ByteObject, IntObject, FloatObject, CharObject, ShortObject, BoolObject, ByteArrayObject, IntArrayObject,
    ShortArrayObject, FloatArrayObject, BoolArrayObject, CharArrayObject, TimestampObject, String, BinaryEnumObject,
    TimestampArrayObject, BinaryEnumArrayObject, ObjectArrayObject, CollectionObject, MapObject
)
from pyignite.utils import unsigned

put_get_data_params = [
    # integers
    (42, None),
    (42, ByteObject),
    (42, ShortObject),
    (42, IntObject),

    # floats
    (3.1415, None),  # True for Double but not Float
    (3.5, FloatObject),

    # char is never autodetected
    ('ы', CharObject),
    ('カ', CharObject),

    # bool
    (True, None),
    (False, None),
    (True, BoolObject),
    (False, BoolObject),

    # arrays of integers
    ([1, 2, 3, 5], None),
    (b'buzz', ByteArrayObject),
    (bytearray([7, 8, 8, 11]), None),
    (bytearray([7, 8, 8, 11]), ByteArrayObject),
    ([1, 2, 3, 5], ShortArrayObject),
    ([1, 2, 3, 5], IntArrayObject),

    # arrays of floats
    ([2.2, 4.4, 6.6], None),
    ([2.5, 6.5], FloatArrayObject),

    # array of char
    (['ы', 'カ'], CharArrayObject),

    # array of bool
    ([True, False, True], None),
    ([True, False], BoolArrayObject),
    ([False, True], BoolArrayObject),
    ([True, False, True, False], BoolArrayObject),

    # string
    ('Little Mary had a lamb', None),
    ('This is a test', String),

    # decimals
    (decimal.Decimal('2.5'), None),
    (decimal.Decimal('-1.3'), None),

    # uuid
    (uuid.uuid4(), None),

    # date
    (datetime(year=1998, month=4, day=6, hour=18, minute=30), None),

    # no autodetection for timestamp either
    (
        (datetime(year=1998, month=4, day=6, hour=18, minute=30), 1000),
        TimestampObject
    ),

    # time
    (timedelta(days=4, hours=4, minutes=24), None),

    # enum is useless in Python, except for interoperability with Java.
    # Also no autodetection
    ((5, 6), BinaryEnumObject),

    # arrays of standard types
    (['String 1', 'String 2'], None),
    (['Some of us are empty', None, 'But not the others'], None),

    ([decimal.Decimal('2.71828'), decimal.Decimal('100')], None),
    ([decimal.Decimal('2.1'), None, decimal.Decimal('3.1415')], None),

    ([uuid.uuid4(), uuid.uuid4()], None),
    (
        [
            datetime(year=2010, month=1, day=1),
            datetime(year=2010, month=12, day=31),
        ],
        None,
    ),
    ([timedelta(minutes=30), timedelta(hours=2)], None),
    (
        [
            (datetime(year=2010, month=1, day=1), 1000),
            (datetime(year=2010, month=12, day=31), 200),
        ],
        TimestampArrayObject
    ),
    ((-1, [(6001, 1), (6002, 2), (6003, 3)]), BinaryEnumArrayObject),

    # object array
    ((ObjectArrayObject.OBJECT, [1, 2, decimal.Decimal('3')]), ObjectArrayObject),

    # collection
    ((CollectionObject.LINKED_LIST, [1, 2, 3]), None),

    # map
    ((MapObject.HASH_MAP, {'key': 4, 5: 6.0}), None),
    ((MapObject.LINKED_HASH_MAP, OrderedDict([('key', 4), (5, 6.0)])), None),
]


@pytest.mark.parametrize(
    'value, value_hint',
    put_get_data_params
)
def test_put_get_data(cache, value, value_hint):
    cache.put('my_key', value, value_hint=value_hint)
    assert cache.get('my_key') == value


@pytest.mark.parametrize(
    'value, value_hint',
    put_get_data_params
)
@pytest.mark.asyncio
async def test_put_get_data_async(async_cache, value, value_hint):
    await async_cache.put('my_key', value, value_hint=value_hint)
    assert await async_cache.get('my_key') == value


bytearray_params = [
    [1, 2, 3, 5],
    (7, 8, 13, 18),
    (-128, -1, 0, 1, 127, 255),
]


@pytest.mark.parametrize(
    'value',
    bytearray_params
)
def test_bytearray_from_list_or_tuple(cache, value):
    """
    ByteArrayObject's pythonic type is `bytearray`, but it should also accept
    lists or tuples as a content.
    """

    cache.put('my_key', value, value_hint=ByteArrayObject)

    assert cache.get('my_key') == bytearray([unsigned(ch, ctypes.c_ubyte) for ch in value])


@pytest.mark.parametrize(
    'value',
    bytearray_params
)
@pytest.mark.asyncio
async def test_bytearray_from_list_or_tuple_async(async_cache, value):
    """
    ByteArrayObject's pythonic type is `bytearray`, but it should also accept
    lists or tuples as a content.
    """

    await async_cache.put('my_key', value, value_hint=ByteArrayObject)

    result = await async_cache.get('my_key')
    assert result == bytearray([unsigned(ch, ctypes.c_ubyte) for ch in value])


uuid_params = [
    'd57babad-7bc1-4c82-9f9c-e72841b92a85',
    '5946c0c0-2b76-479d-8694-a2e64a3968da',
    'a521723d-ad5d-46a6-94ad-300f850ef704',
]

uuid_table_create_sql = "CREATE TABLE test_uuid_repr (id INTEGER PRIMARY KEY, uuid_field UUID)"
uuid_table_drop_sql = "DROP TABLE test_uuid_repr IF EXISTS"
uuid_table_insert_sql = "INSERT INTO test_uuid_repr(id, uuid_field) VALUES (?, ?)"
uuid_table_query_sql = "SELECT * FROM test_uuid_repr WHERE uuid_field=?"


@pytest.fixture()
async def uuid_table(client):
    client.sql(uuid_table_drop_sql)
    client.sql(uuid_table_create_sql)
    yield None
    client.sql(uuid_table_drop_sql)


@pytest.fixture()
async def uuid_table_async(async_client):
    await async_client.sql(uuid_table_drop_sql)
    await async_client.sql(uuid_table_create_sql)
    yield None
    await async_client.sql(uuid_table_drop_sql)


@pytest.mark.parametrize(
    'uuid_string',
    uuid_params
)
def test_uuid_representation(client, uuid_string, uuid_table):
    """ Test if textual UUID representation is correct. """
    uuid_value = uuid.UUID(uuid_string)

    # use uuid.UUID class to insert data
    client.sql(uuid_table_insert_sql, query_args=[1, uuid_value])
    # use hex string to retrieve data
    with client.sql(uuid_table_query_sql, query_args=[str(uuid_value)]) as cursor:
        result = list(cursor)

        # if a line was retrieved, our test was successful
        assert len(result) == 1
        assert result[0][1] == uuid_value


@pytest.mark.parametrize(
    'uuid_string',
    uuid_params
)
@pytest.mark.asyncio
async def test_uuid_representation_async(async_client, uuid_string, uuid_table_async):
    """ Test if textual UUID representation is correct. """
    uuid_value = uuid.UUID(uuid_string)

    # use uuid.UUID class to insert data
    await async_client.sql(uuid_table_insert_sql, query_args=[1, uuid_value])
    # use hex string to retrieve data
    async with async_client.sql(uuid_table_query_sql, query_args=[str(uuid_value)]) as cursor:
        result = [row async for row in cursor]

        # if a line was retrieved, our test was successful
        assert len(result) == 1
        assert result[0][1] == uuid_value