Skip to content

Commit 9993a0e

Browse files
technigetechnige
committed
Tidied up driver tests
1 parent 23efe78 commit 9993a0e

File tree

1 file changed

+101
-122
lines changed

1 file changed

+101
-122
lines changed

‎test/test_driver.py

Lines changed: 101 additions & 122 deletions
Original file line numberDiff line numberDiff line change
@@ -25,228 +25,207 @@
2525

2626
from neo4j.v1 import ServiceUnavailable, ProtocolError, READ_ACCESS, WRITE_ACCESS, \
2727
TRUST_ON_FIRST_USE, TRUST_CUSTOM_CA_SIGNED_CERTIFICATES, GraphDatabase, basic_auth, \
28-
custom_auth, SSL_AVAILABLE, SessionExpired, DirectDriver
28+
custom_auth, SSL_AVAILABLE, SessionExpired, DirectDriver, RoutingDriver
2929
from test.util import ServerTestCase, StubCluster
3030

3131
BOLT_URI = "bolt://localhost:7687"
3232
BOLT_ROUTING_URI = "bolt+routing://localhost:7687"
33-
AUTH_TOKEN = basic_auth("neotest", "neotest")
3433

34+
TEST_USER = "neotest"
35+
TEST_PASSWORD = "neotest"
36+
AUTH_TOKEN = basic_auth(TEST_USER, TEST_PASSWORD)
3537

36-
class DriverTestCase(ServerTestCase):
3738

38-
def test_driver_with_block(self):
39-
with GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=False) as driver:
40-
assert isinstance(driver, DirectDriver)
39+
class BasicAuthTestCase(ServerTestCase):
40+
41+
def test_can_provide_realm_with_basic_auth_token(self):
42+
token = basic_auth(TEST_USER, TEST_PASSWORD, "native")
43+
with GraphDatabase.driver(BOLT_URI, auth=token) as driver:
44+
with driver.session() as session:
45+
result = session.run("RETURN 1").consume()
46+
assert result is not None
47+
48+
49+
class CustomAuthTestCase(ServerTestCase):
50+
51+
def test_can_create_custom_auth_token(self):
52+
token = custom_auth(TEST_USER, TEST_PASSWORD, "native", "basic")
53+
with GraphDatabase.driver(BOLT_URI, auth=token) as driver:
54+
with driver.session() as session:
55+
result = session.run("RETURN 1").consume()
56+
assert result is not None
57+
58+
def test_can_create_custom_auth_token_with_additional_parameters(self):
59+
token = custom_auth(TEST_USER, TEST_PASSWORD, "native", "basic", secret=42)
60+
with GraphDatabase.driver(BOLT_URI, auth=token) as driver:
61+
with driver.session() as session:
62+
result = session.run("RETURN 1").consume()
63+
assert result is not None
64+
65+
66+
class DriverTestCase(ServerTestCase):
4167

4268
def test_must_use_valid_url_scheme(self):
4369
with self.assertRaises(ProtocolError):
4470
GraphDatabase.driver("x://xxx", auth=AUTH_TOKEN)
4571

4672
def test_connections_are_reused(self):
47-
driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN)
48-
session_1 = driver.session()
49-
connection_1 = session_1.connection
50-
session_1.close()
51-
session_2 = driver.session()
52-
connection_2 = session_2.connection
53-
session_2.close()
54-
assert connection_1 is connection_2
55-
56-
def test_connections_are_not_shared_between_sessions(self):
57-
driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN)
58-
session_1 = driver.session()
59-
session_2 = driver.session()
60-
try:
61-
assert session_1.connection is not session_2.connection
62-
finally:
73+
with GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN) as driver:
74+
session_1 = driver.session()
75+
connection_1 = session_1.connection
6376
session_1.close()
77+
session_2 = driver.session()
78+
connection_2 = session_2.connection
6479
session_2.close()
80+
assert connection_1 is connection_2
6581

66-
def test_fail_nicely_when_connecting_to_http_port(self):
67-
driver = GraphDatabase.driver("bolt://localhost:7474", auth=AUTH_TOKEN, encrypted=False)
68-
with self.assertRaises(ServiceUnavailable) as context:
69-
driver.session()
70-
71-
def test_can_provide_realm_with_basic_auth_token(self):
72-
token = basic_auth("neotest", "neotest", "native")
73-
driver = GraphDatabase.driver("bolt://localhost", auth=token)
74-
session = driver.session()
75-
result = session.run("RETURN 1").consume()
76-
session.close()
77-
assert result is not None
78-
79-
def test_can_create_custom_auth_token(self):
80-
token = custom_auth("neotest", "neotest", "native", "basic")
81-
driver = GraphDatabase.driver("bolt://localhost", auth=token)
82-
session = driver.session()
83-
result = session.run("RETURN 1").consume()
84-
session.close()
85-
assert result is not None
82+
def test_connections_are_not_shared_between_sessions(self):
83+
with GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN) as driver:
84+
session_1 = driver.session()
85+
session_2 = driver.session()
86+
try:
87+
assert session_1.connection is not session_2.connection
88+
finally:
89+
session_1.close()
90+
session_2.close()
8691

87-
def test_can_create_custom_auth_token_with_additional_parameters(self):
88-
token = custom_auth("neotest", "neotest", "native", "basic", secret=42)
89-
driver = GraphDatabase.driver("bolt://localhost", auth=token)
90-
session = driver.session()
91-
result = session.run("RETURN 1").consume()
92-
session.close()
93-
assert result is not None
92+
def test_fail_nicely_when_connecting_to_http_port(self):
93+
uri = "bolt://localhost:7474"
94+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
95+
with self.assertRaises(ServiceUnavailable):
96+
driver.session()
9497

9598

9699
class DirectDriverTestCase(ServerTestCase):
97100

98-
def tearDown(self):
99-
self.await_all_servers()
101+
def test_bolt_uri_constructs_direct_driver(self):
102+
with GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=False) as driver:
103+
assert isinstance(driver, DirectDriver)
100104

101105
def test_direct_disconnect_on_run(self):
102-
self.start_stub_server(9001, "disconnect_on_run.script")
103-
uri = "bolt://127.0.0.1:9001"
104-
driver = GraphDatabase.driver(uri, auth=basic_auth("neo4j", "password"), encrypted=False)
105-
try:
106-
with driver.session() as session:
107-
with self.assertRaises(ServiceUnavailable):
108-
session.run("RETURN $x", {"x": 1}).consume()
109-
finally:
110-
driver.close()
106+
with StubCluster({9001: "disconnect_on_run.script"}):
107+
uri = "bolt://127.0.0.1:9001"
108+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
109+
with driver.session() as session:
110+
with self.assertRaises(ServiceUnavailable):
111+
session.run("RETURN $x", {"x": 1}).consume()
111112

112113
def test_direct_disconnect_on_pull_all(self):
113-
self.start_stub_server(9001, "disconnect_on_pull_all.script")
114-
uri = "bolt://127.0.0.1:9001"
115-
driver = GraphDatabase.driver(uri, auth=basic_auth("neo4j", "password"), encrypted=False)
116-
try:
117-
with driver.session() as session:
118-
with self.assertRaises(ServiceUnavailable):
119-
session.run("RETURN $x", {"x": 1}).consume()
120-
finally:
121-
driver.close()
114+
with StubCluster({9001: "disconnect_on_pull_all.script"}):
115+
uri = "bolt://127.0.0.1:9001"
116+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
117+
with driver.session() as session:
118+
with self.assertRaises(ServiceUnavailable):
119+
session.run("RETURN $x", {"x": 1}).consume()
122120

123121

124122
class RoutingDriverTestCase(ServerTestCase):
125123

124+
def test_bolt_plus_routing_uri_constructs_routing_driver(self):
125+
with StubCluster({9001: "router.script"}):
126+
uri = "bolt+routing://127.0.0.1:9001"
127+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
128+
assert isinstance(driver, RoutingDriver)
129+
126130
def test_cannot_discover_servers_on_non_router(self):
127131
with StubCluster({9001: "non_router.script"}):
128132
uri = "bolt+routing://127.0.0.1:9001"
129-
auth_token = basic_auth("neo4j", "password")
130133
with self.assertRaises(ServiceUnavailable):
131-
GraphDatabase.driver(uri, auth=auth_token, encrypted=False)
134+
_ = GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False)
132135

133136
def test_cannot_discover_servers_on_silent_router(self):
134137
with StubCluster({9001: "silent_router.script"}):
135138
uri = "bolt+routing://127.0.0.1:9001"
136-
auth_token = basic_auth("neo4j", "password")
137139
with self.assertRaises(ProtocolError):
138-
GraphDatabase.driver(uri, auth=auth_token, encrypted=False)
140+
_ = GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False)
139141

140142
def test_should_discover_servers_on_driver_construction(self):
141143
with StubCluster({9001: "router.script"}):
142144
uri = "bolt+routing://127.0.0.1:9001"
143-
auth_token = basic_auth("neo4j", "password")
144-
driver = GraphDatabase.driver(uri, auth=auth_token, encrypted=False)
145-
try:
145+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
146146
table = driver.pool.routing_table
147147
assert table.routers == {('127.0.0.1', 9001), ('127.0.0.1', 9002),
148148
('127.0.0.1', 9003)}
149149
assert table.readers == {('127.0.0.1', 9004), ('127.0.0.1', 9005)}
150150
assert table.writers == {('127.0.0.1', 9006)}
151-
finally:
152-
driver.close()
153151

154152
def test_should_be_able_to_read(self):
155153
with StubCluster({9001: "router.script", 9004: "return_1.script"}):
156154
uri = "bolt+routing://127.0.0.1:9001"
157-
auth_token = basic_auth("neo4j", "password")
158-
driver = GraphDatabase.driver(uri, auth=auth_token, encrypted=False)
159-
try:
155+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
160156
with driver.session(READ_ACCESS) as session:
161157
result = session.run("RETURN $x", {"x": 1})
162158
for record in result:
163159
assert record["x"] == 1
164160
assert session.connection.address == ('127.0.0.1', 9004)
165-
finally:
166-
driver.close()
167161

168162
def test_should_be_able_to_write(self):
169163
with StubCluster({9001: "router.script", 9006: "create_a.script"}):
170164
uri = "bolt+routing://127.0.0.1:9001"
171-
driver = GraphDatabase.driver(uri, auth=basic_auth("neo4j", "password"), encrypted=False)
172-
try:
165+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
173166
with driver.session(WRITE_ACCESS) as session:
174167
result = session.run("CREATE (a $x)", {"x": {"name": "Alice"}})
175168
assert not list(result)
176169
assert session.connection.address == ('127.0.0.1', 9006)
177-
finally:
178-
driver.close()
179170

180171
def test_should_be_able_to_write_as_default(self):
181172
with StubCluster({9001: "router.script", 9006: "create_a.script"}):
182173
uri = "bolt+routing://127.0.0.1:9001"
183-
auth_token = basic_auth("neo4j", "password")
184-
driver = GraphDatabase.driver(uri, auth=auth_token, encrypted=False)
185-
try:
174+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
186175
with driver.session() as session:
187176
result = session.run("CREATE (a $x)", {"x": {"name": "Alice"}})
188177
assert not list(result)
189178
assert session.connection.address == ('127.0.0.1', 9006)
190-
finally:
191-
driver.close()
192179

193180
def test_routing_disconnect_on_run(self):
194181
with StubCluster({9001: "router.script", 9004: "disconnect_on_run.script"}):
195182
uri = "bolt+routing://127.0.0.1:9001"
196-
auth_token = basic_auth("neo4j", "password")
197-
driver = GraphDatabase.driver(uri, auth=auth_token, encrypted=False)
198-
try:
183+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
199184
with driver.session(READ_ACCESS) as session:
200185
with self.assertRaises(SessionExpired):
201186
session.run("RETURN $x", {"x": 1}).consume()
202-
finally:
203-
driver.close()
204187

205188
def test_routing_disconnect_on_pull_all(self):
206189
with StubCluster({9001: "router.script", 9004: "disconnect_on_pull_all.script"}):
207190
uri = "bolt+routing://127.0.0.1:9001"
208-
auth_token = basic_auth("neo4j", "password")
209-
driver = GraphDatabase.driver(uri, auth=auth_token, encrypted=False)
210-
try:
191+
with GraphDatabase.driver(uri, auth=AUTH_TOKEN, encrypted=False) as driver:
211192
with driver.session(READ_ACCESS) as session:
212193
with self.assertRaises(SessionExpired):
213194
session.run("RETURN $x", {"x": 1}).consume()
214-
finally:
215-
driver.close()
216195

217196

218197
class SecurityTestCase(ServerTestCase):
219198

220199
def test_insecure_session_uses_normal_socket(self):
221-
driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=False)
222-
with driver.session() as session:
223-
connection = session.connection
224-
assert isinstance(connection.channel.socket, socket)
225-
assert connection.der_encoded_server_certificate is None
200+
with GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=False) as driver:
201+
with driver.session() as session:
202+
connection = session.connection
203+
assert isinstance(connection.channel.socket, socket)
204+
assert connection.der_encoded_server_certificate is None
226205

227206
@skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
228207
def test_tofu_session_uses_secure_socket(self):
229-
driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=True, trust=TRUST_ON_FIRST_USE)
230-
with driver.session() as session:
231-
connection = session.connection
232-
assert isinstance(connection.channel.socket, SSLSocket)
233-
assert connection.der_encoded_server_certificate is not None
208+
with GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, trust=TRUST_ON_FIRST_USE) as driver:
209+
with driver.session() as session:
210+
connection = session.connection
211+
assert isinstance(connection.channel.socket, SSLSocket)
212+
assert connection.der_encoded_server_certificate is not None
234213

235214
@skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
236215
def test_tofu_session_trusts_certificate_after_first_use(self):
237-
driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=True, trust=TRUST_ON_FIRST_USE)
238-
with driver.session() as session:
239-
connection = session.connection
240-
certificate = connection.der_encoded_server_certificate
241-
with driver.session() as session:
242-
connection = session.connection
243-
assert connection.der_encoded_server_certificate == certificate
216+
with GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, trust=TRUST_ON_FIRST_USE) as driver:
217+
with driver.session() as session:
218+
connection = session.connection
219+
certificate = connection.der_encoded_server_certificate
220+
with driver.session() as session:
221+
connection = session.connection
222+
assert connection.der_encoded_server_certificate == certificate
244223

245224
def test_routing_driver_not_compatible_with_tofu(self):
246225
with self.assertRaises(ValueError):
247-
GraphDatabase.driver(BOLT_ROUTING_URI, auth=AUTH_TOKEN, trust=TRUST_ON_FIRST_USE)
226+
_ = GraphDatabase.driver(BOLT_ROUTING_URI, auth=AUTH_TOKEN, trust=TRUST_ON_FIRST_USE)
248227

249228
def test_custom_ca_not_implemented(self):
250229
with self.assertRaises(NotImplementedError):
251-
GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN,
252-
trust=TRUST_CUSTOM_CA_SIGNED_CERTIFICATES)
230+
_ = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN,
231+
trust=TRUST_CUSTOM_CA_SIGNED_CERTIFICATES)

0 commit comments

Comments
 (0)