Test Failed
Push — master ( 0b6ef6...7a9489 )
by Dominik
10:34
created

OSMAlchemyModelTestsMySQL.tearDown()   A

Complexity

Conditions 1

Size

Total Lines 5

Duplication

Lines 5
Ratio 100 %

Importance

Changes 1
Bugs 0 Features 0
Metric Value
cc 1
c 1
b 0
f 0
dl 5
loc 5
rs 9.4285
1
#!/usr/bin/env python3
2
# ~*~ coding: utf-8 ~*~
3
#-
4
# OSMAlchemy - OpenStreetMap to SQLAlchemy bridge
5
# Copyright (c) 2016 Dominik George <[email protected]>
6
#
7
# Permission is hereby granted, free of charge, to any person obtaining a copy
8
# of this software and associated documentation files (the "Software"), to deal
9
# in the Software without restriction, including without limitation the rights
10
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11
# copies of the Software, and to permit persons to whom the Software is
12
# furnished to do so, subject to the following conditions:
13
#
14
# The above copyright notice and this permission notice shall be included in all
15
# copies or substantial portions of the Software.
16
#
17
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23
# SOFTWARE.
24
#
25
# Alternatively, you are free to use OSMAlchemy under Simplified BSD, The
26
# MirOS Licence, GPL-2+, LGPL-2.1+, AGPL-3+ or the same terms as Python
27
# itself.
28
29
""" Tests concerning the OSM data model implementation. """
30
31
# Standard unit testing framework
32
import unittest
33
34
# We want to profile test cases
35
import time
36
37
# Helper libraries for different database engines
38
from testing.mysqld import MysqldFactory
39
from testing.postgresql import PostgresqlFactory
40
41
# Module to be tested
42
from osmalchemy import OSMAlchemy
43
44
# SQLAlchemy for working with model and data
45
from sqlalchemy import create_engine
46
from sqlalchemy.ext.declarative import declarative_base
47
from sqlalchemy.orm import sessionmaker, scoped_session
48
49
# Flask for testing integration with Flask
50
from flask import Flask
51
from flask_sqlalchemy import SQLAlchemy as FlaskSQLAlchemy
52
53
# Create database engine factories to enable caching
54
Postgresql = PostgresqlFactory(cache_initialized_db=True)
55
Mysqld = MysqldFactory(cache_initialized_db=True)
56
57
# Dictionary to store profiling information about tests
58
profile = {}
59
60
def tearDownModule():
61
    """ Global test suite tear down code """
62
63
    # Purge caches of database engines
64
    Postgresql.clear_cache()
65
    Mysqld.clear_cache()
66
67
    # Print profiling info
68
    print("Database model test times")
69
    print("=========================")
70
    print("")
71
    for suite in profile:
72
        print("%8.3f s\t%s" % (sum([profile[suite][test] for test in profile[suite]]), suite))
73
        for test in profile[suite]:
74
            print("\t%8.3f s\t%s" % (profile[suite][test], test))
75
76
class OSMAlchemyModelTests(object):
77
    """ Incomplete base class for common test routines.
78
79
    Subclassed in engine-dependent test classes.
80
    """
81
82
    def setUp(self):
83
        if not self.__class__.__name__ in profile:
84
            profile[self.__class__.__name__] = {}
85
        profile[self.__class__.__name__][self.id().split(".")[-1]] = time.time()
86
87
    def tearDown(self):
88
        profile[self.__class__.__name__][self.id().split(".")[-1]] -= time.time()
89
        profile[self.__class__.__name__][self.id().split(".")[-1]] *= -1
90
91
    def test_create_node(self):
92
        # Create node
93
        node = self.osmalchemy.node()
94
        node.latitude = 51.0
95
        node.longitude = 7.0
96
97
        # Store node
98
        self.session.add(node)
99
        self.session.commit()
100
        # Ensure removal from ORM
101
        self.session.remove()
102
103
        # Query for node and check
104
        node = self.session.query(self.osmalchemy.node).filter_by(latitude=51.0).first()
105
        self.assertEqual(node.latitude, 51.0)
106
        self.assertEqual(node.longitude, 7.0)
107
        self.assertEqual(len(node.tags), 0)
108
109
    def test_create_node_with_tags(self):
110
        # Create node and tags
111
        node = self.osmalchemy.node(51.0, 7.0)
112
        node.tags[u"name"] = u"test"
113
        node.tags[u"foo"] = u"bar"
114
115
        # Store everything
116
        self.session.add(node)
117
        self.session.commit()
118
        # Ensure removal from ORM
119
        self.session.remove()
120
121
        # Query for node and check
122
        node = self.session.query(self.osmalchemy.node).filter_by(latitude=51.0).first()
123
        self.assertEqual(node.latitude, 51.0)
124
        self.assertEqual(node.longitude, 7.0)
125
        self.assertEqual(len(node.tags), 2)
126
        self.assertEqual(node.tags[u"name"], u"test")
127
        self.assertEqual(node.tags[u"foo"], u"bar")
128
129
    def test_create_way_with_nodes(self):
130
        # Create way and nodes
131
        way = self.osmalchemy.way()
132
        way.nodes = [self.osmalchemy.node(51.0, 7.0),
133
                     self.osmalchemy.node(51.1, 7.1),
134
                     self.osmalchemy.node(51.2, 7.2),
135
                     self.osmalchemy.node(51.3, 7.3),
136
                     self.osmalchemy.node(51.4, 7.4)]
137
138
        # Store everything
139
        self.session.add(way)
140
        self.session.commit()
141
        # Ensure removal from ORM
142
        self.session.remove()
143
144
        # Query for way and check
145
        way = self.session.query(self.osmalchemy.way).first()
146
        self.assertEqual(len(way.nodes), 5)
147
        self.assertEqual((way.nodes[0].latitude, way.nodes[0].longitude), (51.0, 7.0))
148
        self.assertEqual((way.nodes[1].latitude, way.nodes[1].longitude), (51.1, 7.1))
149
        self.assertEqual((way.nodes[2].latitude, way.nodes[2].longitude), (51.2, 7.2))
150
        self.assertEqual((way.nodes[3].latitude, way.nodes[3].longitude), (51.3, 7.3))
151
        self.assertEqual((way.nodes[4].latitude, way.nodes[4].longitude), (51.4, 7.4))
152
153
    def test_create_way_with_nodes_and_tags(self):
154
        # Create way and nodes
155
        way = self.osmalchemy.way()
156
        way.nodes = [self.osmalchemy.node(51.0, 7.0),
157
                     self.osmalchemy.node(51.1, 7.1),
158
                     self.osmalchemy.node(51.2, 7.2),
159
                     self.osmalchemy.node(51.3, 7.3),
160
                     self.osmalchemy.node(51.4, 7.4)]
161
        way.tags[u"name"] = u"Testway"
162
        way.tags[u"foo"] = u"bar"
163
164
        # Store everything
165
        self.session.add(way)
166
        self.session.commit()
167
        # Ensure removal from ORM
168
        self.session.remove()
169
170
        # Query for way and check
171
        way = self.session.query(self.osmalchemy.way).first()
172
        self.assertEqual(len(way.nodes), 5)
173
        self.assertEqual((way.nodes[0].latitude, way.nodes[0].longitude), (51.0, 7.0))
174
        self.assertEqual((way.nodes[1].latitude, way.nodes[1].longitude), (51.1, 7.1))
175
        self.assertEqual((way.nodes[2].latitude, way.nodes[2].longitude), (51.2, 7.2))
176
        self.assertEqual((way.nodes[3].latitude, way.nodes[3].longitude), (51.3, 7.3))
177
        self.assertEqual((way.nodes[4].latitude, way.nodes[4].longitude), (51.4, 7.4))
178
        self.assertEqual(len(way.tags), 2)
179
        self.assertEqual(way.tags[u"name"], u"Testway")
180
        self.assertEqual(way.tags[u"foo"], u"bar")
181
182
    def test_create_way_with_nodes_and_tags_and_tags_on_node(self):
183
        # Create way and nodes
184
        way = self.osmalchemy.way()
185
        way.nodes = [self.osmalchemy.node(51.0, 7.0),
186
                     self.osmalchemy.node(51.1, 7.1),
187
                     self.osmalchemy.node(51.2, 7.2),
188
                     self.osmalchemy.node(51.3, 7.3),
189
                     self.osmalchemy.node(51.4, 7.4)]
190
        way.tags[u"name"] = u"Testway"
191
        way.tags[u"foo"] = u"bar"
192
        way.nodes[2].tags[u"name"] = u"Testampel"
193
        way.nodes[2].tags[u"foo"] = u"bar"
194
195
        # Store everything
196
        self.session.add(way)
197
        self.session.commit()
198
        # Ensure removal from ORM
199
        self.session.remove()
200
201
        # Query for way and check
202
        way = self.session.query(self.osmalchemy.way).first()
203
        self.assertEqual(len(way.nodes), 5)
204
        self.assertEqual((way.nodes[0].latitude, way.nodes[0].longitude), (51.0, 7.0))
205
        self.assertEqual((way.nodes[1].latitude, way.nodes[1].longitude), (51.1, 7.1))
206
        self.assertEqual((way.nodes[2].latitude, way.nodes[2].longitude), (51.2, 7.2))
207
        self.assertEqual((way.nodes[3].latitude, way.nodes[3].longitude), (51.3, 7.3))
208
        self.assertEqual((way.nodes[4].latitude, way.nodes[4].longitude), (51.4, 7.4))
209
        self.assertEqual(len(way.tags), 2)
210
        self.assertEqual(way.tags[u"name"], u"Testway")
211
        self.assertEqual(way.tags[u"foo"], u"bar")
212
        self.assertEqual(len(way.nodes[2].tags), 2)
213
        self.assertEqual(way.nodes[2].tags[u"name"], u"Testampel")
214
        self.assertEqual(way.nodes[2].tags[u"foo"], u"bar")
215
216
    def test_create_relation_with_nodes(self):
217
        # Create way and add nodes
218
        relation = self.osmalchemy.relation()
219
        relation.members = [(self.osmalchemy.node(51.0, 7.0), u""),
220
                            (self.osmalchemy.node(51.1, 7.1), u""),
221
                            (self.osmalchemy.node(51.2, 7.2), u""),
222
                            (self.osmalchemy.node(51.3, 7.3), u""),
223
                            (self.osmalchemy.node(51.4, 7.4), u"")]
224
225
        # Store everything
226
        self.session.add(relation)
227
        self.session.commit()
228
        # Ensure removal from ORM
229
        self.session.remove()
230
231
        # Query for way and check
232
        relation = self.session.query(self.osmalchemy.relation).first()
233
        self.assertEqual((relation.members[0][0].latitude, relation.members[0][0].longitude),
234
                         (51.0, 7.0))
235
        self.assertEqual((relation.members[1][0].latitude, relation.members[1][0].longitude),
236
                         (51.1, 7.1))
237
        self.assertEqual((relation.members[2][0].latitude, relation.members[2][0].longitude),
238
                         (51.2, 7.2))
239
        self.assertEqual((relation.members[3][0].latitude, relation.members[3][0].longitude),
240
                         (51.3, 7.3))
241
        self.assertEqual((relation.members[4][0].latitude, relation.members[4][0].longitude),
242
                         (51.4, 7.4))
243
244
    def test_create_relation_with_nodes_and_ways(self):
245
        # Create way and add nodes and ways
246
        relation = self.osmalchemy.relation()
247
        relation.members = [(self.osmalchemy.node(51.0, 7.0), u''),
248
                            (self.osmalchemy.way(), u''),
249
                            (self.osmalchemy.node(51.1, 7.1), u''),
250
                            (self.osmalchemy.way(), u''),
251
                            (self.osmalchemy.node(51.2, 7.2), u''),
252
                            (self.osmalchemy.way(), u''),
253
                            (self.osmalchemy.node(51.3, 7.3), u''),
254
                            (self.osmalchemy.way(), u''),
255
                            (self.osmalchemy.node(51.4, 7.4), u'')]
256
        relation.members[3][0].nodes.append(relation.members[8][0])
257
258
        # Store everything
259
        self.session.add(relation)
260
        self.session.commit()
261
        # Ensure removal from ORM
262
        self.session.remove()
263
264
        # Query for way and check
265
        relation = self.session.query(self.osmalchemy.relation).first()
266
        self.assertEqual((relation.members[0][0].latitude, relation.members[0][0].longitude),
267
                         (51.0, 7.0))
268
        self.assertEqual((relation.members[2][0].latitude, relation.members[2][0].longitude),
269
                         (51.1, 7.1))
270
        self.assertEqual((relation.members[4][0].latitude, relation.members[4][0].longitude),
271
                         (51.2, 7.2))
272
        self.assertEqual((relation.members[6][0].latitude, relation.members[6][0].longitude),
273
                         (51.3, 7.3))
274
        self.assertEqual((relation.members[8][0].latitude, relation.members[8][0].longitude),
275
                         (51.4, 7.4))
276
        self.assertIs(relation.members[3][0].nodes[0], relation.members[8][0])
277
278
    def test_create_relation_with_nodes_and_ways_and_tags_everywhere(self):
279
        # Create way and add nodes and ways
280
        relation = self.osmalchemy.relation()
281
        relation.members = [(self.osmalchemy.node(51.0, 7.0), u''),
282
                            (self.osmalchemy.way(), u''),
283
                            (self.osmalchemy.node(51.1, 7.1), u''),
284
                            (self.osmalchemy.way(), u''),
285
                            (self.osmalchemy.node(51.2, 7.2), u''),
286
                            (self.osmalchemy.way(), u''),
287
                            (self.osmalchemy.node(51.3, 7.3), u''),
288
                            (self.osmalchemy.way(), u''),
289
                            (self.osmalchemy.node(51.4, 7.4), u'')]
290
        relation.tags[u"name"] = u"weirdest roads in Paris"
291
        relation.members[3][0].nodes.append(relation.members[8][0])
292
        relation.members[7][0].tags[u"foo"] = u"bar"
293
        relation.members[7][0].tags[u"bang"] = u"baz"
294
        relation.members[8][0].tags[u"name"] = u"Doppelknoten"
295
296
        # Store everything
297
        self.session.add(relation)
298
        self.session.commit()
299
        # Ensure removal from ORM
300
        self.session.remove()
301
302
        # Query for way and check
303
        relation = self.session.query(self.osmalchemy.relation).first()
304
        self.assertEqual((relation.members[0][0].latitude, relation.members[0][0].longitude),
305
                         (51.0, 7.0))
306
        self.assertEqual((relation.members[2][0].latitude, relation.members[2][0].longitude),
307
                         (51.1, 7.1))
308
        self.assertEqual((relation.members[4][0].latitude, relation.members[4][0].longitude),
309
                         (51.2, 7.2))
310
        self.assertEqual((relation.members[6][0].latitude, relation.members[6][0].longitude),
311
                         (51.3, 7.3))
312
        self.assertEqual((relation.members[8][0].latitude, relation.members[8][0].longitude),
313
                         (51.4, 7.4))
314
        self.assertIs(relation.members[3][0].nodes[0], relation.members[8][0])
315
        self.assertEqual(relation.tags[u"name"], u"weirdest roads in Paris")
316
        self.assertEqual(relation.members[7][0].tags[u"foo"], u"bar")
317
        self.assertEqual(relation.members[7][0].tags[u"bang"], u"baz")
318
        self.assertEqual(relation.members[8][0].tags, relation.members[3][0].nodes[0].tags)
319
320
class OSMAlchemyModelTestsSQLite(OSMAlchemyModelTests, unittest.TestCase):
321
    """ Tests run with SQLite """
322
323
    def setUp(self):
324
        self.engine = create_engine("sqlite:///:memory:")
325
        self.base = declarative_base(bind=self.engine)
326
        self.session = scoped_session(sessionmaker(bind=self.engine))
327
        self.osmalchemy = OSMAlchemy((self.engine, self.base, self.session))
328
        self.base.metadata.create_all()
329
        OSMAlchemyModelTests.setUp(self)
330
331
    def tearDown(self):
332
        self.session.remove()
333
        self.engine.dispose()
334
        OSMAlchemyModelTests.tearDown(self)
335
336 View Code Duplication
class OSMAlchemyModelTestsPostgres(OSMAlchemyModelTests, unittest.TestCase):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
337
    """ Tests run with PostgreSQL """
338
339
    def setUp(self):
340
        self.postgresql = Postgresql()
341
        self.engine = create_engine(self.postgresql.url())
342
        self.base = declarative_base(bind=self.engine)
343
        self.session = scoped_session(sessionmaker(bind=self.engine))
344
        self.osmalchemy = OSMAlchemy((self.engine, self.base, self.session))
345
        self.base.metadata.create_all()
346
        OSMAlchemyModelTests.setUp(self)
347
348
    def tearDown(self):
349
        self.session.remove()
350
        self.engine.dispose()
351
        self.postgresql.stop()
352
        OSMAlchemyModelTests.tearDown(self)
353
354 View Code Duplication
class OSMAlchemyModelTestsMySQL(OSMAlchemyModelTests, unittest.TestCase):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
355
    """ Tests run with MySQL """
356
357
    def setUp(self):
358
        self.mysql = Mysqld()
359
        self.engine = create_engine(self.mysql.url() + "?charset=utf8mb4")
360
        self.base = declarative_base(bind=self.engine)
361
        self.session = scoped_session(sessionmaker(bind=self.engine))
362
        self.osmalchemy = OSMAlchemy((self.engine, self.base, self.session))
363
        self.base.metadata.create_all()
364
        OSMAlchemyModelTests.setUp(self)
365
366
    def tearDown(self):
367
        self.session.remove()
368
        self.engine.dispose()
369
        self.mysql.stop()
370
        OSMAlchemyModelTests.tearDown(self)
371
372
class OSMAlchemyModelTestsFlaskSQLAlchemy(OSMAlchemyModelTests, unittest.TestCase):
373
    """ Tests run with SQLite """
374
375
    def setUp(self):
376
        app = Flask("test")
377
        db = FlaskSQLAlchemy(app)
378
        self.session = db.session
379
        self.osmalchemy = OSMAlchemy(db)
380
        db.create_all()
381
        OSMAlchemyModelTests.setUp(self)
382
383
    def tearDown(self):
384
        self.session.remove()
385
        OSMAlchemyModelTests.tearDown(self)
386
387
# Make runnable as standalone script
388
if __name__ == "__main__":
389
    unittest.main()
390