Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion tests/db_engine_specs/base_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,5 +31,6 @@ def sql_limit_regex(
limited = engine_spec_class.apply_limit_to_sql(sql, limit, main)
self.assertEqual(expected_sql, limited)

def get_dttm(self):
@classmethod
def get_dttm(cls):
return datetime.strptime("2019-01-02 03:04:05.678900", "%Y-%m-%d %H:%M:%S.%f")
53 changes: 35 additions & 18 deletions tests/db_engine_specs/oracle_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
from unittest import mock

import pytest
from sqlalchemy import column
from sqlalchemy.dialects import oracle
from sqlalchemy.dialects.oracle import DATE, NVARCHAR, VARCHAR
Expand All @@ -37,24 +40,6 @@ def test_oracle_time_expression_reserved_keyword_1m_grain(self):
self.assertEqual(result, "TRUNC(CAST(\"decimal\" as DATE), 'MONTH')")
dttm = self.get_dttm()

def test_convert_dttm(self):
dttm = self.get_dttm()

test_cases = (
(
OracleEngineSpec.convert_dttm("DATE", dttm),
"TO_DATE('2019-01-02', 'YYYY-MM-DD')",
),
(
OracleEngineSpec.convert_dttm("DATETIME", dttm),
"""TO_DATE('2019-01-02T03:04:05', 'YYYY-MM-DD"T"HH24:MI:SS')""",
),
(
OracleEngineSpec.convert_dttm("TIMESTAMP", dttm),
"""TO_TIMESTAMP('2019-01-02T03:04:05.678900', 'YYYY-MM-DD"T"HH24:MI:SS.ff6')""",
),
)

def test_column_datatype_to_string(self):
test_cases = (
(DATE(), "DATE"),
Expand All @@ -68,3 +53,35 @@ def test_column_datatype_to_string(self):
original, oracle.dialect()
)
self.assertEqual(actual, expected)

def test_fetch_data_no_description(self):
cursor = mock.MagicMock()
cursor.description = []
assert OracleEngineSpec.fetch_data(cursor) == []

def test_fetch_data(self):
cursor = mock.MagicMock()
result = ["a", "b"]
cursor.fetchall.return_value = result
assert OracleEngineSpec.fetch_data(cursor) == result


@pytest.mark.parametrize(
"date_format,expected",
[
("DATE", "TO_DATE('2019-01-02', 'YYYY-MM-DD')"),
("DATETIME", """TO_DATE('2019-01-02T03:04:05', 'YYYY-MM-DD"T"HH24:MI:SS')"""),
(
"TIMESTAMP",
"""TO_TIMESTAMP('2019-01-02T03:04:05.678900', 'YYYY-MM-DD"T"HH24:MI:SS.ff6')""",
),
(
"timestamp",
"""TO_TIMESTAMP('2019-01-02T03:04:05.678900', 'YYYY-MM-DD"T"HH24:MI:SS.ff6')""",
),
("Other", None),
],
)
def test_convert_dttm(date_format, expected):
dttm = TestOracleDbEngineSpec.get_dttm()
assert OracleEngineSpec.convert_dttm(date_format, dttm) == expected