zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-07-25 12:08:40 +02:00
Code Issues Projects Releases Packages Wiki Activity

suppressed logging from all test cases; made slight changes to handling of exceptions in connection pooling

Browse source
This commit is contained in:
David Kennedy 2024-01-29 18:22:35 -05:00
parent bf8d9159ce
commit 0546bd08e5
No known key found for this signature in database
GPG key ID: 6528A5386E66B96B
12 changed files with 3145 additions and 3166 deletions
Download patch file Download diff file Expand all files Collapse all files

51
src/epplibwrapper/tests/common.py Normal file
View file

@ -0,0 +1,51 @@
import os
import logging
from contextlib import contextmanager
def get_handlers():
"""Obtain pointers to all StreamHandlers."""
handlers = {}
rootlogger = logging.getLogger()
for h in rootlogger.handlers:
if isinstance(h, logging.StreamHandler):
handlers[h.name] = h
for logger in logging.Logger.manager.loggerDict.values():
if not isinstance(logger, logging.PlaceHolder):
for h in logger.handlers:
if isinstance(h, logging.StreamHandler):
handlers[h.name] = h
return handlers
@contextmanager
def less_console_noise():
"""
Context manager to use in tests to silence console logging.
This is helpful on tests which trigger console messages
(such as errors) which are normal and expected.
It can easily be removed to debug a failing test.
"""
restore = {}
handlers = get_handlers()
devnull = open(os.devnull, "w")
# redirect all the streams
for handler in handlers.values():
prior = handler.setStream(devnull)
restore[handler.name] = prior
try:
# run the test
yield
finally:
# restore the streams
for handler in handlers.values():
handler.setStream(restore[handler.name])
# close the file we opened
devnull.close()

101
src/epplibwrapper/tests/test_pool.py
View file

@ -9,7 +9,7 @@ from epplibwrapper.socket import Socket
from epplibwrapper.utility.pool import EPPConnectionPool
from registrar.models.domain import registry
from contextlib import ExitStack
from .common import less_console_noise
import logging
try:
@ -135,23 +135,26 @@ class TestConnectionPool(TestCase):
stack.enter_context(patch.object(EPPConnectionPool, "kill_all_connections", do_nothing))
stack.enter_context(patch.object(SocketTransport, "send", self.fake_send))
stack.enter_context(patch.object(SocketTransport, "receive", fake_receive))
# Restart the connection pool
registry.start_connection_pool()
# Pool should be running, and be the right size
self.assertEqual(registry.pool_status.connection_success, True)
self.assertEqual(registry.pool_status.pool_running, True)
with less_console_noise():
# Restart the connection pool
registry.start_connection_pool()
# Pool should be running, and be the right size
self.assertEqual(registry.pool_status.connection_success, True)
self.assertEqual(registry.pool_status.pool_running, True)
# Send a command
result = registry.send(commands.InfoDomain(name="test.gov"), cleaned=True)
# Send a command
result = registry.send(commands.InfoDomain(name="test.gov"), cleaned=True)
# Should this ever fail, it either means that the schema has changed,
# or the pool is broken.
# If the schema has changed: Update the associated infoDomain.xml file
self.assertEqual(result.__dict__, expected_result)
# Should this ever fail, it either means that the schema has changed,
# or the pool is broken.
# If the schema has changed: Update the associated infoDomain.xml file
self.assertEqual(result.__dict__, expected_result)
# The number of open pools should match the number of requested ones.
# If it is 0, then they failed to open
self.assertEqual(len(registry._pool.conn), self.pool_options["size"])
# The number of open pools should match the number of requested ones.
# If it is 0, then they failed to open
self.assertEqual(len(registry._pool.conn), self.pool_options["size"])
# Kill the connection pool
registry.kill_pool()
@patch.object(EPPLibWrapper, "_test_registry_connection_success", patch_success)
def test_pool_restarts_on_send(self):
@ -198,51 +201,63 @@ class TestConnectionPool(TestCase):
xml = (location).read_bytes()
return xml
def do_nothing(command):
pass
# Mock what happens inside the "with"
with ExitStack() as stack:
stack.enter_context(patch.object(EPPConnectionPool, "_create_socket", self.fake_socket))
stack.enter_context(patch.object(Socket, "connect", self.fake_client))
stack.enter_context(patch.object(EPPConnectionPool, "kill_all_connections", do_nothing))
stack.enter_context(patch.object(SocketTransport, "send", self.fake_send))
stack.enter_context(patch.object(SocketTransport, "receive", fake_receive))
# Kill the connection pool
registry.kill_pool()
with less_console_noise():
# Start the connection pool
registry.start_connection_pool()
# Kill the connection pool
registry.kill_pool()
self.assertEqual(registry.pool_status.connection_success, False)
self.assertEqual(registry.pool_status.pool_running, False)
self.assertEqual(registry.pool_status.pool_running, False)
# An exception should be raised as end user will be informed
# that they cannot connect to EPP
with self.assertRaises(RegistryError):
expected = "InfoDomain failed to execute due to a connection error."
# An exception should be raised as end user will be informed
# that they cannot connect to EPP
with self.assertRaises(RegistryError):
expected = "InfoDomain failed to execute due to a connection error."
result = registry.send(commands.InfoDomain(name="test.gov"), cleaned=True)
self.assertEqual(result, expected)
# A subsequent command should be successful, as the pool restarts
result = registry.send(commands.InfoDomain(name="test.gov"), cleaned=True)
self.assertEqual(result, expected)
# Should this ever fail, it either means that the schema has changed,
# or the pool is broken.
# If the schema has changed: Update the associated infoDomain.xml file
self.assertEqual(result.__dict__, expected_result)
# A subsequent command should be successful, as the pool restarts
result = registry.send(commands.InfoDomain(name="test.gov"), cleaned=True)
# Should this ever fail, it either means that the schema has changed,
# or the pool is broken.
# If the schema has changed: Update the associated infoDomain.xml file
self.assertEqual(result.__dict__, expected_result)
# The number of open pools should match the number of requested ones.
# If it is 0, then they failed to open
self.assertEqual(len(registry._pool.conn), self.pool_options["size"])
# The number of open pools should match the number of requested ones.
# If it is 0, then they failed to open
self.assertEqual(len(registry._pool.conn), self.pool_options["size"])
# Kill the connection pool
registry.kill_pool()
@patch.object(EPPLibWrapper, "_test_registry_connection_success", patch_success)
def test_raises_connection_error(self):
"""A .send is invoked on the pool, but registry connection is lost
right as we send a command."""
with ExitStack() as stack:
stack.enter_context(patch.object(EPPConnectionPool, "_create_socket", self.fake_socket))
stack.enter_context(patch.object(Socket, "connect", self.fake_client))
with less_console_noise():
# Start the connection pool
registry.start_connection_pool()
# Pool should be running
self.assertEqual(registry.pool_status.connection_success, True)
self.assertEqual(registry.pool_status.pool_running, True)
# Pool should be running
self.assertEqual(registry.pool_status.connection_success, True)
self.assertEqual(registry.pool_status.pool_running, True)
# Try to send a command out - should fail
with self.assertRaises(RegistryError):
expected = "InfoDomain failed to execute due to a connection error."
result = registry.send(commands.InfoDomain(name="test.gov"), cleaned=True)
self.assertEqual(result, expected)
# Try to send a command out - should fail
with self.assertRaises(RegistryError):
expected = "InfoDomain failed to execute due to a connection error."
result = registry.send(commands.InfoDomain(name="test.gov"), cleaned=True)
self.assertEqual(result, expected)

13
src/epplibwrapper/utility/pool.py
View file

@ -85,6 +85,19 @@ class EPPConnectionPool(ConnectionPool):
logger.error(message, exc_info=True)
raise PoolError(code=PoolErrorCodes.KEEP_ALIVE_FAILED) from err
def _keepalive_periodic(self):
delay = float(self.keepalive) / self.size
while 1:
try:
with self.get() as c:
self._keepalive(c)
except PoolError as err:
logger.error(err.message, exc_info=True)
except self.exc_classes:
# Nothing to do, the pool will generate a new connection later
pass
gevent.sleep(delay)
def _create_socket(self, client, login) -> Socket:
"""Creates and returns a socket instance"""
socket = Socket(client, login)