Updated examples

Author: technige

examples/test_examples.py

@@ -19,18 +19,26 @@
 # limitations under the License.
 
 
+from unittest import skip
+
+from neo4j.v1 import basic_auth, TRUST_ON_FIRST_USE, TRUST_SIGNED_CERTIFICATES
 from test.util import ServerTestCase
 
+# Do not change the contents of this tagged section without good reason*
 # tag::minimal-example-import[]
-from neo4j.v1 import GraphDatabase, basic_auth
+from neo4j.v1 import GraphDatabase
 # end::minimal-example-import[]
+# (* "good reason" is defined as knowing what you are doing)
+
+
+auth_token = basic_auth("neo4j", "password")
 
 
 class FreshDatabaseTestCase(ServerTestCase):
 
     def setUp(self):
         ServerTestCase.setUp(self)
-        session = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password")).session()
+        session = GraphDatabase.driver("bolt://localhost", auth=auth_token).session()
         session.run("MATCH (n) DETACH DELETE n")
         session.close()
 
@@ -68,40 +76,40 @@ def test_configuration(self):
 
     def test_tls_require_encryption(self):
         # tag::tls-require-encryption[]
-        # TODO: Unfortunately, this feature is not yet implemented for Python
-        pass
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token, encrypted=True)
         # end::tls-require-encryption[]
 
     def test_tls_trust_on_first_use(self):
         # tag::tls-trust-on-first-use[]
-        # TODO: Unfortunately, this feature is not yet implemented for Python
-        pass
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token, encrypted=True, trust=TRUST_ON_FIRST_USE)
         # end::tls-trust-on-first-use[]
+        assert driver
 
+    @skip("testing verified certificates not yet supported ")
     def test_tls_signed(self):
         # tag::tls-signed[]
-        # TODO: Unfortunately, this feature is not yet implemented for Python
-        pass
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token, encrypted=True, trust=TRUST_SIGNED_CERTIFICATES)
         # end::tls-signed[]
+        assert driver
 
     def test_statement(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::statement[]
         session.run("CREATE (person:Person {name: {name}})", {"name": "Neo"}).close()
         # end::statement[]
         session.close()
 
     def test_statement_without_parameters(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::statement-without-parameters[]
         session.run("CREATE (person:Person {name: 'Neo'})").close()
         # end::statement-without-parameters[]
         session.close()
 
     def test_result_cursor(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::result-cursor[]
         search_term = "hammer"
@@ -114,7 +122,7 @@ def test_result_cursor(self):
         session.close()
 
     def test_cursor_nesting(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::retain-result-query[]
         result = session.run("MATCH (person:Person) WHERE person.dept = {dept} "
@@ -127,7 +135,7 @@ def test_cursor_nesting(self):
         session.close()
 
     def test_result_retention(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::retain-result-process[]
         result = session.run("MATCH (person:Person) WHERE person.dept = {dept} "
@@ -142,7 +150,7 @@ def test_result_retention(self):
         session.close()
 
     def test_transaction_commit(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::transaction-commit[]
         tx = session.begin_transaction()
@@ -156,7 +164,7 @@ def test_transaction_commit(self):
         session.close()
 
     def test_transaction_rollback(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::transaction-rollback[]
         tx = session.begin_transaction()
@@ -170,7 +178,7 @@ def test_transaction_rollback(self):
         session.close()
 
     def test_result_summary_query_profile(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::result-summary-query-profile[]
         result = session.run("PROFILE MATCH (p:Person {name: {name}}) "
@@ -183,7 +191,7 @@ def test_result_summary_query_profile(self):
         session.close()
 
     def test_result_summary_notifications(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token)
         session = driver.session()
         # tag::result-summary-notifications[]
         result = session.run("EXPLAIN MATCH (a), (b) RETURN a,b")

neo4j/v1/connection.py

@@ -33,7 +33,7 @@
 from struct import pack as struct_pack, unpack as struct_unpack, unpack_from as struct_unpack_from
 
 from .constants import DEFAULT_PORT, DEFAULT_USER_AGENT, KNOWN_HOSTS, MAGIC_PREAMBLE, \
-    SECURITY_DEFAULT, SECURITY_TRUST_ON_FIRST_USE
+    TRUST_DEFAULT, TRUST_ON_FIRST_USE
 from .compat import hex2
 from .exceptions import ProtocolError
 from .packstream import Packer, Unpacker
@@ -398,8 +398,8 @@ def connect(host, port=None, ssl_context=None, **config):
             der_encoded_server_certificate = s.getpeercert(binary_form=True)
             if der_encoded_server_certificate is None:
                 raise ProtocolError("When using a secure socket, the server should always provide a certificate")
-            security = config.get("security", SECURITY_DEFAULT)
-            if security == SECURITY_TRUST_ON_FIRST_USE:
+            trust = config.get("trust", TRUST_DEFAULT)
+            if trust == TRUST_ON_FIRST_USE:
                 store = PersonalCertificateStore()
                 if not store.match_or_trust(host, der_encoded_server_certificate):
                     raise ProtocolError("Server certificate does not match known certificate for %r; check "

neo4j/v1/constants.py

@@ -31,8 +31,9 @@
 
 MAGIC_PREAMBLE = 0x6060B017
 
-SECURITY_NONE = 0
-SECURITY_TRUST_ON_FIRST_USE = 1
-SECURITY_VERIFIED = 2
+ENCRYPTED_DEFAULT = True
 
-SECURITY_DEFAULT = SECURITY_TRUST_ON_FIRST_USE
+TRUST_ON_FIRST_USE = 0
+TRUST_SIGNED_CERTIFICATES = 1
+
+TRUST_DEFAULT = TRUST_ON_FIRST_USE

neo4j/v1/session.py

@@ -33,7 +33,7 @@ class which can be used to obtain `Driver` instances that are used for
 
 from .compat import integer, string, urlparse
 from .connection import connect, Response, RUN, PULL_ALL
-from .constants import SECURITY_NONE, SECURITY_VERIFIED, SECURITY_DEFAULT
+from .constants import ENCRYPTED_DEFAULT, TRUST_DEFAULT, TRUST_SIGNED_CERTIFICATES
 from .exceptions import CypherError, ResultError
 from .typesystem import hydrated
 
@@ -99,11 +99,12 @@ def __init__(self, url, **config):
         self.config = config
         self.max_pool_size = config.get("max_pool_size", DEFAULT_MAX_POOL_SIZE)
         self.session_pool = deque()
-        self.security = security = config.get("security", SECURITY_DEFAULT)
-        if security > SECURITY_NONE:
+        self.encrypted = encrypted = config.get("encrypted", ENCRYPTED_DEFAULT)
+        self.trust = trust = config.get("trust", TRUST_DEFAULT)
+        if encrypted:
             ssl_context = SSLContext(PROTOCOL_SSLv23)
             ssl_context.options |= OP_NO_SSLv2
-            if security >= SECURITY_VERIFIED:
+            if trust >= TRUST_SIGNED_CERTIFICATES:
                 ssl_context.verify_mode = CERT_REQUIRED
             ssl_context.load_default_certs(Purpose.SERVER_AUTH)
             self.ssl_context = ssl_context

test/tck/tck_util.py

@@ -18,11 +18,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from neo4j.v1 import GraphDatabase, Relationship, Node, Path, SECURITY_NONE, basic_auth
+
+from neo4j.v1 import GraphDatabase, Relationship, Node, Path, basic_auth
 from neo4j.v1.compat import string
 
 
-driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"), security=SECURITY_NONE)
+driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"), encrypted=False)
+
 
 def send_string(text):
     session = driver.session()

test/test_session.py

@@ -23,7 +23,7 @@
 from ssl import SSLSocket
 
 from mock import patch
-from neo4j.v1.constants import SECURITY_NONE, SECURITY_TRUST_ON_FIRST_USE
+from neo4j.v1.constants import TRUST_ON_FIRST_USE
 from neo4j.v1.exceptions import CypherError, ResultError
 from neo4j.v1.session import GraphDatabase, basic_auth, Record, record
 from neo4j.v1.typesystem import Node, Relationship, Path
@@ -92,26 +92,26 @@ class SecurityTestCase(ServerTestCase):
 
     def test_default_session_uses_tofu(self):
         driver = GraphDatabase.driver("bolt://localhost")
-        assert driver.security == SECURITY_TRUST_ON_FIRST_USE
+        assert driver.trust == TRUST_ON_FIRST_USE
 
     def test_insecure_session_uses_normal_socket(self):
-        driver = GraphDatabase.driver("bolt://localhost",  auth=auth_token, security=SECURITY_NONE)
+        driver = GraphDatabase.driver("bolt://localhost",  auth=auth_token, encrypted=False)
         session = driver.session()
         connection = session.connection
         assert isinstance(connection.channel.socket, socket)
         assert connection.der_encoded_server_certificate is None
         session.close()
 
     def test_tofu_session_uses_secure_socket(self):
-        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token, security=SECURITY_TRUST_ON_FIRST_USE)
+        driver = GraphDatabase.driver("bolt://localhost", auth=auth_token, encrypted=True, trust=TRUST_ON_FIRST_USE)
         session = driver.session()
         connection = session.connection
         assert isinstance(connection.channel.socket, SSLSocket)
         assert connection.der_encoded_server_certificate is not None
         session.close()
 
     def test_tofu_session_trusts_certificate_after_first_use(self):
-        driver = GraphDatabase.driver("bolt://localhost",  auth=auth_token, security=SECURITY_TRUST_ON_FIRST_USE)
+        driver = GraphDatabase.driver("bolt://localhost",  auth=auth_token, encrypted=True, trust=TRUST_ON_FIRST_USE)
         session = driver.session()
         connection = session.connection
         certificate = connection.der_encoded_server_certificate