21434659066

Committed 28 Jan 2026 07:33AM UTC coverage: 84.984% (-11.1%) from 96.083%

Build # 21434659066

Build Type

push

github

Committed by

web-flow

Commit Message

Support asyncio execution (#522)

Run Details

1562 of 1838 relevant lines covered (84.98%)

0.85 hits per line

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

82.11

/test/test_dialect.py

"""SAP HANA Dialect testing."""

from __future__ import annotations

import sys
from unittest import mock
from unittest.mock import Mock

import pytest
from hdbcli.dbapi import Error
from sqlalchemy import create_engine
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.engine.url import make_url
from sqlalchemy.exc import ArgumentError, DBAPIError
from sqlalchemy.testing import assert_raises_message, config, eq_
from sqlalchemy.testing.engines import testing_engine
from sqlalchemy.testing.fixtures import TestBase

DEFAULT_ISOLATION_LEVEL = "READ COMMITTED"
NON_DEFAULT_ISOLATION_LEVEL = "SERIALIZABLE"


class DialectTest(TestBase):
    def test_detection_by_error_code(self) -> None:
        dialect = config.db.dialect
        assert dialect.is_disconnect(Error(-10709, "Connect failed"), None, None)

    def test_detection_by_isconnected_function(self) -> None:
        dialect = config.db.dialect

        mock_connection = Mock(isconnected=Mock(return_value=False))
        assert dialect.is_disconnect(None, mock_connection, None)

        mock_connection = Mock(isconnected=Mock(return_value=True))
        assert not dialect.is_disconnect(None, mock_connection, None)

    @pytest.mark.parametrize(
        "kwargs,supports_native_boolean",
        [
            ({}, True),
            ({"use_native_boolean": True}, True),
            ({"use_native_boolean": False}, False),
        ],
    )
    def test_supports_native_boolean(
        self, kwargs: dict, supports_native_boolean: bool
    ) -> None:
        engine = create_engine("hana://username:secret-password@example.com", **kwargs)
        assert engine.dialect.supports_native_boolean == supports_native_boolean

    def test_hdbcli_tenant_url_default_port(self) -> None:
        """If the URL includes a tenant database, the dialect pass the adjusted values to hdbcli.

        Beside the parameter databaseName, it should also adjust the default port to the SYSTEMDB
        SQL port for HANA's automated tenant redirect as the SQL ports of tenant datbases are are
        transient.
        """
        _, result_kwargs = config.db.dialect.create_connect_args(
            make_url("hana://username:secret-password@example.com/TENANT_NAME")
        )
        assert result_kwargs["address"] == "example.com"
        assert result_kwargs["port"] == 30013
        assert result_kwargs["user"] == "username"
        assert result_kwargs["password"] == "secret-password"
        assert result_kwargs["databaseName"] == "TENANT_NAME"

    def test_hdbcli_tenant_url_changed_port(self) -> None:
        """If the URL includes a tenant database, the dialect pass the adjusted values to hdbcli.

        It doesn't adjust the port if the user explicitly defined it.
        """
        _, result_kwargs = config.db.dialect.create_connect_args(
            make_url("hana://username:secret-password@example.com:30041/TENANT_NAME")
        )
        assert result_kwargs["address"] == "example.com"
        assert result_kwargs["port"] == 30041
        assert result_kwargs["user"] == "username"
        assert result_kwargs["password"] == "secret-password"
        assert result_kwargs["databaseName"] == "TENANT_NAME"

    def test_parsing_userkey_hdbcli(self) -> None:
        """Test that a userkey can be parsed.

        With HDBCLI, the user may reference to a local HDBUserStore key which holds
        the connection details. SQLAlchemy-HANA should only pass the userkey name to
        HDBCLI for the connection creation.
        """
        _, result_kwargs = config.db.dialect.create_connect_args(
            make_url("hana://userkey=myuserkeyname")
        )
        assert result_kwargs == {"userkey": "myuserkeyname", "vectoroutputtype": "list"}

    def test_pass_uri_query_as_kwargs(self) -> None:
        """SQLAlchemy-HANA should passes all URL parameters to hdbcli."""
        urls = [
            "hana://username:secret-password@example.com/?encrypt=true&compress=true",
            "hana://username:secret-password@example.com/TENANT_NAME?encrypt=true&compress=true",
        ]

        for url in urls:
            _, result_kwargs = config.db.dialect.create_connect_args(make_url(url))
            assert result_kwargs["encrypt"] == "true"
            assert result_kwargs["compress"] == "true"

    def test_server_version_info(self) -> None:
        # Test that the attribute is defined
        assert config.db.dialect.server_version_info

    def test_get_isolation_level(self) -> None:
        eng = testing_engine(options={})
        isolation_level = eng.dialect.get_isolation_level(eng.connect().connection)
        eq_(isolation_level, DEFAULT_ISOLATION_LEVEL)

    def test_set_isolation_level(self) -> None:
        eng = testing_engine(options={})
        conn = eng.connect()
        eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)

        eng.dialect.set_isolation_level(conn.connection, NON_DEFAULT_ISOLATION_LEVEL)

        eq_(
            eng.dialect.get_isolation_level(conn.connection),
            NON_DEFAULT_ISOLATION_LEVEL,
        )
        conn.close()

    def test_reset_level(self) -> None:
        eng = testing_engine(options={})
        conn = eng.connect()
        eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)

        eng.dialect.set_isolation_level(conn.connection, NON_DEFAULT_ISOLATION_LEVEL)
        eq_(
            eng.dialect.get_isolation_level(conn.connection),
            NON_DEFAULT_ISOLATION_LEVEL,
        )

        eng.dialect.reset_isolation_level(conn.connection)
        eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)
        conn.close()

    def test_set_level_with_setting(self) -> None:
        eng = testing_engine(options={"isolation_level": NON_DEFAULT_ISOLATION_LEVEL})
        conn = eng.connect()
        eq_(
            eng.dialect.get_isolation_level(conn.connection),
            NON_DEFAULT_ISOLATION_LEVEL,
        )

        eng.dialect.set_isolation_level(conn.connection, DEFAULT_ISOLATION_LEVEL)
        eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)
        conn.close()

    def test_invalid_level(self) -> None:
        eng = testing_engine(options={"isolation_level": "FOO"})
        levels = ", ".join(eng.dialect._isolation_lookup)
        assert_raises_message(
            ArgumentError,
            "Invalid value 'FOO' for isolation_level. "
            f"Valid isolation levels for {eng.dialect.name} are {levels}",
            eng.connect,
        )

    def test_with_execution_options(self) -> None:
        eng = create_engine(
            config.db.url,
            execution_options={"isolation_level": NON_DEFAULT_ISOLATION_LEVEL},
        )
        conn = eng.connect()
        eq_(
            eng.dialect.get_isolation_level(conn.connection),
            NON_DEFAULT_ISOLATION_LEVEL,
        )
        conn.close()

    def test_with_isolation_level_in_create_engine(self) -> None:
        eng = create_engine(config.db.url, isolation_level=NON_DEFAULT_ISOLATION_LEVEL)
        conn = eng.connect()
        eq_(
            eng.dialect.get_isolation_level(conn.connection),
            NON_DEFAULT_ISOLATION_LEVEL,
        )
        conn.close()

    def test_do_rollback_to_savepoint_no_error(self) -> None:
        dialect = config.db.dialect
        connection = Mock()

        with mock.patch.object(
            DefaultDialect, "do_rollback_to_savepoint"
        ) as super_rollback:
            dialect.do_rollback_to_savepoint(connection, "savepoint")
            super_rollback.assert_called_once_with(connection, "savepoint")

    def test_do_rollback_to_savepoint_unrelated_error(self) -> None:
        dialect = config.db.dialect
        connection = Mock()

        with (
            mock.patch.object(
                sys, "exc_info", return_value=(ValueError, ValueError(), Mock())
            ),
            mock.patch.object(
                DefaultDialect, "do_rollback_to_savepoint"
            ) as super_rollback,
        ):
            dialect.do_rollback_to_savepoint(connection, "savepoint")
            super_rollback.assert_called_once_with(connection, "savepoint")

    def test_do_rollback_to_savepoint_ignores_error(self) -> None:
        dialect = config.db.dialect
        connection = Mock()

        error = Error(133, "transaction rolled back: deadlock")
        dbapi_error = DBAPIError(None, None, error)

        with (
            mock.patch.object(
                sys, "exc_info", return_value=(DBAPIError, dbapi_error, Mock())
            ),
            mock.patch.object(
                DefaultDialect, "do_rollback_to_savepoint"
            ) as super_rollback,
        ):
            dialect.do_rollback_to_savepoint(connection, "savepoint")
            super_rollback.assert_not_called()

    def test_vectoroutputtype_is_blocked(self) -> None:
        url = "hana://username:secret-password@example.com/?encrypt=true&vectoroutputtype=list"
        with pytest.raises(ValueError, match="vectoroutputtype"):
            config.db.dialect.create_connect_args(make_url(url))

    @pytest.mark.parametrize(
        "kwargs,vector_output_type",
        [
            ({}, "list"),
            ({"vector_output_type": "list"}, "list"),
            ({"vector_output_type": "memoryview"}, "memoryview"),
        ],
    )
    def test_vector_output_type(self, kwargs: dict, vector_output_type: str) -> None:
        engine = create_engine("hana://username:secret-password@example.com", **kwargs)
        assert engine.dialect.vector_output_type == vector_output_type

1	"""SAP HANA Dialect testing."""
2
3	from __future__ import annotations	1✔
4
5	import sys	1✔
6	from unittest import mock	1✔
7	from unittest.mock import Mock	1✔
8
9	import pytest	1✔
10	from hdbcli.dbapi import Error	1✔
11	from sqlalchemy import create_engine	1✔
12	from sqlalchemy.engine.default import DefaultDialect	1✔
13	from sqlalchemy.engine.url import make_url	1✔
14	from sqlalchemy.exc import ArgumentError, DBAPIError	1✔
15	from sqlalchemy.testing import assert_raises_message, config, eq_	1✔
16	from sqlalchemy.testing.engines import testing_engine	1✔
17	from sqlalchemy.testing.fixtures import TestBase	1✔
18
19	DEFAULT_ISOLATION_LEVEL = "READ COMMITTED"	1✔
20	NON_DEFAULT_ISOLATION_LEVEL = "SERIALIZABLE"	1✔
21
22
23	class DialectTest(TestBase):	1✔
24	def test_detection_by_error_code(self) -> None:	1✔
25	dialect = config.db.dialect	1✔
26	assert dialect.is_disconnect(Error(-10709, "Connect failed"), None, None)	1✔
27
28	def test_detection_by_isconnected_function(self) -> None:	1✔
29	dialect = config.db.dialect	1✔
30
31	mock_connection = Mock(isconnected=Mock(return_value=False))	1✔
32	assert dialect.is_disconnect(None, mock_connection, None)	1✔
33
34	mock_connection = Mock(isconnected=Mock(return_value=True))	1✔
35	assert not dialect.is_disconnect(None, mock_connection, None)	1✔
36
37	@pytest.mark.parametrize(	1✔
38	"kwargs,supports_native_boolean",
39	[
40	({}, True),
41	({"use_native_boolean": True}, True),
42	({"use_native_boolean": False}, False),
43	],
44	)
45	def test_supports_native_boolean(	1✔
46	self, kwargs: dict, supports_native_boolean: bool
47	) -> None:
48	engine = create_engine("hana://username:secret-password@example.com", **kwargs)	1✔
49	assert engine.dialect.supports_native_boolean == supports_native_boolean	1✔
50
51	def test_hdbcli_tenant_url_default_port(self) -> None:	1✔
52	"""If the URL includes a tenant database, the dialect pass the adjusted values to hdbcli.
53
54	Beside the parameter databaseName, it should also adjust the default port to the SYSTEMDB
55	SQL port for HANA's automated tenant redirect as the SQL ports of tenant datbases are are
56	transient.
57	"""
58	_, result_kwargs = config.db.dialect.create_connect_args(	1✔
59	make_url("hana://username:secret-password@example.com/TENANT_NAME")
60	)
61	assert result_kwargs["address"] == "example.com"	1✔
62	assert result_kwargs["port"] == 30013	1✔
63	assert result_kwargs["user"] == "username"	1✔
64	assert result_kwargs["password"] == "secret-password"	1✔
65	assert result_kwargs["databaseName"] == "TENANT_NAME"	1✔
66
67	def test_hdbcli_tenant_url_changed_port(self) -> None:	1✔
68	"""If the URL includes a tenant database, the dialect pass the adjusted values to hdbcli.
69
70	It doesn't adjust the port if the user explicitly defined it.
71	"""
72	_, result_kwargs = config.db.dialect.create_connect_args(	1✔
73	make_url("hana://username:secret-password@example.com:30041/TENANT_NAME")
74	)
75	assert result_kwargs["address"] == "example.com"	1✔
76	assert result_kwargs["port"] == 30041	1✔
77	assert result_kwargs["user"] == "username"	1✔
78	assert result_kwargs["password"] == "secret-password"	1✔
79	assert result_kwargs["databaseName"] == "TENANT_NAME"	1✔
80
81	def test_parsing_userkey_hdbcli(self) -> None:	1✔
82	"""Test that a userkey can be parsed.
83
84	With HDBCLI, the user may reference to a local HDBUserStore key which holds
85	the connection details. SQLAlchemy-HANA should only pass the userkey name to
86	HDBCLI for the connection creation.
87	"""
88	_, result_kwargs = config.db.dialect.create_connect_args(	1✔
89	make_url("hana://userkey=myuserkeyname")
90	)
91	assert result_kwargs == {"userkey": "myuserkeyname", "vectoroutputtype": "list"}	1✔
92
93	def test_pass_uri_query_as_kwargs(self) -> None:	1✔
94	"""SQLAlchemy-HANA should passes all URL parameters to hdbcli."""
95	urls = [	1✔
96	"hana://username:secret-password@example.com/?encrypt=true&compress=true",
97	"hana://username:secret-password@example.com/TENANT_NAME?encrypt=true&compress=true",
98	]
99
100	for url in urls:	1✔
101	_, result_kwargs = config.db.dialect.create_connect_args(make_url(url))	1✔
102	assert result_kwargs["encrypt"] == "true"	1✔
103	assert result_kwargs["compress"] == "true"	1✔
104
105	def test_server_version_info(self) -> None:	1✔
106	# Test that the attribute is defined
107	assert config.db.dialect.server_version_info	1✔
108
109	def test_get_isolation_level(self) -> None:	1✔
110	eng = testing_engine(options={})	1✔
111	isolation_level = eng.dialect.get_isolation_level(eng.connect().connection)	1✔
112	eq_(isolation_level, DEFAULT_ISOLATION_LEVEL)	1✔
113
114	def test_set_isolation_level(self) -> None:	×
115	eng = testing_engine(options={})	1✔
116	conn = eng.connect()	1✔
117	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	1✔
118
119	eng.dialect.set_isolation_level(conn.connection, NON_DEFAULT_ISOLATION_LEVEL)	×
120
121	eq_(	1✔
122	eng.dialect.get_isolation_level(conn.connection),
123	NON_DEFAULT_ISOLATION_LEVEL,
124	)
125	conn.close()	×
126
127	def test_reset_level(self) -> None:	×
128	eng = testing_engine(options={})	1✔
129	conn = eng.connect()	1✔
130	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	1✔
131
132	eng.dialect.set_isolation_level(conn.connection, NON_DEFAULT_ISOLATION_LEVEL)	×
133	eq_(	1✔
134	eng.dialect.get_isolation_level(conn.connection),
135	NON_DEFAULT_ISOLATION_LEVEL,
136	)
137
138	eng.dialect.reset_isolation_level(conn.connection)	×
139	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	1✔
140	conn.close()	1✔
141
142	def test_set_level_with_setting(self) -> None:	×
143	eng = testing_engine(options={"isolation_level": NON_DEFAULT_ISOLATION_LEVEL})	1✔
144	conn = eng.connect()	1✔
145	eq_(	1✔
146	eng.dialect.get_isolation_level(conn.connection),
147	NON_DEFAULT_ISOLATION_LEVEL,
148	)
149
150	eng.dialect.set_isolation_level(conn.connection, DEFAULT_ISOLATION_LEVEL)	×
151	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	1✔
152	conn.close()	1✔
153
154	def test_invalid_level(self) -> None:	×
155	eng = testing_engine(options={"isolation_level": "FOO"})	1✔
156	levels = ", ".join(eng.dialect._isolation_lookup)	1✔
157	assert_raises_message(	1✔
158	ArgumentError,
159	"Invalid value 'FOO' for isolation_level. "
160	f"Valid isolation levels for {eng.dialect.name} are {levels}",
161	eng.connect,
162	)
163
164	def test_with_execution_options(self) -> None:	×
165	eng = create_engine(	1✔
166	config.db.url,
167	execution_options={"isolation_level": NON_DEFAULT_ISOLATION_LEVEL},
168	)
169	conn = eng.connect()	×
170	eq_(	1✔
171	eng.dialect.get_isolation_level(conn.connection),
172	NON_DEFAULT_ISOLATION_LEVEL,
173	)
174	conn.close()	×
175
176	def test_with_isolation_level_in_create_engine(self) -> None:	×
177	eng = create_engine(config.db.url, isolation_level=NON_DEFAULT_ISOLATION_LEVEL)	1✔
178	conn = eng.connect()	1✔
179	eq_(	1✔
180	eng.dialect.get_isolation_level(conn.connection),
181	NON_DEFAULT_ISOLATION_LEVEL,
182	)
183	conn.close()	×
184
185	def test_do_rollback_to_savepoint_no_error(self) -> None:	×
186	dialect = config.db.dialect	1✔
187	connection = Mock()	1✔
188
189	with mock.patch.object(	1✔
190	DefaultDialect, "do_rollback_to_savepoint"
191	) as super_rollback:
192	dialect.do_rollback_to_savepoint(connection, "savepoint")	1✔
193	super_rollback.assert_called_once_with(connection, "savepoint")	1✔
194
195	def test_do_rollback_to_savepoint_unrelated_error(self) -> None:	×
196	dialect = config.db.dialect	1✔
197	connection = Mock()	1✔
198
199	with (	1✔
200	mock.patch.object(
201	sys, "exc_info", return_value=(ValueError, ValueError(), Mock())
202	),
203	mock.patch.object(
204	DefaultDialect, "do_rollback_to_savepoint"
205	) as super_rollback,
206	):
207	dialect.do_rollback_to_savepoint(connection, "savepoint")	×
208	super_rollback.assert_called_once_with(connection, "savepoint")	1✔
209
210	def test_do_rollback_to_savepoint_ignores_error(self) -> None:	×
211	dialect = config.db.dialect	1✔
212	connection = Mock()	1✔
213
214	error = Error(133, "transaction rolled back: deadlock")	×
215	dbapi_error = DBAPIError(None, None, error)	1✔
216
217	with (	1✔
218	mock.patch.object(
219	sys, "exc_info", return_value=(DBAPIError, dbapi_error, Mock())
220	),
221	mock.patch.object(
222	DefaultDialect, "do_rollback_to_savepoint"
223	) as super_rollback,
224	):
225	dialect.do_rollback_to_savepoint(connection, "savepoint")	×
226	super_rollback.assert_not_called()	1✔
227
228	def test_vectoroutputtype_is_blocked(self) -> None:	×
229	url = "hana://username:secret-password@example.com/?encrypt=true&vectoroutputtype=list"	1✔
230	with pytest.raises(ValueError, match="vectoroutputtype"):	1✔
231	config.db.dialect.create_connect_args(make_url(url))	1✔
232
233	@pytest.mark.parametrize(	1✔
234	"kwargs,vector_output_type",
235	[
236	({}, "list"),
237	({"vector_output_type": "list"}, "list"),
238	({"vector_output_type": "memoryview"}, "memoryview"),
239	],
240	)
241	def test_vector_output_type(self, kwargs: dict, vector_output_type: str) -> None:	×
242	engine = create_engine("hana://username:secret-password@example.com", **kwargs)	1✔
243	assert engine.dialect.vector_output_type == vector_output_type	1✔

SAP / sqlalchemy-hana / 21434659066

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