1
- from nose . tools import assert_true , assert_list_equal , assert_false , raises
1
+ import pytest
2
2
import hashlib
3
+ from pathlib import Path
3
4
from datajoint import DataJointError
5
+ import datajoint as dj
4
6
from .schema_university import *
5
- from . import PREFIX , CONN_INFO
7
+ from . import PREFIX , schema_university
6
8
7
9
8
10
def _hash4 (table ):
9
- """hash of table contents"""
11
+ """Hash of table contents"""
10
12
data = table .fetch (order_by = "KEY" , as_dict = True )
11
13
blob = dj .blob .pack (data , compress = False )
12
14
return hashlib .md5 (blob ).digest ().hex ()[:4 ]
13
15
14
16
15
- @raises (DataJointError )
16
- def test_activate_unauthorized ():
17
- schema .activate ("unauthorized" , connection = dj .conn (** CONN_INFO ))
18
-
19
-
20
- def test_activate ():
21
- schema .activate (
22
- PREFIX + "_university" , connection = dj .conn (** CONN_INFO )
23
- ) # deferred activation
17
+ @pytest .fixture
18
+ def schema_uni_inactive ():
19
+ schema = dj .Schema (context = schema_university .LOCALS_UNI )
20
+ schema (Student )
21
+ schema (Department )
22
+ schema (StudentMajor )
23
+ schema (Course )
24
+ schema (Term )
25
+ schema (Section )
26
+ schema (CurrentTerm )
27
+ schema (Enroll )
28
+ schema (LetterGrade )
29
+ schema (Grade )
30
+ yield schema
31
+ schema .drop ()
32
+
33
+
34
+ @pytest .fixture
35
+ def schema_uni (db_creds_test , schema_uni_inactive , connection_test ):
36
+ # Deferred activation
37
+ schema_uni_inactive .activate (
38
+ PREFIX + "_university" , connection = dj .conn (** db_creds_test )
39
+ )
24
40
# --------------- Fill University -------------------
41
+ test_data_dir = Path (__file__ ).parent / "data"
25
42
for table in (
26
43
Student ,
27
44
Department ,
@@ -33,12 +50,19 @@ def test_activate():
33
50
Enroll ,
34
51
Grade ,
35
52
):
36
- from pathlib import Path
53
+ path = test_data_dir / Path (table .__name__ + ".csv" )
54
+ assert path .is_file (), f"File { path } is not a file"
55
+ assert path .exists (), f"File { path } does not exist"
56
+ table ().insert (path )
57
+ return schema_uni_inactive
58
+
37
59
38
- table ().insert (Path ("./data/" + table .__name__ + ".csv" ))
60
+ def test_activate_unauthorized (schema_uni_inactive , db_creds_test , connection_test ):
61
+ with pytest .raises (DataJointError ):
62
+ schema_uni_inactive .activate ("unauthorized" , connection = dj .conn (** db_creds_test ))
39
63
40
64
41
- def test_fill ():
65
+ def test_fill (schema_uni ):
42
66
"""check that the randomized tables are consistently defined"""
43
67
# check randomized tables
44
68
assert len (Student ()) == 300 and _hash4 (Student ) == "1e1a"
@@ -48,7 +72,7 @@ def test_fill():
48
72
assert len (Grade ()) == 3027 and _hash4 (Grade ) == "4a9d"
49
73
50
74
51
- def test_restrict ():
75
+ def test_restrict (schema_uni ):
52
76
"""
53
77
test diverse restrictions from the university database.
54
78
This test relies on a specific instantiation of the database.
@@ -90,7 +114,7 @@ def test_restrict():
90
114
assert len (special ) == 158
91
115
92
116
93
- def test_advanced_join ():
117
+ def test_advanced_join (schema_uni ):
94
118
"""test advanced joins"""
95
119
# Students with ungraded courses in current term
96
120
ungraded = Enroll * CurrentTerm - Grade
@@ -102,14 +126,14 @@ def test_advanced_join():
102
126
assert len (ungraded .join (major )) == len (ungraded & major ) == 31
103
127
104
128
105
- def test_union ():
129
+ def test_union (schema_uni ):
106
130
# effective left join Enroll with Major
107
131
q1 = (Enroll & "student_id=101" ) + (Enroll & "student_id=102" )
108
132
q2 = Enroll & "student_id in (101, 102)"
109
133
assert len (q1 ) == len (q2 ) == 41
110
134
111
135
112
- def test_aggr ():
136
+ def test_aggr (schema_uni ):
113
137
avg_grade_per_course = Course .aggr (
114
138
Grade * LetterGrade , avg_grade = "round(avg(points), 2)"
115
139
)
0 commit comments