|
1 | 1 | import configparser |
| 2 | +import datetime |
2 | 3 | import os |
3 | 4 | import typing |
4 | 5 | from math import isclose |
@@ -136,3 +137,26 @@ def test_redshift_varbyte_insert_python_bytes(db_kwargs, _input, client_protocol |
136 | 137 | assert len(results) == 1 |
137 | 138 | assert len(results[0]) == 2 |
138 | 139 | assert results[0][1] == data.hex() |
| 140 | + |
| 141 | + |
| 142 | +@pytest.mark.parametrize("client_protocol", ClientProtocolVersion.list()) |
| 143 | +@pytest.mark.parametrize("_input", ["Australia/Eucla", "America/New_York"]) |
| 144 | +def test_redshift_timetz_client_tz_applied(db_kwargs, _input, client_protocol): |
| 145 | + db_kwargs["client_protocol_version"] = client_protocol |
| 146 | + data = _input |
| 147 | + t_time = datetime.time(hour=10, minute=14, second=56, tzinfo=datetime.timezone.utc) |
| 148 | + |
| 149 | + with redshift_connector.connect(**db_kwargs) as con: |
| 150 | + with con.cursor() as cursor: |
| 151 | + cursor.execute("create table t_timetz_tz (v1 timetz)") |
| 152 | + cursor.execute("insert into t_timetz_tz (v1) values(%s)", (t_time,)) |
| 153 | + cursor.execute("select v1 from t_timetz_tz ") |
| 154 | + results: typing.Tuple = cursor.fetchone() |
| 155 | + assert len(results) == 1 |
| 156 | + assert results[0] == t_time |
| 157 | + |
| 158 | + cursor.execute("set timezone = '{}'".format(data)) |
| 159 | + cursor.execute("select v1 from t_timetz_tz ") |
| 160 | + results: typing.Tuple = cursor.fetchone() |
| 161 | + assert len(results) == 1 |
| 162 | + assert results[0] == t_time |
0 commit comments