test_hana_types.py

# Copyright 2022 SAP SE.
#
# 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 datetime
import decimal
import random
import pytest
import sqlalchemy
import sqlalchemy.testing.suite.test_types
from sqlalchemy.testing.suite.test_types import _DateFixture

import sqlalchemy_hana
import sqlalchemy_hana.types

class HANATypesTest(sqlalchemy.testing.fixtures.TestBase):

    __dialect__ = "hana"

    # As documented in "SAP HANA SQL Reference Guide for SAP HANA Platform - Data Types"
    # https://help.sap.com/docs/SAP_HANA_PLATFORM/4fe29514fd584807ac9f2a04f6754767/20a1569875191014b507cf392724b7eb.html
    DOCUMENTED_HANA_TYPES = (
        "DATE", "TIME", "SECONDDATE", "TIMESTAMP",
        "TINYINT", "SMALLINT", "INTEGER", "BIGINT", "SMALLDECIMAL", "DECIMAL", "REAL", "DOUBLE",
        "BOOLEAN",
        "VARCHAR", "NVARCHAR", "ALPHANUM", "SHORTTEXT",
        "VARBINARY",
        "BLOB", 'CLOB', "NCLOB", "TEXT",
        # Not yet supported types:
        # "ARRAY",
        # "ST_GEOMETRY", "ST_POINT"
    )
    # Some types are not documented but still usable and could occur during table reflection
    UNDOCUMENTED_HANA_TYPES = (
        "CHAR", "NCHAR",
    )

    @pytest.mark.parametrize("type_name", DOCUMENTED_HANA_TYPES)
    def test_documented_hana_types_are_available_as_class(self, type_name):
        assert hasattr(sqlalchemy_hana.types, type_name)

    @pytest.mark.parametrize("type_name", UNDOCUMENTED_HANA_TYPES)
    def test_undocumented_hana_types_are_available_as_class(self, type_name):
        assert hasattr(sqlalchemy_hana.types, type_name)


class HANASECONDDATETest(_DateFixture, sqlalchemy.testing.fixtures.TablesTest):
    __backend__ = True
    datatype = sqlalchemy_hana.types.SECONDDATE
    data = datetime.datetime(2012, 10, 15, 12, 57, 18, 12345)
    compare = datetime.datetime(2012, 10, 15, 12, 57, 18)


class _HANATypeTest(sqlalchemy.testing.fixtures.TablesTest):

    compare = None
    hana_version_skip = None

    @classmethod
    def _init_class(cls):
        super()._init_class()

        if cls.hana_version_skip:
            if cls.bind.dialect.server_version_info[0] == cls.hana_version_skip:
                pytest.skip(
                    (
                        "Test/Feature is incompatible with SAP HANA version {} "
                        "(connected to SAP HANA {})"
                    ).format(
                        cls.hana_version_skip,
                        '.'.join((str(i) for i in cls.bind.dialect.server_version_info))
                    )
                )

    @classmethod
    def define_tables(cls, metadata):
        sqlalchemy.Table(
            "test_type",
            metadata,
            sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
            sqlalchemy.Column("data", cls.column_type)
        )

    @property
    def reflected_column_type(self):
        return self.column_type

    def test_round_trip(self, connection):
        test_type = self.tables.test_type

        connection.execute(
            test_type.insert(), {"id": 1, "data": self.data}
        )

        row = connection.execute(
            sqlalchemy.select([test_type.c.data])
        ).first()

        compare = self.compare or self.data
        assert row == (compare,)
        assert isinstance(row[0], type(compare))

    def test_reflection(self, connection):
        columns = sqlalchemy.inspect(connection).get_columns("test_type")
        assert isinstance(columns[1]['type'], self.reflected_column_type.__class__)
        assert repr(columns[1]['type']) == repr(self.reflected_column_type)


class _HANAIntegerTypeTest(_HANATypeTest):

    def test_round_trip(self, connection):
        self.data = random.randint(self.min_value, self.max_value)
        super().test_round_trip(connection)

    def test_underflow_int(self, connection):
        test_type = self.tables.test_type

        with pytest.raises(sqlalchemy.exc.DBAPIError):
            connection.execute(
                test_type.insert(), {"id": 1, "data": self.min_value - 1}
            )

    def test_overflow_int(self, connection):
        test_type = self.tables.test_type

        with pytest.raises(sqlalchemy.exc.DBAPIError):
            connection.execute(
                test_type.insert(), {"id": 1, "data": self.max_value + 1}
            )


class HANATINYINTTest(_HANAIntegerTypeTest):

    column_type = sqlalchemy_hana.types.TINYINT()
    min_value = 0
    max_value = 255


class HANASMALLINTTest(_HANAIntegerTypeTest):

    column_type = sqlalchemy_hana.types.SMALLINT()
    min_value = -32768
    max_value = 32767


class HANAINTEGERTest(_HANAIntegerTypeTest):

    column_type = sqlalchemy_hana.types.INTEGER()
    min_value = -2147483648
    max_value = 2147483647


class HANABIGINTTest(_HANAIntegerTypeTest):

    column_type = sqlalchemy_hana.types.BIGINT()
    min_value = -9_223_372_036_854_775_808
    max_value = 9_223_372_036_854_775_807


class _HANADecimalTypeTest(_HANATypeTest):
    pass


class HANASMALLDECIMALTest(_HANADecimalTypeTest):

    column_type = sqlalchemy_hana.types.SMALLDECIMAL()
    data = decimal.Decimal("3.14")


class HANADECIMALTest(_HANADecimalTypeTest):

    column_type = sqlalchemy_hana.types.DECIMAL(5, 4)
    data = decimal.Decimal("3.141592")
    compare = decimal.Decimal("3.1415")


class HANAREALTest(_HANADecimalTypeTest):

    column_type = sqlalchemy_hana.types.REAL()
    data = 3.141592
    compare = 3.141592025756836

class HANADOUBLETest(_HANADecimalTypeTest):

    column_type = sqlalchemy_hana.types.DOUBLE()
    data = 3.141592

class HANABOOLEANTest(_HANATypeTest):

    column_type = sqlalchemy_hana.types.BOOLEAN()
    data = True


class HANAVARCHARTest(_HANATypeTest):

    column_type = sqlalchemy_hana.types.VARCHAR(length=15)
    data = "Some text"

    @property
    def reflected_column_type(self):
        if self.bind.dialect.server_version_info[0] == 4:
            return sqlalchemy_hana.types.NVARCHAR(length=15)
        else:
            return self.column_type


class HANANVARCHARTest(_HANATypeTest):

    column_type = sqlalchemy_hana.types.NVARCHAR(length=15)
    data = "Some text"


class HANANVARCHAROperatorTest(sqlalchemy.testing.fixtures.TablesTest, sqlalchemy.testing.AssertsCompiledSQL):

    __only_on__ = "hana"
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        sqlalchemy.Table(
            "data_table",
            metadata,
            sqlalchemy.Column("id", sqlalchemy.types.Integer, primary_key=True),
            sqlalchemy.Column("data", sqlalchemy_hana.types.NVARCHAR(50)),
        )

    @classmethod
    def insert_data(cls, connection):
        connection.execute(
            cls.tables.data_table.insert(),
            [
                {"id": 1, "data": "abc123"},
                {"id": 2, "data": "abc"},
                {"id": 3, "data": "123"}
            ],
        )

    def test_nvarchar_supports_like_regex(self):
        data_table = self.tables.data_table
        query = sqlalchemy.select([data_table]).where(data_table.c.data.like_regex(r'^[a-z]+\d+$'))
        self.assert_compile(
            query,
            "SELECT data_table.id, data_table.data FROM data_table WHERE data_table.data LIKE_REGEXPR ?"
        )
        assert sqlalchemy.testing.db.execute(query).fetchall() == [(1, 'abc123'),]

    def test_nvarchar_supports_negate_like_regex(self):
        data_table = self.tables.data_table
        query = sqlalchemy.select([data_table]).where(~data_table.c.data.like_regex(r'\d+'))
        self.assert_compile(
            query,
            "SELECT data_table.id, data_table.data FROM data_table WHERE data_table.data NOT LIKE_REGEXPR ?"
        )
        assert sqlalchemy.testing.db.execute(query).fetchall() == [(2, 'abc'),]

    def test_nvarchar_supports_not_like_regex(self):
        data_table = self.tables.data_table
        query = sqlalchemy.select([data_table]).where(data_table.c.data.notlike_regex(r'[a-z]+'))
        self.assert_compile(
            query,
            "SELECT data_table.id, data_table.data FROM data_table WHERE data_table.data NOT LIKE_REGEXPR ?"
        )
        assert sqlalchemy.testing.db.execute(query).fetchall() == [(3, '123'),]

    def test_nvarchar_supports_negate_not_like_regex(self):
        data_table = self.tables.data_table
        query = sqlalchemy.select([data_table]).where(~data_table.c.data.notlike_regex(r'\s+'))
        self.assert_compile(
            query,
            "SELECT data_table.id, data_table.data FROM data_table WHERE data_table.data LIKE_REGEXPR ?"
        )
        assert sqlalchemy.testing.db.execute(query).fetchall() == []


class HANAALPHANUMTest(_HANATypeTest):

    hana_version_skip = 4

    column_type = sqlalchemy_hana.types.ALPHANUM(length=60)
    data = "Some alpha numeric text with numbers 1234567899"


class HANASHORTTEXTTest(_HANATypeTest):

    hana_version_skip = 4

    column_type = sqlalchemy_hana.types.SHORTTEXT(length=60)
    data = "Some data for a HANA short text"


class HANACHARTest(_HANATypeTest):

    column_type = sqlalchemy_hana.types.CHAR(length=60)
    data = "Some data for a HANA short text"

    @property
    def reflected_column_type(self):
        if self.bind.dialect.server_version_info[0] == 4:
            return sqlalchemy_hana.types.NCHAR(length=60)
        else:
            return self.column_type


class HANANCHARTest(_HANATypeTest):

    column_type = sqlalchemy_hana.types.NCHAR(length=60)
    data = "Some data for a HANA short text"


class HANAVARBINARYTest(_HANATypeTest):

    column_type = sqlalchemy_hana.types.VARBINARY(length=60)
    data = b"Some binary for a HANA VARBINARY"


class _HANALOBTest(_HANATypeTest):
    pass

class HANABLOBTest(_HANALOBTest):

    column_type = sqlalchemy_hana.types.BLOB()
    data = b"some binary data"


class HANACLOBTest(_HANALOBTest):

    column_type = sqlalchemy_hana.types.CLOB()
    data = "some test text"

    @property
    def reflected_column_type(self):
        if self.bind.dialect.server_version_info[0] == 4:
            return sqlalchemy_hana.types.NCLOB()
        else:
            return self.column_type


class HANANCBLOBTest(_HANALOBTest):

    column_type = sqlalchemy_hana.types.NCLOB()
    data = "some test text"

class HANATEXTTest(_HANALOBTest):

    hana_version_skip = 4

    column_type = sqlalchemy_hana.types.TEXT()
    data = "some test text"