format changes for satisfying lint, ignore types mainly

This commit is contained in:
David Kennedy 2024-03-01 22:25:58 -05:00
parent 99d34682b1
commit 78c4bb4a8d
No known key found for this signature in database
GPG key ID: 6528A5386E66B96B
2 changed files with 24 additions and 26 deletions

View file

@ -42,7 +42,7 @@ class EPPLibWrapper:
# set _client to None initially. In the event that the __init__ fails # set _client to None initially. In the event that the __init__ fails
# before _client initializes, app should still start and be in a state # before _client initializes, app should still start and be in a state
# that it can attempt _client initialization on send attempts # that it can attempt _client initialization on send attempts
self._client = None self._client = None # type: ignore
# prepare (but do not send) a Login command # prepare (but do not send) a Login command
self._login = commands.Login( self._login = commands.Login(
cl_id=settings.SECRET_REGISTRY_CL_ID, cl_id=settings.SECRET_REGISTRY_CL_ID,
@ -62,7 +62,7 @@ class EPPLibWrapper:
client. Raises errors if initialization fails. client. Raises errors if initialization fails.
This method will be called at app initialization, and also during retries.""" This method will be called at app initialization, and also during retries."""
# establish a client object with a TCP socket transport # establish a client object with a TCP socket transport
self._client = Client( self._client = Client( # type: ignore
SocketTransport( SocketTransport(
settings.SECRET_REGISTRY_HOSTNAME, settings.SECRET_REGISTRY_HOSTNAME,
cert_file=CERT.filename, cert_file=CERT.filename,
@ -72,10 +72,10 @@ class EPPLibWrapper:
) )
try: try:
# use the _client object to connect # use the _client object to connect
self._client.connect() self._client.connect() # type: ignore
response = self._client.send(self._login) response = self._client.send(self._login) # type:ignore
if response.code >= 2000: # type: ignore if response.code >= 2000: # type: ignore
self._client.close() self._client.close() # type:ignore
raise LoginError(response.msg) # type: ignore raise LoginError(response.msg) # type: ignore
except TransportError as err: except TransportError as err:
message = "_initialize_client failed to execute due to a connection error." message = "_initialize_client failed to execute due to a connection error."
@ -91,8 +91,8 @@ class EPPLibWrapper:
def _disconnect(self) -> None: def _disconnect(self) -> None:
"""Close the connection.""" """Close the connection."""
try: try:
self._client.send(commands.Logout()) self._client.send(commands.Logout()) # type: ignore
self._client.close() self._client.close() # type: ignore
except Exception: except Exception:
logger.warning("Connection to registry was not cleanly closed.") logger.warning("Connection to registry was not cleanly closed.")

View file

@ -6,11 +6,7 @@ from .common import less_console_noise
import logging import logging
try: try:
from epplib import commands
from epplib.client import Client
from epplib.exceptions import TransportError from epplib.exceptions import TransportError
from epplib.transport import SocketTransport
from epplib.models import common, info
from epplib.responses import Result from epplib.responses import Result
except ImportError: except ImportError:
pass pass
@ -23,13 +19,7 @@ class TestClient(TestCase):
def fake_result(self, code, msg): def fake_result(self, code, msg):
"""Helper function to create a fake Result object""" """Helper function to create a fake Result object"""
return Result( return Result(code=code, msg=msg, res_data=[], cl_tr_id="cl_tr_id", sv_tr_id="sv_tr_id")
code=code,
msg=msg,
res_data=[],
cl_tr_id="cl_tr_id",
sv_tr_id="sv_tr_id"
)
@patch("epplibwrapper.client.Client") @patch("epplibwrapper.client.Client")
def test_initialize_client_success(self, mock_client): def test_initialize_client_success(self, mock_client):
@ -141,6 +131,7 @@ class TestClient(TestCase):
# first connect() should raise an Exception # first connect() should raise an Exception
# subsequent connect() calls should return success # subsequent connect() calls should return success
connect_call_count = 0 connect_call_count = 0
def connect_side_effect(*args, **kwargs): def connect_side_effect(*args, **kwargs):
nonlocal connect_call_count nonlocal connect_call_count
connect_call_count += 1 connect_call_count += 1
@ -148,12 +139,14 @@ class TestClient(TestCase):
raise Exception("Connection failed") raise Exception("Connection failed")
else: else:
return command_success_result return command_success_result
mock_connect = MagicMock(side_effect=connect_side_effect) mock_connect = MagicMock(side_effect=connect_side_effect)
mock_client.return_value.connect = mock_connect mock_client.return_value.connect = mock_connect
# side_effect for the send() calls # side_effect for the send() calls
# first send will be the send("InfoDomainCommand") and should fail # first send will be the send("InfoDomainCommand") and should fail
# subsequend send() calls should return success # subsequend send() calls should return success
send_call_count = 0 send_call_count = 0
def send_side_effect(*args, **kwargs): def send_side_effect(*args, **kwargs):
nonlocal send_call_count nonlocal send_call_count
send_call_count += 1 send_call_count += 1
@ -161,6 +154,7 @@ class TestClient(TestCase):
return command_failure_result return command_failure_result
else: else:
return command_success_result return command_success_result
mock_send = MagicMock(side_effect=send_side_effect) mock_send = MagicMock(side_effect=send_side_effect)
mock_client.return_value.send = mock_send mock_client.return_value.send = mock_send
# Create EPPLibWrapper instance and call send command # Create EPPLibWrapper instance and call send command
@ -168,12 +162,12 @@ class TestClient(TestCase):
wrapper.send("InfoDomainCommand", cleaned=True) wrapper.send("InfoDomainCommand", cleaned=True)
# two connect() calls should be made, the initial failed connect() # two connect() calls should be made, the initial failed connect()
# and the successful connect() during retry() # and the successful connect() during retry()
self.assertEquals(mock_connect.call_count,2) self.assertEquals(mock_connect.call_count, 2)
# close() should only be called once, during retry() # close() should only be called once, during retry()
mock_close.assert_called_once() mock_close.assert_called_once()
# send called 4 times: failed send("InfoDomainCommand"), passed send(logout), # send called 4 times: failed send("InfoDomainCommand"), passed send(logout),
# passed send(login), passed send("InfoDomainCommand") # passed send(login), passed send("InfoDomainCommand")
self.assertEquals(mock_send.call_count,4) self.assertEquals(mock_send.call_count, 4)
@patch("epplibwrapper.client.Client") @patch("epplibwrapper.client.Client")
def test_send_command_failed_retries_and_fails_again(self, mock_client): def test_send_command_failed_retries_and_fails_again(self, mock_client):
@ -192,6 +186,7 @@ class TestClient(TestCase):
# Create a mock Result instance # Create a mock Result instance
send_command_success_result = self.fake_result(1000, "Command completed successfully") send_command_success_result = self.fake_result(1000, "Command completed successfully")
send_command_failure_result = self.fake_result(2400, "Command failed") send_command_failure_result = self.fake_result(2400, "Command failed")
# side_effect for send command, passes for all other sends (login, logout), but # side_effect for send command, passes for all other sends (login, logout), but
# fails for send("InfoDomainCommand") # fails for send("InfoDomainCommand")
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
@ -199,6 +194,7 @@ class TestClient(TestCase):
return send_command_failure_result return send_command_failure_result
else: else:
return send_command_success_result return send_command_success_result
mock_send = MagicMock(side_effect=side_effect) mock_send = MagicMock(side_effect=side_effect)
mock_client.return_value.connect = mock_connect mock_client.return_value.connect = mock_connect
mock_client.return_value.close = mock_close mock_client.return_value.close = mock_close
@ -211,12 +207,12 @@ class TestClient(TestCase):
wrapper.send("InfoDomainCommand", cleaned=True) wrapper.send("InfoDomainCommand", cleaned=True)
# connect() should be called twice, once during initialization, second time # connect() should be called twice, once during initialization, second time
# during retry # during retry
self.assertEquals(mock_connect.call_count,2) self.assertEquals(mock_connect.call_count, 2)
# close() is called once during retry # close() is called once during retry
mock_close.assert_called_once() mock_close.assert_called_once()
# send() is called 5 times: send(login), send(command) fails, send(logout) # send() is called 5 times: send(login), send(command) fails, send(logout)
# send(login), send(command) # send(login), send(command)
self.assertEquals(mock_send.call_count,5) self.assertEquals(mock_send.call_count, 5)
@patch("epplibwrapper.client.Client") @patch("epplibwrapper.client.Client")
def test_send_command_failure_prompts_successful_retry(self, mock_client): def test_send_command_failure_prompts_successful_retry(self, mock_client):
@ -237,6 +233,7 @@ class TestClient(TestCase):
# side_effect for send call, initial send(login) succeeds during initialization, next send(command) # side_effect for send call, initial send(login) succeeds during initialization, next send(command)
# fails, subsequent sends (logout, login, command) all succeed # fails, subsequent sends (logout, login, command) all succeed
send_call_count = 0 send_call_count = 0
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
nonlocal send_call_count nonlocal send_call_count
send_call_count += 1 send_call_count += 1
@ -244,6 +241,7 @@ class TestClient(TestCase):
return send_command_failure_result return send_command_failure_result
else: else:
return send_command_success_result return send_command_success_result
mock_send = MagicMock(side_effect=side_effect) mock_send = MagicMock(side_effect=side_effect)
mock_client.return_value.connect = mock_connect mock_client.return_value.connect = mock_connect
mock_client.return_value.close = mock_close mock_client.return_value.close = mock_close
@ -252,8 +250,8 @@ class TestClient(TestCase):
wrapper = EPPLibWrapper() wrapper = EPPLibWrapper()
wrapper.send("InfoDomainCommand", cleaned=True) wrapper.send("InfoDomainCommand", cleaned=True)
# connect() is called twice, once during initialization of app, once during retry # connect() is called twice, once during initialization of app, once during retry
self.assertEquals(mock_connect.call_count,2) self.assertEquals(mock_connect.call_count, 2)
# close() is called once, during retry # close() is called once, during retry
mock_close.assert_called_once() mock_close.assert_called_once()
# send() is called 5 times: send(login), send(command) fail, send(logout), send(login), send(command) # send() is called 5 times: send(login), send(command) fail, send(logout), send(login), send(command)
self.assertEquals(mock_send.call_count,5) self.assertEquals(mock_send.call_count, 5)