|
19 | 19 | # limitations under the License. |
20 | 20 |
|
21 | 21 |
|
| 22 | +import pytest |
| 23 | + |
22 | 24 | from neo4j.exceptions import ServiceUnavailable |
23 | 25 |
|
24 | | -from neo4j import GraphDatabase, BoltDriver |
| 26 | +from neo4j import ( |
| 27 | + GraphDatabase, |
| 28 | + BoltDriver, |
| 29 | +) |
| 30 | + |
| 31 | +from tests.stub.conftest import ( |
| 32 | + StubCluster, |
| 33 | +) |
25 | 34 |
|
26 | | -from tests.stub.conftest import StubTestCase, StubCluster |
| 35 | +# python -m pytest tests/stub/test_directdriver.py |
27 | 36 |
|
28 | 37 |
|
29 | | -class BoltDriverTestCase(StubTestCase): |
| 38 | +@pytest.mark.parametrize( |
| 39 | + "test_script", |
| 40 | + [ |
| 41 | + "v3/empty.script", |
| 42 | + "v4x0/empty.script", |
| 43 | + ] |
| 44 | +) |
| 45 | +def test_bolt_uri_constructs_bolt_driver(driver_info, test_script): |
| 46 | + # python -m pytest tests/stub/test_directdriver.py -s -v -k test_bolt_uri_constructs_bolt_driver |
| 47 | + with StubCluster(test_script): |
| 48 | + uri = "bolt://127.0.0.1:9001" |
| 49 | + with GraphDatabase.driver(uri, auth=driver_info["auth_token"]) as driver: |
| 50 | + assert isinstance(driver, BoltDriver) |
30 | 51 |
|
31 | | - def test_bolt_uri_constructs_bolt_driver(self): |
32 | | - with StubCluster("v3/empty.script"): |
33 | | - uri = "bolt://127.0.0.1:9001" |
34 | | - with GraphDatabase.driver(uri, auth=self.auth_token) as driver: |
35 | | - assert isinstance(driver, BoltDriver) |
36 | 52 |
|
37 | | - def test_direct_disconnect_on_run(self): |
38 | | - with StubCluster("v3/disconnect_on_run.script"): |
39 | | - uri = "bolt://127.0.0.1:9001" |
40 | | - with GraphDatabase.driver(uri, auth=self.auth_token) as driver: |
41 | | - with self.assertRaises(ServiceUnavailable): |
42 | | - with driver.session() as session: |
43 | | - session.run("RETURN $x", {"x": 1}).consume() |
| 53 | +def test_direct_disconnect_on_run(driver_info): |
| 54 | + # python -m pytest tests/stub/test_directdriver.py -s -v -k test_direct_disconnect_on_run |
| 55 | + with StubCluster("v3/disconnect_on_run.script"): |
| 56 | + uri = "bolt://127.0.0.1:9001" |
| 57 | + with GraphDatabase.driver(uri, auth=driver_info["auth_token"]) as driver: |
| 58 | + with pytest.raises(ServiceUnavailable): |
| 59 | + with driver.session() as session: |
| 60 | + session.run("RETURN $x", {"x": 1}).consume() |
44 | 61 |
|
45 | | - def test_direct_disconnect_on_pull_all(self): |
46 | | - with StubCluster("v3/disconnect_on_pull_all.script"): |
47 | | - uri = "bolt://127.0.0.1:9001" |
48 | | - with GraphDatabase.driver(uri, auth=self.auth_token) as driver: |
49 | | - with self.assertRaises(ServiceUnavailable): |
50 | | - with driver.session() as session: |
51 | | - session.run("RETURN $x", {"x": 1}).consume() |
52 | 62 |
|
53 | | - def test_direct_session_close_after_server_close(self): |
54 | | - with StubCluster("v3/disconnect_after_init.script"): |
55 | | - uri = "bolt://127.0.0.1:9001" |
56 | | - with GraphDatabase.driver(uri, auth=self.auth_token, max_retry_time=0, |
57 | | - acquire_timeout=3, user_agent="test") as driver: |
| 63 | +def test_direct_disconnect_on_pull_all(driver_info): |
| 64 | + # python -m pytest tests/stub/test_directdriver.py -s -v -k test_direct_disconnect_on_pull_all |
| 65 | + with StubCluster("v3/disconnect_on_pull_all.script"): |
| 66 | + uri = "bolt://127.0.0.1:9001" |
| 67 | + with GraphDatabase.driver(uri, auth=driver_info["auth_token"]) as driver: |
| 68 | + with pytest.raises(ServiceUnavailable): |
58 | 69 | with driver.session() as session: |
59 | | - with self.assertRaises(ServiceUnavailable): |
60 | | - session.write_transaction(lambda tx: tx.run("CREATE (a:Item)")) |
| 70 | + session.run("RETURN $x", {"x": 1}).consume() |
| 71 | + |
| 72 | + |
| 73 | +def test_direct_session_close_after_server_close(driver_info): |
| 74 | + # python -m pytest tests/stub/test_directdriver.py -s -v -k test_direct_session_close_after_server_close |
| 75 | + with StubCluster("v3/disconnect_after_init.script"): |
| 76 | + uri = "bolt://127.0.0.1:9001" |
| 77 | + with GraphDatabase.driver(uri, auth=driver_info["auth_token"], max_retry_time=0, |
| 78 | + acquire_timeout=3, user_agent="test") as driver: |
| 79 | + with driver.session() as session: |
| 80 | + with pytest.raises(ServiceUnavailable): |
| 81 | + session.write_transaction(lambda tx: tx.run("CREATE (a:Item)")) |
0 commit comments