1 files changed, 73 insertions, 0 deletions
diff --git a/tests/hazmat/bindings/test_openssl.py b/tests/hazmat/bindings/test_openssl.py
index d1e85058..35eb7e8d 100644
--- a/tests/hazmat/bindings/test_openssl.py
+++ b/tests/hazmat/bindings/test_openssl.py
@@ -11,6 +11,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import pytest
+
 from cryptography.hazmat.bindings.openssl.binding import Binding
 
 
@@ -23,3 +25,74 @@ class TestOpenSSL(object):
 
     def test_is_available(self):
         assert Binding.is_available() is True
+
+    def test_crypto_lock_init(self):
+        b = Binding()
+        b.init_static_locks()
+        lock_cb = b.lib.CRYPTO_get_locking_callback()
+        assert lock_cb != b.ffi.NULL
+
+    def _skip_if_not_fallback_lock(self, b):
+        # only run this test if we are using our locking cb
+        original_cb = b.lib.CRYPTO_get_locking_callback()
+        if original_cb != b._lock_cb_handle:
+            pytest.skip(
+                "Not using the fallback Python locking callback "
+                "implementation. Probably because import _ssl set one"
+            )
+
+    def test_fallback_crypto_lock_via_openssl_api(self):
+        b = Binding()
+        b.init_static_locks()
+
+        self._skip_if_not_fallback_lock(b)
+
+        # check that the lock state changes appropriately
+        lock = b._locks[b.lib.CRYPTO_LOCK_SSL]
+
+        # starts out unlocked
+        assert lock.acquire(False)
+        lock.release()
+
+        b.lib.CRYPTO_lock(
+            b.lib.CRYPTO_LOCK | b.lib.CRYPTO_READ,
+            b.lib.CRYPTO_LOCK_SSL, b.ffi.NULL, 0
+        )
+
+        # becomes locked
+        assert not lock.acquire(False)
+
+        b.lib.CRYPTO_lock(
+            b.lib.CRYPTO_UNLOCK | b.lib.CRYPTO_READ,
+            b.lib.CRYPTO_LOCK_SSL, b.ffi.NULL, 0
+        )
+
+        # then unlocked
+        assert lock.acquire(False)
+        lock.release()
+
+    def test_fallback_crypto_lock_via_binding_api(self):
+        b = Binding()
+        b.init_static_locks()
+
+        self._skip_if_not_fallback_lock(b)
+
+        lock = b._locks[b.lib.CRYPTO_LOCK_SSL]
+
+        with pytest.raises(RuntimeError):
+            b._lock_cb(0, b.lib.CRYPTO_LOCK_SSL, "<test>", 1)
+
+        # errors shouldnt cause locking
+        assert lock.acquire(False)
+        lock.release()
+
+        b._lock_cb(b.lib.CRYPTO_LOCK | b.lib.CRYPTO_READ,
+                   b.lib.CRYPTO_LOCK_SSL, "<test>", 1)
+        # locked
+        assert not lock.acquire(False)
+
+        b._lock_cb(b.lib.CRYPTO_UNLOCK | b.lib.CRYPTO_READ,
+                   b.lib.CRYPTO_LOCK_SSL, "<test>", 1)
+        # unlocked
+        assert lock.acquire(False)
+        lock.release()