← Back to branch summary

~awuerl/blitzortung-python/master

« back to all changes in this revision

Viewing changes to tests/db/test_db_query.py

  • Committer: Andreas Wuerl
  • Date: 2019-06-24 19:32:25 UTC
  • mto: This revision was merged to the branch mainline in revision 394.
  • Revision ID: git-v1:29d4be282c05035bdff8d4828573537ed6d8ceea
update tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/db/test_db_query.py
18
18
 
19
19
"""
20
20
 
 
21
import datetime
21
22
from unittest import TestCase
22
 
import datetime
23
 
from hamcrest import assert_that, is_, equal_to
24
 
from nose.tools import raises
 
23
 
 
24
import pytest
25
25
import shapely.wkb
 
26
from assertpy import assert_that
26
27
 
27
28
import blitzortung
28
29
import blitzortung.db.query
50
51
        self.assertEqual(id_interval.end, 5678)
51
52
        self.assertEqual(str(id_interval), "[1234 : 5678]")
52
53
 
53
 
    @raises(ValueError)
54
54
    def test_exception_when_start_is_not_integer(self):
55
 
        blitzortung.db.query.IdInterval("asdf")
 
55
        with pytest.raises(ValueError):
 
56
            blitzortung.db.query.IdInterval("asdf")
56
57
 
57
 
    @raises(ValueError)
58
58
    def test_exception_when_end_is_not_integer(self):
59
 
        blitzortung.db.query.IdInterval(1, "asdf")
 
59
        with pytest.raises(ValueError):
 
60
            blitzortung.db.query.IdInterval(1, "asdf")
60
61
 
61
62
 
62
63
class TimeIntervalTest(TestCase):
88
89
 
89
90
        self.assertEqual(id_interval.duration, datetime.timedelta(minutes=10))
90
91
 
91
 
    @raises(ValueError)
92
92
    def test_exception_when_start_is_not_integer(self):
93
 
        blitzortung.db.query.TimeInterval("asdf")
 
93
        with pytest.raises(ValueError):
 
94
            blitzortung.db.query.TimeInterval("asdf")
94
95
 
95
 
    @raises(ValueError)
96
96
    def test_exception_when_end_is_not_integer(self):
97
 
        blitzortung.db.query.TimeInterval(datetime.datetime.utcnow(), "asdf")
 
97
        with pytest.raises(ValueError):
 
98
            blitzortung.db.query.TimeInterval(datetime.datetime.utcnow(), "asdf")
98
99
 
99
100
 
100
101
class QueryTest(TestCase):
130
131
 
131
132
    def test_add_condition_with_parameters(self):
132
133
        self.query.add_condition("column LIKE %(name)s", name='<name>')
133
 
        assert_that(str(self.query), is_(equal_to("WHERE column LIKE %(name)s")))
134
 
        assert_that(self.query.get_parameters(), is_(equal_to({'name': '<name>'})))
 
134
        assert_that(str(self.query)).is_equal_to("WHERE column LIKE %(name)s")
 
135
        assert_that(self.query.get_parameters()).is_equal_to({'name': '<name>'})
135
136
 
136
137
        self.query.add_condition("other LIKE %(type)s", type='<type>')
137
 
        assert_that(str(self.query), is_(equal_to("WHERE column LIKE %(name)s AND other LIKE %(type)s")))
138
 
        assert_that(self.query.get_parameters(), is_(equal_to({'name': '<name>', 'type': '<type>'})))
 
138
        assert_that(str(self.query)).is_equal_to("WHERE column LIKE %(name)s AND other LIKE %(type)s")
 
139
        assert_that(self.query.get_parameters()).is_equal_to({'name': '<name>', 'type': '<type>'})
139
140
 
140
141
    def test_add_order(self):
141
142
        self.query.set_order(blitzortung.db.query.Order("bar"))
156
157
    def test_parse_args_with_id_interval(self):
157
158
        self.query.set_default_conditions(id_interval=blitzortung.db.query.IdInterval(10, 15))
158
159
 
159
 
        assert_that(str(self.query), is_(equal_to("WHERE id >= %(start_id)s AND id < %(end_id)s")))
160
 
        assert_that(self.query.get_parameters(), is_(equal_to({'start_id': 10, 'end_id': 15})))
 
160
        assert_that(str(self.query)).is_equal_to("WHERE id >= %(start_id)s AND id < %(end_id)s")
 
161
        assert_that(self.query.get_parameters()).is_equal_to({'start_id': 10, 'end_id': 15})
161
162
 
162
163
    def test_parse_args_with_time_interval(self):
163
164
        self.query.set_default_conditions(time_interval=blitzortung.db.query.TimeInterval(
164
165
            datetime.datetime(2013, 10, 9, 17, 20),
165
166
            datetime.datetime(2013, 10, 11, 6, 30)))
166
167
 
167
 
        assert_that(str(self.query), is_(equal_to(
 
168
        assert_that(str(self.query)).is_equal_to(
168
169
            "WHERE \"timestamp\" >= %(start_time)s AND \"timestamp\" < %(end_time)s"
169
 
        )))
170
 
        assert_that(self.query.get_parameters(), is_(equal_to({
 
170
        )
 
171
        assert_that(self.query.get_parameters()).is_equal_to({
171
172
            'start_time': datetime.datetime(2013, 10, 9, 17, 20),
172
 
            'end_time': datetime.datetime(2013, 10, 11, 6, 30)})))
 
173
            'end_time': datetime.datetime(2013, 10, 11, 6, 30)})
173
174
 
174
175
    def test_parse_args_with_order(self):
175
176
        self.query.set_default_conditions(order='test')
176
177
 
177
 
        assert_that(str(self.query), is_(equal_to(
 
178
        assert_that(str(self.query)).is_equal_to(
178
179
            "ORDER BY test"
179
 
        )))
 
180
        )
180
181
 
181
182
    def test_parse_args_with_limit(self):
182
183
        self.query.set_default_conditions(limit=10)
183
184
 
184
 
        assert_that(str(self.query), is_(equal_to(
 
185
        assert_that(str(self.query)).is_equal_to(
185
186
            "LIMIT 10"
186
 
        )))
 
187
        )
187
188
 
188
189
 
189
190
class SelectQueryTest(TestCase):
215
216
        self.assertEqual(str(self.query), "SELECT FROM foo WHERE qux AND quux")
216
217
 
217
218
    def test_add_parameters(self):
218
 
        assert_that(self.query.get_parameters(), is_(equal_to({})))
 
219
        assert_that(self.query.get_parameters()).is_equal_to({})
219
220
 
220
221
        self.query.add_parameters(foo='bar', baz='qux')
221
222
 
222
 
        assert_that(self.query.get_parameters(), is_(equal_to({'foo': 'bar', 'baz': 'qux'})))
 
223
        assert_that(self.query.get_parameters()).is_equal_to({'foo': 'bar', 'baz': 'qux'})
223
224
 
224
225
 
225
226
class GridQueryTest(TestCase):
229
230
        query = blitzortung.db.query.GridQuery(raster)
230
231
        query.set_table_name('strikes')
231
232
 
232
 
        assert_that(str(query), is_(equal_to(
 
233
        assert_that(str(query)).is_equal_to(
233
234
            'SELECT '
234
235
            'TRUNC((ST_X(ST_Transform(geog::geometry, %(srid)s)) - %(xmin)s) / %(xdiv)s)::integer AS rx, '
235
236
            'TRUNC((ST_Y(ST_Transform(geog::geometry, %(srid)s)) - %(ymin)s) / %(ydiv)s)::integer AS ry, '
237
238
            'max("timestamp") as "timestamp" '
238
239
            'FROM strikes '
239
240
            'WHERE ST_GeomFromWKB(%(envelope)s, %(envelope_srid)s) && geog '
240
 
            'GROUP BY rx, ry')))
 
241
            'GROUP BY rx, ry')
241
242
 
242
243
        parameters = query.get_parameters()
243
 
        assert_that(parameters['xmin'], is_(equal_to(-10)))
244
 
        assert_that(parameters['xdiv'], is_(equal_to(1.5)))
245
 
        assert_that(parameters['ymin'], is_(equal_to(15)))
246
 
        assert_that(parameters['ydiv'], is_(equal_to(1)))
247
 
        assert_that(parameters['srid'], is_(equal_to(4326)))
248
 
        assert_that(parameters['envelope_srid'], is_(equal_to(4326)))
 
244
        assert_that(parameters['xmin']).is_equal_to(-10)
 
245
        assert_that(parameters['xdiv']).is_equal_to(1.5)
 
246
        assert_that(parameters['ymin']).is_equal_to(15)
 
247
        assert_that(parameters['ydiv']).is_equal_to(1)
 
248
        assert_that(parameters['srid']).is_equal_to(4326)
 
249
        assert_that(parameters['envelope_srid']).is_equal_to(4326)
249
250
        envelope = shapely.wkb.loads(parameters['envelope'].adapted)
250
 
        assert_that(envelope.bounds, is_(equal_to((-10, 15, 20, 35))))
251
 
 
 
251
        assert_that(envelope.bounds).is_equal_to((-10, 15, 20, 35))