Coverage for cc_modules/tests/cc_sqla_coltypes_tests.py : 21%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
3"""
4camcops_server/cc_modules/tests/cc_sqla_coltypes_tests.py
6===============================================================================
8 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com).
10 This file is part of CamCOPS.
12 CamCOPS is free software: you can redistribute it and/or modify
13 it under the terms of the GNU General Public License as published by
14 the Free Software Foundation, either version 3 of the License, or
15 (at your option) any later version.
17 CamCOPS is distributed in the hope that it will be useful,
18 but WITHOUT ANY WARRANTY; without even the implied warranty of
19 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20 GNU General Public License for more details.
22 You should have received a copy of the GNU General Public License
23 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
25===============================================================================
26"""
28import datetime
29from typing import Union
30import unittest
32from cardinal_pythonlib.datetimefunc import coerce_to_pendulum
33from cardinal_pythonlib.sqlalchemy.session import SQLITE_MEMORY_URL
34from pendulum import DateTime as Pendulum, Duration
35from semantic_version import Version
36from sqlalchemy.sql.functions import func
37from sqlalchemy.sql.schema import Column
38from sqlalchemy.sql.sqltypes import (
39 DateTime,
40 Integer,
41)
43from camcops_server.cc_modules.cc_sqla_coltypes import (
44 isotzdatetime_to_utcdatetime,
45 PendulumDateTimeAsIsoTextColType,
46 PendulumDurationAsIsoTextColType,
47 SemanticVersionColType,
48 unknown_field_to_utcdatetime,
49)
52# =============================================================================
53# Unit testing
54# =============================================================================
56class SqlaColtypesTest(unittest.TestCase):
57 """
58 Unit tests.
59 """
60 # don't inherit from ExtendedTestCase; circular import
62 @staticmethod
63 def _assert_dt_equal(a: Union[datetime.datetime, Pendulum],
64 b: Union[datetime.datetime, Pendulum]) -> None:
65 # Accept that one may have been truncated or rounded to milliseconds.
66 a = coerce_to_pendulum(a)
67 b = coerce_to_pendulum(b)
68 diff = a - b
69 assert diff.microseconds < 1000, f"{a!r} != {b!r}"
71 def test_iso_datetime_field(self) -> None:
72 import pendulum
73 from sqlalchemy.engine import create_engine
74 from sqlalchemy.sql.expression import select
75 from sqlalchemy.sql.schema import MetaData, Table
77 engine = create_engine(SQLITE_MEMORY_URL, echo=True)
78 meta = MetaData()
79 meta.bind = engine # adds execute() method to select() etc.
80 # ... http://docs.sqlalchemy.org/en/latest/core/connections.html
82 id_colname = 'id'
83 dt_local_colname = 'dt_local'
84 dt_utc_colname = 'dt_utc'
85 iso_colname = 'iso'
86 id_col = Column(id_colname, Integer, primary_key=True)
87 dt_local_col = Column(dt_local_colname, DateTime)
88 dt_utc_col = Column(dt_utc_colname, DateTime)
89 iso_col = Column(iso_colname, PendulumDateTimeAsIsoTextColType)
91 table = Table('testtable', meta,
92 id_col, dt_local_col, dt_utc_col, iso_col)
93 table.create()
95 now = Pendulum.now()
96 now_utc = now.in_tz(pendulum.UTC)
97 yesterday = now.subtract(days=1)
98 yesterday_utc = yesterday.in_tz(pendulum.UTC)
100 table.insert().values([
101 {
102 id_colname: 1,
103 dt_local_colname: now,
104 dt_utc_colname: now_utc,
105 iso_colname: now,
106 },
107 {
108 id_colname: 2,
109 dt_local_colname: yesterday,
110 dt_utc_colname: yesterday_utc,
111 iso_colname: yesterday
112 },
113 ]).execute()
114 select_fields = [
115 id_col,
116 dt_local_col,
117 dt_utc_col,
118 iso_col,
119 func.length(dt_local_col).label("len_dt_local_col"),
120 func.length(dt_utc_col).label("len_dt_utc_col"),
121 func.length(iso_col).label("len_iso_col"),
122 isotzdatetime_to_utcdatetime(iso_col).label("iso_to_utcdt"),
123 unknown_field_to_utcdatetime(dt_utc_col).label("uk_utcdt_to_utcdt"),
124 unknown_field_to_utcdatetime(iso_col).label("uk_iso_to_utc_dt"),
125 ]
126 rows = list(
127 select(select_fields)
128 .select_from(table)
129 .order_by(id_col)
130 .execute()
131 )
132 self._assert_dt_equal(rows[0][dt_local_col], now)
133 self._assert_dt_equal(rows[0][dt_utc_col], now_utc)
134 self._assert_dt_equal(rows[0][iso_colname], now)
135 self._assert_dt_equal(rows[0]["iso_to_utcdt"], now_utc)
136 self._assert_dt_equal(rows[0]["uk_utcdt_to_utcdt"], now_utc)
137 self._assert_dt_equal(rows[0]["uk_iso_to_utc_dt"], now_utc)
138 self._assert_dt_equal(rows[1][dt_local_col], yesterday)
139 self._assert_dt_equal(rows[1][dt_utc_col], yesterday_utc)
140 self._assert_dt_equal(rows[1][iso_colname], yesterday)
141 self._assert_dt_equal(rows[1]["iso_to_utcdt"], yesterday_utc)
142 self._assert_dt_equal(rows[1]["uk_utcdt_to_utcdt"], yesterday_utc)
143 self._assert_dt_equal(rows[1]["uk_iso_to_utc_dt"], yesterday_utc)
145 @staticmethod
146 def _assert_duration_equal(a: Duration, b: Duration) -> None:
147 assert a == b, f"{a!r} != {b!r}"
149 def test_iso_duration_field(self) -> None:
150 from sqlalchemy.engine import create_engine
151 from sqlalchemy.sql.expression import select
152 from sqlalchemy.sql.schema import MetaData, Table
154 # As above:
155 engine = create_engine(SQLITE_MEMORY_URL, echo=True)
156 meta = MetaData()
157 meta.bind = engine
159 id_colname = 'id'
160 duration_colname = 'duration_iso'
161 id_col = Column(id_colname, Integer, primary_key=True)
162 duration_col = Column(duration_colname,
163 PendulumDurationAsIsoTextColType)
165 table = Table('testtable', meta,
166 id_col, duration_col)
167 table.create()
169 d1 = Duration(years=1, months=3, seconds=3, microseconds=4)
170 d2 = Duration(seconds=987.654321)
171 d3 = Duration(days=-5)
173 table.insert().values([
174 {
175 id_colname: 1,
176 duration_colname: d1,
177 },
178 {
179 id_colname: 2,
180 duration_colname: d2,
181 },
182 {
183 id_colname: 3,
184 duration_colname: d3,
185 },
186 ]).execute()
187 select_fields = [
188 id_col,
189 duration_col,
190 ]
191 rows = list(
192 select(select_fields)
193 .select_from(table)
194 .order_by(id_col)
195 .execute()
196 )
197 self._assert_duration_equal(rows[0][duration_col], d1)
198 self._assert_duration_equal(rows[1][duration_col], d2)
199 self._assert_duration_equal(rows[2][duration_col], d3)
201 @staticmethod
202 def _assert_version_equal(a: Version, b: Version) -> None:
203 assert a == b, f"{a!r} != {b!r}"
205 def test_semantic_version_field(self) -> None:
206 from sqlalchemy.engine import create_engine
207 from sqlalchemy.sql.expression import select
208 from sqlalchemy.sql.schema import MetaData, Table
210 # As above:
211 engine = create_engine(SQLITE_MEMORY_URL, echo=True)
212 meta = MetaData()
213 meta.bind = engine
215 id_colname = 'id'
216 version_colname = 'version'
217 id_col = Column(id_colname, Integer, primary_key=True)
218 version_col = Column(version_colname, SemanticVersionColType)
220 table = Table('testtable', meta,
221 id_col, version_col)
222 table.create()
224 v1 = Version("1.1.0")
225 v2 = Version("2.0.1")
226 v3 = Version("14.0.0")
228 table.insert().values([
229 {
230 id_colname: 1,
231 version_colname: v1,
232 },
233 {
234 id_colname: 2,
235 version_colname: v2,
236 },
237 {
238 id_colname: 3,
239 version_colname: v3,
240 },
241 ]).execute()
242 select_fields = [
243 id_col,
244 version_col,
245 ]
246 rows = list(
247 select(select_fields)
248 .select_from(table)
249 .order_by(id_col)
250 .execute()
251 )
252 self._assert_version_equal(rows[0][version_col], v1)
253 self._assert_version_equal(rows[1][version_col], v2)
254 self._assert_version_equal(rows[2][version_col], v3)