Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix incorrect asserts in test and ensure connections are closed #3004

Merged
merged 3 commits into from
Feb 22, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
31 changes: 21 additions & 10 deletions tests/test_ssl.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,13 +26,15 @@ def test_ssl_with_invalid_cert(self, request):
sslclient = redis.from_url(ssl_url)
with pytest.raises(ConnectionError) as e:
sslclient.ping()
assert "SSL: CERTIFICATE_VERIFY_FAILED" in str(e)
assert "SSL: CERTIFICATE_VERIFY_FAILED" in str(e)
sslclient.close()

def test_ssl_connection(self, request):
ssl_url = request.config.option.redis_ssl_url
p = urlparse(ssl_url)[1].split(":")
r = redis.Redis(host=p[0], port=p[1], ssl=True, ssl_cert_reqs="none")
assert r.ping()
r.close()

def test_ssl_connection_without_ssl(self, request):
ssl_url = request.config.option.redis_ssl_url
Expand All @@ -41,7 +43,8 @@ def test_ssl_connection_without_ssl(self, request):

with pytest.raises(ConnectionError) as e:
r.ping()
assert "Connection closed by server" in str(e)
assert "Connection closed by server" in str(e)
r.close()

def test_validating_self_signed_certificate(self, request):
ssl_url = request.config.option.redis_ssl_url
Expand All @@ -56,6 +59,7 @@ def test_validating_self_signed_certificate(self, request):
ssl_ca_certs=self.SERVER_CERT,
)
assert r.ping()
r.close()

def test_validating_self_signed_string_certificate(self, request):
with open(self.SERVER_CERT) as f:
Expand All @@ -72,6 +76,7 @@ def test_validating_self_signed_string_certificate(self, request):
ssl_ca_data=cert_data,
)
assert r.ping()
r.close()

def _create_oscp_conn(self, request):
ssl_url = request.config.option.redis_ssl_url
Expand All @@ -92,22 +97,25 @@ def _create_oscp_conn(self, request):
def test_ssl_ocsp_called(self, request):
r = self._create_oscp_conn(request)
with pytest.raises(RedisError) as e:
assert r.ping()
assert "cryptography not installed" in str(e)
r.ping()
assert "cryptography is not installed" in str(e)
r.close()

@skip_if_nocryptography()
def test_ssl_ocsp_called_withcrypto(self, request):
r = self._create_oscp_conn(request)
with pytest.raises(ConnectionError) as e:
assert r.ping()
assert "No AIA information present in ssl certificate" in str(e)
assert "No AIA information present in ssl certificate" in str(e)
r.close()

# rediss://, url based
ssl_url = request.config.option.redis_ssl_url
sslclient = redis.from_url(ssl_url)
with pytest.raises(ConnectionError) as e:
sslclient.ping()
assert "No AIA information present in ssl certificate" in str(e)
assert "No AIA information present in ssl certificate" in str(e)
sslclient.close()

@skip_if_nocryptography()
def test_valid_ocsp_cert_http(self):
Expand All @@ -132,7 +140,7 @@ def test_revoked_ocsp_certificate(self):
ocsp = OCSPVerifier(wrapped, hostname, 443)
with pytest.raises(ConnectionError) as e:
assert ocsp.is_valid()
assert "REVOKED" in str(e)
assert "REVOKED" in str(e)

@skip_if_nocryptography()
def test_unauthorized_ocsp(self):
Expand All @@ -157,7 +165,7 @@ def test_ocsp_not_present_in_response(self):
ocsp = OCSPVerifier(wrapped, hostname, 443)
with pytest.raises(ConnectionError) as e:
assert ocsp.is_valid()
assert "from the" in str(e)
assert "from the" in str(e)

@skip_if_nocryptography()
def test_unauthorized_then_direct(self):
Expand Down Expand Up @@ -193,6 +201,7 @@ def test_mock_ocsp_staple(self, request):

with pytest.raises(RedisError):
r.ping()
r.close()

ctx = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD)
ctx.use_certificate_file(self.SERVER_CERT)
Expand All @@ -213,7 +222,8 @@ def test_mock_ocsp_staple(self, request):

with pytest.raises(ConnectionError) as e:
r.ping()
assert "no ocsp response present" in str(e)
assert "no ocsp response present" in str(e)
r.close()

r = redis.Redis(
host=p[0],
Expand All @@ -228,4 +238,5 @@ def test_mock_ocsp_staple(self, request):

with pytest.raises(ConnectionError) as e:
r.ping()
assert "no ocsp response present" in str(e)
assert "no ocsp response present" in str(e)
r.close()