15992939927

Committed 01 Jul 2025 07:06AM UTC coverage: 91.316% (-3.9%) from 95.248%

Build # 15992939927

Build Type

push

github

Committed by

web-flow

Commit Message

Bump diff-cover[toml] from 9.4.0 to 9.4.1 (#400)

Bumps [diff-cover[toml]](https://github.com/Bachmann1234/diff-cover)
from 9.4.0 to 9.4.1.
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a
href="https://github.com/Bachmann1234/diff_cover/blob/main/CHANGELOG">diff-cover[toml]'s
changelog</a>.</em></p>
<blockquote>
<p>06/23/2025 v9.4.1</p>
<ul>
<li>Update to LCOV to support pytest-cov. Pytest-cov uses an obsolute
lcov format that needed additional handling. <a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/505">PR
505</a> Thanks <a
href="https://github.com/129tyc"><code>@129tyc</code></a></li>
</ul>
<p>06/22/2025 v9.4.0</p>
<ul>
<li>
<p>Many improvements to internal structure, dependencies, and tests.
(Thanks <a
href="https://github.com/kingbuzzman"><code>@kingbuzzman</code></a>)</p>
<ul>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/491">PR
491</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/492">PR
492</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/494">PR
494</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/495">PR
495</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/496">PR
496</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/497">PR
497</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/498">PR
498</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/501">PR
501</a></li>
<li><a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/502">PR
502</a></li>
</ul>
</li>
<li>
<p>You can now use stdout and stderr when using --format <a
href="https://redirect.github.com/Bachmann1234/diff_cover/pull/500">PR
500</a> (Thanks <a
href="https://github.com/kingbuzzman"><code>@kingbuzzman</code></a>)</p>
</li>
<li>
<p>Significant enhancements to LCOV parsing! <a
href="https://redirect.github.com/Bac... (continued)

Run Details

1672 of 1831 relevant lines covered (91.32%)

0.91 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:
        """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	"""With HDBCLI, the user may reference to a local HDBUserStore key which holds
83	the connection details. SQLAlchemy-HANA should only pass the userkey name to
84	HDBCLI for the connection creation.
85	"""
86	_, result_kwargs = config.db.dialect.create_connect_args(	1✔
87	make_url("hana://userkey=myuserkeyname")
88	)
89	assert result_kwargs == {"userkey": "myuserkeyname", "vectoroutputtype": "list"}	1✔
90
91	def test_pass_uri_query_as_kwargs(self) -> None:	1✔
92	"""SQLAlchemy-HANA should passes all URL parameters to hdbcli."""
93	urls = [	1✔
94	"hana://username:secret-password@example.com/?encrypt=true&compress=true",
95	"hana://username:secret-password@example.com/TENANT_NAME?encrypt=true&compress=true",
96	]
97
98	for url in urls:	×
99	_, result_kwargs = config.db.dialect.create_connect_args(make_url(url))	×
100	assert result_kwargs["encrypt"] == "true"	1✔
101	assert result_kwargs["compress"] == "true"	1✔
102
103	def test_server_version_info(self) -> None:	1✔
104	# Test that the attribute is defined
105	assert config.db.dialect.server_version_info	1✔
106
107	def test_get_isolation_level(self) -> None:	1✔
108	eng = testing_engine(options={})	×
109	isolation_level = eng.dialect.get_isolation_level(eng.connect().connection)	1✔
110	eq_(isolation_level, DEFAULT_ISOLATION_LEVEL)	1✔
111
112	def test_set_isolation_level(self) -> None:	1✔
113	eng = testing_engine(options={})	×
114	conn = eng.connect()	1✔
115	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	1✔
116
117	eng.dialect.set_isolation_level(conn.connection, NON_DEFAULT_ISOLATION_LEVEL)	1✔
118
119	eq_(	1✔
120	eng.dialect.get_isolation_level(conn.connection),
121	NON_DEFAULT_ISOLATION_LEVEL,
122	)
123	conn.close()	1✔
124
125	def test_reset_level(self) -> None:	1✔
126	eng = testing_engine(options={})	×
127	conn = eng.connect()	1✔
128	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	1✔
129
130	eng.dialect.set_isolation_level(conn.connection, NON_DEFAULT_ISOLATION_LEVEL)	1✔
131	eq_(	1✔
132	eng.dialect.get_isolation_level(conn.connection),
133	NON_DEFAULT_ISOLATION_LEVEL,
134	)
135
136	eng.dialect.reset_isolation_level(conn.connection)	×
137	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	×
138	conn.close()	1✔
139
140	def test_set_level_with_setting(self) -> None:	1✔
141	eng = testing_engine(options={"isolation_level": NON_DEFAULT_ISOLATION_LEVEL})	×
142	conn = eng.connect()	1✔
143	eq_(	1✔
144	eng.dialect.get_isolation_level(conn.connection),
145	NON_DEFAULT_ISOLATION_LEVEL,
146	)
147
148	eng.dialect.set_isolation_level(conn.connection, DEFAULT_ISOLATION_LEVEL)	×
149	eq_(eng.dialect.get_isolation_level(conn.connection), DEFAULT_ISOLATION_LEVEL)	×
150	conn.close()	1✔
151
152	def test_invalid_level(self) -> None:	1✔
153	eng = testing_engine(options={"isolation_level": "FOO"})	×
154	levels = ", ".join(eng.dialect._isolation_lookup)	1✔
155	assert_raises_message(	1✔
156	ArgumentError,
157	"Invalid value 'FOO' for isolation_level. "
158	f"Valid isolation levels for {eng.dialect.name} are {levels}",
159	eng.connect,
160	)
161
162	def test_with_execution_options(self) -> None:	×
163	eng = create_engine(	1✔
164	config.db.url,
165	execution_options={"isolation_level": NON_DEFAULT_ISOLATION_LEVEL},
166	)
167	conn = eng.connect()	1✔
168	eq_(	1✔
169	eng.dialect.get_isolation_level(conn.connection),
170	NON_DEFAULT_ISOLATION_LEVEL,
171	)
172	conn.close()	1✔
173
174	def test_with_isolation_level_in_create_engine(self) -> None:	1✔
175	eng = create_engine(config.db.url, isolation_level=NON_DEFAULT_ISOLATION_LEVEL)	×
176	conn = eng.connect()	1✔
177	eq_(	1✔
178	eng.dialect.get_isolation_level(conn.connection),
179	NON_DEFAULT_ISOLATION_LEVEL,
180	)
181	conn.close()	1✔
182
183	def test_do_rollback_to_savepoint_no_error(self) -> None:	1✔
184	dialect = config.db.dialect	×
185	connection = Mock()	1✔
186
187	with mock.patch.object(	1✔
188	DefaultDialect, "do_rollback_to_savepoint"
189	) as super_rollback:
190	dialect.do_rollback_to_savepoint(connection, "savepoint")	1✔
191	super_rollback.assert_called_once_with(connection, "savepoint")	1✔
192
193	def test_do_rollback_to_savepoint_unrelated_error(self) -> None:	1✔
194	dialect = config.db.dialect	×
195	connection = Mock()	1✔
196
197	with (	1✔
198	mock.patch.object(
199	sys, "exc_info", return_value=(ValueError, ValueError(), Mock())
200	),
201	mock.patch.object(
202	DefaultDialect, "do_rollback_to_savepoint"
203	) as super_rollback,
204	):
205	dialect.do_rollback_to_savepoint(connection, "savepoint")	1✔
206	super_rollback.assert_called_once_with(connection, "savepoint")	×
207
208	def test_do_rollback_to_savepoint_ignores_error(self) -> None:	1✔
209	dialect = config.db.dialect	×
210	connection = Mock()	1✔
211
212	error = Error(133, "transaction rolled back: deadlock")	1✔
213	dbapi_error = DBAPIError(None, None, error)	×
214
215	with (	1✔
216	mock.patch.object(
217	sys, "exc_info", return_value=(DBAPIError, dbapi_error, Mock())
218	),
219	mock.patch.object(
220	DefaultDialect, "do_rollback_to_savepoint"
221	) as super_rollback,
222	):
223	dialect.do_rollback_to_savepoint(connection, "savepoint")	1✔
224	super_rollback.assert_not_called()	×
225
226	def test_vectoroutputtype_is_blocked(self) -> None:	1✔
227	url = "hana://username:secret-password@example.com/?encrypt=true&vectoroutputtype=list"	×
228	with pytest.raises(ValueError, match="vectoroutputtype"):	1✔
229	config.db.dialect.create_connect_args(make_url(url))	1✔
230
231	@pytest.mark.parametrize(	1✔
232	"kwargs,vector_output_type",
233	[
234	({}, "list"),
235	({"vector_output_type": "list"}, "list"),
236	({"vector_output_type": "memoryview"}, "memoryview"),
237	],
238	)
239	def test_vector_output_type(self, kwargs: dict, vector_output_type: str) -> None:	×
240	engine = create_engine("hana://username:secret-password@example.com", **kwargs)	×
241	assert engine.dialect.vector_output_type == vector_output_type	1✔

SAP / sqlalchemy-hana / 15992939927

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