Hide keyboard shortcuts

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 

2 

3""" 

4camcops_server/cc_modules/tests/cc_sqla_coltypes_tests.py 

5 

6=============================================================================== 

7 

8 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com). 

9 

10 This file is part of CamCOPS. 

11 

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. 

16 

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. 

21 

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/>. 

24 

25=============================================================================== 

26""" 

27 

28import datetime 

29from typing import Union 

30import unittest 

31 

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) 

42 

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) 

50 

51 

52# ============================================================================= 

53# Unit testing 

54# ============================================================================= 

55 

56class SqlaColtypesTest(unittest.TestCase): 

57 """ 

58 Unit tests. 

59 """ 

60 # don't inherit from ExtendedTestCase; circular import 

61 

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}" 

70 

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 

76 

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 

81 

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) 

90 

91 table = Table('testtable', meta, 

92 id_col, dt_local_col, dt_utc_col, iso_col) 

93 table.create() 

94 

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) 

99 

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) 

144 

145 @staticmethod 

146 def _assert_duration_equal(a: Duration, b: Duration) -> None: 

147 assert a == b, f"{a!r} != {b!r}" 

148 

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 

153 

154 # As above: 

155 engine = create_engine(SQLITE_MEMORY_URL, echo=True) 

156 meta = MetaData() 

157 meta.bind = engine 

158 

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) 

164 

165 table = Table('testtable', meta, 

166 id_col, duration_col) 

167 table.create() 

168 

169 d1 = Duration(years=1, months=3, seconds=3, microseconds=4) 

170 d2 = Duration(seconds=987.654321) 

171 d3 = Duration(days=-5) 

172 

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) 

200 

201 @staticmethod 

202 def _assert_version_equal(a: Version, b: Version) -> None: 

203 assert a == b, f"{a!r} != {b!r}" 

204 

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 

209 

210 # As above: 

211 engine = create_engine(SQLITE_MEMORY_URL, echo=True) 

212 meta = MetaData() 

213 meta.bind = engine 

214 

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) 

219 

220 table = Table('testtable', meta, 

221 id_col, version_col) 

222 table.create() 

223 

224 v1 = Version("1.1.0") 

225 v2 = Version("2.0.1") 

226 v3 = Version("14.0.0") 

227 

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)