FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
datajoint-python/tests/test_connection.py at master · mspacek/datajoint-python · GitHub
mspacek
datajoint-python
Repository navigation
Code
Pull requests
Actions
Projects
Wiki
Security and quality
Insights
Expand file tree
Breadcrumbs
datajoint-python
/
tests
/
test_connection.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
113 lines (94 loc) · 3.39 KB
Breadcrumbs
datajoint-python
/
tests
/
test_connection.py
Copy path
File metadata and controls
113 lines (94 loc) · 3.39 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
"""
Collection of test cases to test connection module.
"""
from
nose
.
tools
import
assert_true
,
assert_equal
import
datajoint
as
dj
import
numpy
as
np
from
datajoint
import
DataJointError
from
.
import
CONN_INFO
,
PREFIX
def
test_dj_conn
():
"""
Should be able to establish a connection
"""
c
=
dj
.
conn
(
**
CONN_INFO
)
assert_true
(
c
.
is_connected
)
def
test_dj_connection_class
():
"""
Should be able to establish a connection
"""
c
=
dj
.
Connection
(
**
CONN_INFO
)
assert_true
(
c
.
is_connected
)
def
test_persistent_dj_conn
():
"""
conn() method should provide persistent connection across calls.
Setting reset=True should create a new persistent connection.
"""
c1
=
dj
.
conn
(
**
CONN_INFO
)
c2
=
dj
.
conn
()
c3
=
dj
.
conn
(
**
CONN_INFO
)
c4
=
dj
.
conn
(
reset
=
True
,
**
CONN_INFO
)
c5
=
dj
.
conn
(
**
CONN_INFO
)
assert_true
(
c1
is
c2
)
assert_true
(
c1
is
c3
)
assert_true
(
c1
is
not
c4
)
assert_true
(
c4
is
c5
)
def
test_repr
():
c1
=
dj
.
conn
(
**
CONN_INFO
)
assert_true
(
'disconnected'
not
in
repr
(
c1
)
and
'connected'
in
repr
(
c1
))
class
TestTransactions
:
"""
test transaction management
"""
schema
=
dj
.
Schema
(
PREFIX
+
'_transactions'
,
locals
(),
connection
=
dj
.
conn
(
**
CONN_INFO
))
@
schema
class
Subjects
(
dj
.
Manual
):
definition
=
"""
#Basic subject
subject_id : int # unique subject id
---
real_id : varchar(40) # real-world name
species = "mouse" : enum('mouse', 'monkey', 'human') # species
"""
@
classmethod
def
setup_class
(
cls
):
cls
.
relation
=
cls
.
Subjects
()
cls
.
conn
=
dj
.
conn
(
**
CONN_INFO
)
def
teardown
(
self
):
self
.
relation
.
delete_quick
()
def
test_active
(
self
):
with
self
.
conn
.
transaction
as
conn
:
assert_true
(
conn
.
in_transaction
,
"Transaction is not active"
)
def
test_transaction_rollback
(
self
):
"""Test transaction cancellation using a with statement"""
tmp
=
np
.
array
([
(
1
,
'Peter'
,
'mouse'
),
(
2
,
'Klara'
,
'monkey'
)
],
self
.
relation
.
heading
.
as_dtype
)
self
.
relation
.
delete
()
with
self
.
conn
.
transaction
:
self
.
relation
.
insert1
(
tmp
[
0
])
try
:
with
self
.
conn
.
transaction
:
self
.
relation
.
insert1
(
tmp
[
1
])
raise
DataJointError
(
"Testing rollback"
)
except
DataJointError
:
pass
assert_equal
(
len
(
self
.
relation
),
1
,
"Length is not 1. Expected because rollback should have happened."
)
assert_equal
(
len
(
self
.
relation
&
'subject_id = 2'
),
0
,
"Length is not 0. Expected because rollback should have happened."
)
def
test_cancel
(
self
):
"""Tests cancelling a transaction explicitly"""
tmp
=
np
.
array
([
(
1
,
'Peter'
,
'mouse'
),
(
2
,
'Klara'
,
'monkey'
)
],
self
.
relation
.
heading
.
as_dtype
)
self
.
relation
.
delete_quick
()
self
.
relation
.
insert1
(
tmp
[
0
])
self
.
conn
.
start_transaction
()
self
.
relation
.
insert1
(
tmp
[
1
])
self
.
conn
.
cancel_transaction
()
assert_equal
(
len
(
self
.
relation
),
1
,
"Length is not 1. Expected because rollback should have happened."
)
assert_equal
(
len
(
self
.
relation
&
'subject_id = 2'
),
0
,
"Length is not 0. Expected because rollback should have happened."
)
Back
|
FazBrowse Home
|
New Git URL