18649071910

Committed 20 Oct 2025 09:42AM UTC coverage: 86.907% (+2.9%) from 84.048%

Build # 18649071910

Build Type

push

github

Committed by

web-flow

Commit Message

Bump to 4.0.0 (#443)

Run Details

41 of 54 new or added lines in 5 files covered. (75.93%)

125 existing lines in 4 files now uncovered.

1593 of 1833 relevant lines covered (86.91%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.61

/test/test_sqlalchemy_dialect_suite.py

# pylint: disable=wildcard-import,unused-wildcard-import,function-redefined,arguments-differ
# pylint: disable=redefined-outer-name
"""SQLAlchemy dialect test suite."""

from __future__ import annotations

from typing import Any

import pytest
import sqlalchemy.types as sql_types
from sqlalchemy import (
    ForeignKey,
    ForeignKeyConstraint,
    Integer,
    String,
    inspect,
    testing,
)
from sqlalchemy.exc import DBAPIError
from sqlalchemy.testing.assertions import assert_raises, eq_
from sqlalchemy.testing.provision import temp_table_keyword_args
from sqlalchemy.testing.schema import Column, Table
from sqlalchemy.testing.suite import *  # noqa: F401, F403
from sqlalchemy.testing.suite.test_cte import CTETest as _CTETest
from sqlalchemy.testing.suite.test_reflection import (
    BizarroCharacterTest as _BizarroCharacterTest,
)
from sqlalchemy.testing.suite.test_reflection import (
    ComponentReflectionTest as _ComponentReflectionTest,
)
from sqlalchemy.testing.suite.test_reflection import (
    ComponentReflectionTestExtra as _ComponentReflectionTestExtra,
)
from sqlalchemy.testing.suite.test_reflection import (
    IdentityReflectionTest as _IdentityReflectionTest,
)
from sqlalchemy.testing.suite.test_reflection import (
    TempTableElementsTest as _TempTableElementsTest,
)
from sqlalchemy.testing.suite.test_select import (
    IdentityColumnTest as _IdentityColumnTest,
)
from sqlalchemy.testing.suite.test_types import JSONTest as _JSONTest

# Import dialect test suite provided by SQLAlchemy into SQLAlchemy-HANA test collection.
# Please don't add other tests in this file. Only adjust or overview SQLAlchemy tests
# for compatibility with SAP HANA.


@temp_table_keyword_args.for_db("*")
def _temp_table_keyword_args(*args: Any, **kwargs: Any) -> dict[str, list[str]]:
    return {
        "prefixes": ["GLOBAL TEMPORARY"],
    }


class ComponentReflectionTestExtra(_ComponentReflectionTestExtra):
    @testing.combinations(
        ("CASCADE", None),
        (None, "SET NULL"),
        (None, "RESTRICT"),
        (None, "RESTRICT"),
        argnames="ondelete,onupdate",
    )
    def test_get_foreign_key_options(self, connection, metadata, ondelete, onupdate):
        options = {}
        if ondelete:
            options["ondelete"] = ondelete
        if onupdate:
            options["onupdate"] = onupdate

        options.setdefault("ondelete", "RESTRICT")
        options.setdefault("onupdate", "RESTRICT")

        Table(
            "x",
            metadata,
            Column("id", Integer, primary_key=True),
            test_needs_fk=True,
        )

        Table(
            "table",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("x_id", Integer, ForeignKey("x.id", name="xid")),
            Column("test", String(10)),
            test_needs_fk=True,
        )

        Table(
            "user",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(50), nullable=False),
            Column("tid", Integer),
            ForeignKeyConstraint(["tid"], ["table.id"], name="myfk", **options),
            test_needs_fk=True,
        )

        metadata.create_all(connection)

        insp = inspect(connection)

        # test 'options' is always present for a backend
        # that can reflect these, since alembic looks for this
        opts = insp.get_foreign_keys("table")[0]["options"]

        eq_(
            {k: opts[k] for k in opts if opts[k]},
            {"onupdate": "RESTRICT", "ondelete": "RESTRICT"},
        )

        opts = insp.get_foreign_keys("user")[0]["options"]
        eq_(opts, options)

    @testing.requires.table_reflection
    @testing.combinations(
        sql_types.String,
        sql_types.VARCHAR,
        sql_types.CHAR,
        (sql_types.NVARCHAR, testing.requires.nvarchar_types),
        (sql_types.NCHAR, testing.requires.nvarchar_types),
        argnames="type_",
    )
    def test_string_length_reflection(self, connection, metadata, type_):
        typ = self._type_round_trip(connection, metadata, type_(52))[0]
        if issubclass(type_, sql_types.VARCHAR):
            assert isinstance(typ, sql_types.VARCHAR | sql_types.NVARCHAR)
        elif issubclass(type_, sql_types.CHAR):
            assert isinstance(typ, sql_types.CHAR | sql_types.NCHAR)
        else:
            assert isinstance(typ, sql_types.String)

        eq_(typ.length, 52)
        assert isinstance(typ.length, int)


class ComponentReflectionTest(_ComponentReflectionTest):
    def _check_table_dict(self, result, exp, req_keys=None, make_lists=False):
        def _normalize(data):
            for entry in data.values():
                if isinstance(entry, list):
                    for subentry in entry:
                        if "sqltext" not in subentry:
                            continue
                        sqltext = subentry["sqltext"]
                        # ignore casing
                        sqltext = sqltext.lower()
                        # removes spaces because they are not consistent
                        sqltext = sqltext.replace(" ", "")
                        # remove quotes because they are not consistent
                        sqltext = sqltext.replace('"', "")
                        # remove a potential leading (
                        if sqltext.startswith("("):
                            sqltext = sqltext[1:]
                        # remove a potential closing )
                        if sqltext.endswith(")"):
                            sqltext = sqltext[:-1]
                        subentry["sqltext"] = sqltext

        _normalize(result)
        _normalize(exp)
        return super()._check_table_dict(result, exp, req_keys, make_lists)


class CTETest(_CTETest):
    def test_select_recursive_round_trip(self, connection):
        pytest.skip("Recursive CTEs are not supported by SAP HANA")

    def test_insert_from_select_round_trip(self, connection):
        pytest.skip("Insert CTEs are not supported by SAP HANA")


class IdentityReflectionTest(_IdentityReflectionTest):
    def test_reflect_identity(self, connection):
        pytest.skip("Identity column reflection is not supported")

    def test_reflect_identity_schema(self, connection):
        pytest.skip("Identity column reflection is not supported")


class IdentityColumnTest(_IdentityColumnTest):
    def test_insert_always_error(self, connection):
        def fn():
            connection.execute(
                self.tables.tbl_a.insert(),
                [{"id": 200, "desc": "a"}],
            )

        assert_raises((DBAPIError,), fn)


class JSONTest(_JSONTest):
    @classmethod
    def define_tables(cls, metadata):
        Table(
            "data_table",
            metadata,
            Column("id", Integer, autoincrement=True, primary_key=True),
            Column("name", String(30), nullable=False),
            Column("data", cls.datatype, nullable=False),
            Column("nulldata", cls.datatype(none_as_null=True)),
        )

    def test_index_typed_access(self):
        pytest.skip("Index typed access is not supported")

    def test_index_typed_comparison(self):
        pytest.skip("Index typed access is not supported")

    def test_path_typed_comparison(self):
        pytest.skip("Path typed access is not supported")


class BizarroCharacterTest(_BizarroCharacterTest):

    def test_reflect_identity(self):
        pytest.skip("Identity reflection is not supported")


class TempTableElementsTest(_TempTableElementsTest):

    def test_reflect_identity(self):
        pytest.skip("Identity reflection is not supported")

1	# pylint: disable=wildcard-import,unused-wildcard-import,function-redefined,arguments-differ
2	# pylint: disable=redefined-outer-name
3	"""SQLAlchemy dialect test suite."""
4
UNCOV 5	from __future__ import annotations	×
6
UNCOV 7	from typing import Any	×
8
9	import pytest	1✔
10	import sqlalchemy.types as sql_types	1✔
11	from sqlalchemy import (	1✔
12	ForeignKey,
13	ForeignKeyConstraint,
14	Integer,
15	String,
16	inspect,
17	testing,
18	)
19	from sqlalchemy.exc import DBAPIError	1✔
20	from sqlalchemy.testing.assertions import assert_raises, eq_	1✔
21	from sqlalchemy.testing.provision import temp_table_keyword_args	1✔
22	from sqlalchemy.testing.schema import Column, Table	1✔
23	from sqlalchemy.testing.suite import * # noqa: F401, F403	1✔
24	from sqlalchemy.testing.suite.test_cte import CTETest as _CTETest	1✔
25	from sqlalchemy.testing.suite.test_reflection import (	1✔
26	BizarroCharacterTest as _BizarroCharacterTest,
27	)
28	from sqlalchemy.testing.suite.test_reflection import (	1✔
29	ComponentReflectionTest as _ComponentReflectionTest,
30	)
31	from sqlalchemy.testing.suite.test_reflection import (	1✔
32	ComponentReflectionTestExtra as _ComponentReflectionTestExtra,
33	)
34	from sqlalchemy.testing.suite.test_reflection import (	1✔
35	IdentityReflectionTest as _IdentityReflectionTest,
36	)
37	from sqlalchemy.testing.suite.test_reflection import (	1✔
38	TempTableElementsTest as _TempTableElementsTest,
39	)
40	from sqlalchemy.testing.suite.test_select import (	1✔
41	IdentityColumnTest as _IdentityColumnTest,
42	)
43	from sqlalchemy.testing.suite.test_types import JSONTest as _JSONTest	×
44
45	# Import dialect test suite provided by SQLAlchemy into SQLAlchemy-HANA test collection.
46	# Please don't add other tests in this file. Only adjust or overview SQLAlchemy tests
47	# for compatibility with SAP HANA.
48
49
50	@temp_table_keyword_args.for_db("*")	1✔
51	def _temp_table_keyword_args(args: Any, *kwargs: Any) -> dict[str, list[str]]:	1✔
52	return {	1✔
53	"prefixes": ["GLOBAL TEMPORARY"],
54	}
55
56
57	class ComponentReflectionTestExtra(_ComponentReflectionTestExtra):	1✔
58	@testing.combinations(	1✔
59	("CASCADE", None),
60	(None, "SET NULL"),
61	(None, "RESTRICT"),
62	(None, "RESTRICT"),
63	argnames="ondelete,onupdate",
64	)
65	def test_get_foreign_key_options(self, connection, metadata, ondelete, onupdate):	1✔
66	options = {}	1✔
67	if ondelete:	1✔
68	options["ondelete"] = ondelete	1✔
69	if onupdate:	1✔
UNCOV 70	options["onupdate"] = onupdate	×
71
72	options.setdefault("ondelete", "RESTRICT")	1✔
UNCOV 73	options.setdefault("onupdate", "RESTRICT")	×
74
75	Table(	1✔
76	"x",
77	metadata,
78	Column("id", Integer, primary_key=True),
79	test_needs_fk=True,
80	)
81
82	Table(	1✔
83	"table",
84	metadata,
85	Column("id", Integer, primary_key=True),
86	Column("x_id", Integer, ForeignKey("x.id", name="xid")),
87	Column("test", String(10)),
88	test_needs_fk=True,
89	)
90
91	Table(	1✔
92	"user",
93	metadata,
94	Column("id", Integer, primary_key=True),
95	Column("name", String(50), nullable=False),
96	Column("tid", Integer),
97	ForeignKeyConstraint(["tid"], ["table.id"], name="myfk", **options),
98	test_needs_fk=True,
99	)
100
UNCOV 101	metadata.create_all(connection)	×
102
UNCOV 103	insp = inspect(connection)	×
104
105	# test 'options' is always present for a backend
106	# that can reflect these, since alembic looks for this
107	opts = insp.get_foreign_keys("table")[0]["options"]	×
108
109	eq_(	1✔
110	{k: opts[k] for k in opts if opts[k]},
111	{"onupdate": "RESTRICT", "ondelete": "RESTRICT"},
112	)
113
114	opts = insp.get_foreign_keys("user")[0]["options"]	1✔
UNCOV 115	eq_(opts, options)	×
116
117	@testing.requires.table_reflection	1✔
118	@testing.combinations(	1✔
119	sql_types.String,
120	sql_types.VARCHAR,
121	sql_types.CHAR,
122	(sql_types.NVARCHAR, testing.requires.nvarchar_types),
123	(sql_types.NCHAR, testing.requires.nvarchar_types),
124	argnames="type_",
125	)
126	def test_string_length_reflection(self, connection, metadata, type_):	1✔
127	typ = self._type_round_trip(connection, metadata, type_(52))[0]	1✔
128	if issubclass(type_, sql_types.VARCHAR):	1✔
129	assert isinstance(typ, sql_types.VARCHAR \| sql_types.NVARCHAR)	1✔
130	elif issubclass(type_, sql_types.CHAR):	1✔
NEW 131	assert isinstance(typ, sql_types.CHAR \| sql_types.NCHAR)	×
132	else:
UNCOV 133	assert isinstance(typ, sql_types.String)	×
134
135	eq_(typ.length, 52)	1✔
UNCOV 136	assert isinstance(typ.length, int)	×
137
138
139	class ComponentReflectionTest(_ComponentReflectionTest):	1✔
140	def _check_table_dict(self, result, exp, req_keys=None, make_lists=False):	1✔
141	def _normalize(data):	1✔
142	for entry in data.values():	1✔
143	if isinstance(entry, list):	1✔
144	for subentry in entry:	1✔
145	if "sqltext" not in subentry:	1✔
146	continue	1✔
UNCOV 147	sqltext = subentry["sqltext"]	×
148	# ignore casing
UNCOV 149	sqltext = sqltext.lower()	×
150	# removes spaces because they are not consistent
151	sqltext = sqltext.replace(" ", "")	×
152	# remove quotes because they are not consistent
UNCOV 153	sqltext = sqltext.replace('"', "")	×
154	# remove a potential leading (
155	if sqltext.startswith("("):	1✔
UNCOV 156	sqltext = sqltext[1:]	×
157	# remove a potential closing )
158	if sqltext.endswith(")"):	1✔
159	sqltext = sqltext[:-1]	1✔
160	subentry["sqltext"] = sqltext	×
161
162	_normalize(result)	1✔
163	_normalize(exp)	1✔
164	return super()._check_table_dict(result, exp, req_keys, make_lists)	×
165
166
167	class CTETest(_CTETest):	1✔
168	def test_select_recursive_round_trip(self, connection):	1✔
UNCOV 169	pytest.skip("Recursive CTEs are not supported by SAP HANA")	×
170
171	def test_insert_from_select_round_trip(self, connection):	1✔
172	pytest.skip("Insert CTEs are not supported by SAP HANA")	×
173
174
175	class IdentityReflectionTest(_IdentityReflectionTest):	1✔
176	def test_reflect_identity(self, connection):	1✔
UNCOV 177	pytest.skip("Identity column reflection is not supported")	×
178
179	def test_reflect_identity_schema(self, connection):	1✔
UNCOV 180	pytest.skip("Identity column reflection is not supported")	×
181
182
183	class IdentityColumnTest(_IdentityColumnTest):	1✔
184	def test_insert_always_error(self, connection):	1✔
185	def fn():	1✔
186	connection.execute(	1✔
187	self.tables.tbl_a.insert(),
188	[{"id": 200, "desc": "a"}],
189	)
190
UNCOV 191	assert_raises((DBAPIError,), fn)	×
192
193
194	class JSONTest(_JSONTest):	1✔
195	@classmethod	1✔
196	def define_tables(cls, metadata):	1✔
197	Table(	1✔
198	"data_table",
199	metadata,
200	Column("id", Integer, autoincrement=True, primary_key=True),
201	Column("name", String(30), nullable=False),
202	Column("data", cls.datatype, nullable=False),
203	Column("nulldata", cls.datatype(none_as_null=True)),
204	)
205
206	def test_index_typed_access(self):	1✔
UNCOV 207	pytest.skip("Index typed access is not supported")	×
208
209	def test_index_typed_comparison(self):	1✔
UNCOV 210	pytest.skip("Index typed access is not supported")	×
211
212	def test_path_typed_comparison(self):	1✔
UNCOV 213	pytest.skip("Path typed access is not supported")	×
214
215
UNCOV 216	class BizarroCharacterTest(_BizarroCharacterTest):	×
217
218	def test_reflect_identity(self):	1✔
UNCOV 219	pytest.skip("Identity reflection is not supported")	×
220
221
222	class TempTableElementsTest(_TempTableElementsTest):	×
223
224	def test_reflect_identity(self):	1✔
225	pytest.skip("Identity reflection is not supported")	×

SAP / sqlalchemy-hana / 18649071910

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous