-
Notifications
You must be signed in to change notification settings - Fork 6
/
Copy pathtest_db.py
86 lines (68 loc) · 2.72 KB
/
test_db.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
from os.path import join
import pytest
from sqlalchemy import Table
from snowexsql.db import get_db, get_table_attributes
from snowexsql.tables import ImageData, LayerData, PointData, SiteData
from .sql_test_base import DBSetup
class TestDB(DBSetup):
base_atts = ['site_name', 'date', 'site_id']
single_loc_atts = ['elevation', 'geom', 'time']
meas_atts = ['instrument', 'type', 'units', 'observers']
site_atts = base_atts + single_loc_atts + \
['slope_angle', 'aspect', 'air_temp', 'total_depth',
'weather_description', 'precip', 'sky_cover', 'wind',
'ground_condition', 'ground_roughness',
'ground_vegetation', 'vegetation_height',
'tree_canopy', 'site_notes']
point_atts = single_loc_atts + meas_atts + \
['version_number', 'equipment', 'value']
layer_atts = single_loc_atts + meas_atts + \
['depth', 'value', 'bottom_depth', 'comments', 'sample_a',
'sample_b', 'sample_c']
raster_atts = meas_atts + ['raster', 'description']
def setup_class(self):
"""
Setup the database one time for testing
"""
super().setup_class()
site_fname = join(self.data_dir, 'site_details.csv')
# only reflect the tables we will use
self.metadata.reflect(self.engine, only=['points', 'layers'])
def test_point_structure(self):
"""
Tests our tables are in the database
"""
t = Table("points", self.metadata, autoload=True)
columns = [m.key for m in t.columns]
for c in self.point_atts:
assert c in columns
def test_layer_structure(self):
"""
Tests our tables are in the database
"""
t = Table("layers", self.metadata, autoload=True)
columns = [m.key for m in t.columns]
for c in self.layer_atts:
assert c in columns
@pytest.mark.parametrize("DataCls,attributes", [
(SiteData, site_atts),
(PointData, point_atts),
(LayerData, layer_atts),
(ImageData, raster_atts)])
def test_get_table_attributes(self, DataCls, attributes):
"""
Test we return a correct list of table columns from db.py
"""
atts = get_table_attributes(DataCls)
for c in attributes:
assert c in atts
# Independent Tests
@pytest.mark.parametrize("return_metadata, expected_objs", [
(False, 2),
(True, 3)])
def test_getting_db(return_metadata, expected_objs):
"""
Test we can receive a connection and opt out of getting the metadata
"""
result = get_db('builder:db_builder@localhost/test', return_metadata=return_metadata)
assert len(result) == expected_objs