Fix kill_pool()

Kill pool was not killing instances correctly. This fixes that, and adds an additional test case
This commit is contained in:
zandercymatics 2023-10-19 15:59:28 -06:00
parent e7d73d2254
commit 055942fe3a
No known key found for this signature in database
GPG key ID: FF4636ABEC9682B7
4 changed files with 189 additions and 65 deletions

View file

@ -68,6 +68,18 @@ class TestConnectionPool(TestCase):
)
return mock
def fake_client(mock_client):
pw = "none"
client = Client(
SocketTransport(
"none",
cert_file="path/to/cert_file",
key_file="path/to/key_file",
password=pw,
)
)
return client
@patch.object(EPPLibWrapper, "_test_registry_connection_success", patch_success)
def test_pool_sends_data(self):
"""A .send is invoked on the pool successfully"""
@ -108,18 +120,6 @@ class TestConnectionPool(TestCase):
"sv_tr_id": "wRRNVhKhQW2m6wsUHbo/lA==-29a",
}
def fake_client(mock_client):
pw = "none"
client = Client(
SocketTransport(
"none",
cert_file="path/to/cert_file",
key_file="path/to/key_file",
password=pw,
)
)
return client
# Mock a response from EPP
def fake_receive(command, cleaned=None):
location = Path(__file__).parent / "utility" / "infoDomain.xml"
@ -131,10 +131,10 @@ class TestConnectionPool(TestCase):
stack.enter_context(
patch.object(EPPConnectionPool, "_create_socket", self.fake_socket)
)
stack.enter_context(patch.object(Socket, "connect", fake_client))
stack.enter_context(patch.object(Socket, "connect", self.fake_client))
stack.enter_context(patch.object(SocketTransport, "send", self.fake_send))
stack.enter_context(patch.object(SocketTransport, "receive", fake_receive))
# Restart the connection pool, since it starts on app startup
# 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)
@ -152,29 +152,97 @@ class TestConnectionPool(TestCase):
# If it is 0, then they failed to open
self.assertEqual(len(registry._pool.conn), self.pool_options["size"])
@patch.object(EPPLibWrapper, "_test_registry_connection_success", patch_success)
def test_pool_restarts_on_send(self):
"""A .send is invoked, but the pool isn't running.
The pool should restart."""
expected_result = {
"cl_tr_id": None,
"code": 1000,
"extensions": [],
"msg": "Command completed successfully",
"msg_q": None,
"res_data": [
info.InfoDomainResultData(
roid="DF1340360-GOV",
statuses=[
common.Status(
state="serverTransferProhibited",
description=None,
lang="en",
),
common.Status(state="inactive", description=None, lang="en"),
],
cl_id="gov2023-ote",
cr_id="gov2023-ote",
cr_date=datetime.datetime(
2023, 8, 15, 23, 56, 36, tzinfo=tzlocal()
),
up_id="gov2023-ote",
up_date=datetime.datetime(2023, 8, 17, 2, 3, 19, tzinfo=tzlocal()),
tr_date=None,
name="test3.gov",
registrant="TuaWnx9hnm84GCSU",
admins=[],
nsset=None,
keyset=None,
ex_date=datetime.date(2024, 8, 15),
auth_info=info.DomainAuthInfo(pw="2fooBAR123fooBaz"),
)
],
"sv_tr_id": "wRRNVhKhQW2m6wsUHbo/lA==-29a",
}
# Mock a response from EPP
def fake_receive(command, cleaned=None):
location = Path(__file__).parent / "utility" / "infoDomain.xml"
xml = (location).read_bytes()
return xml
# 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(SocketTransport, "send", self.fake_send))
stack.enter_context(patch.object(SocketTransport, "receive", fake_receive))
# Kill the connection pool
registry.kill_pool()
self.assertEqual(registry.pool_status.connection_success, 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."
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)
# 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"])
@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."""
# Fake data for the _pool object
def fake_client(self):
pw = "none"
client = Client(
SocketTransport(
"none",
cert_file="path/to/cert_file",
key_file="path/to/key_file",
password=pw,
)
)
return client
with ExitStack() as stack:
stack.enter_context(
patch.object(EPPConnectionPool, "_create_socket", self.fake_socket)
)
stack.enter_context(patch.object(Socket, "connect", fake_client))
stack.enter_context(patch.object(Socket, "connect", self.fake_client))
# Pool should be running
self.assertEqual(registry.pool_status.connection_success, True)