Merge pull request #1066 from cisagov/dk/857-epp-client-hold

857 - EPP place and revert client hold
This commit is contained in:
dave-kennedy-ecs 2023-09-26 13:04:40 -04:00 committed by GitHub
commit 9c1e3fa013
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 185 additions and 18 deletions

View file

@ -67,6 +67,13 @@ class RegistryError(Exception):
def should_retry(self): def should_retry(self):
return self.code == ErrorCode.COMMAND_FAILED return self.code == ErrorCode.COMMAND_FAILED
# connection errors have error code of None and [Errno 99] in the err message
def is_connection_error(self):
return self.code is None
def is_session_error(self):
return self.code is not None and (self.code >= 2501 and self.code <= 2502)
def is_server_error(self): def is_server_error(self):
return self.code is not None and (self.code >= 2400 and self.code <= 2500) return self.code is not None and (self.code >= 2400 and self.code <= 2500)

View file

@ -745,7 +745,23 @@ class DomainAdmin(ListHeaderAdmin):
obj.place_client_hold() obj.place_client_hold()
obj.save() obj.save()
except Exception as err: except Exception as err:
self.message_user(request, err, messages.ERROR) # if error is an error from the registry, display useful
# and readable error
if err.code:
self.message_user(
request,
f"Error placing the hold with the registry: {err}",
messages.ERROR,
)
elif err.is_connection_error():
self.message_user(
request,
"Error connecting to the registry",
messages.ERROR,
)
else:
# all other type error messages, display the error
self.message_user(request, err, messages.ERROR)
else: else:
self.message_user( self.message_user(
request, request,
@ -762,7 +778,23 @@ class DomainAdmin(ListHeaderAdmin):
obj.revert_client_hold() obj.revert_client_hold()
obj.save() obj.save()
except Exception as err: except Exception as err:
self.message_user(request, err, messages.ERROR) # if error is an error from the registry, display useful
# and readable error
if err.code:
self.message_user(
request,
f"Error removing the hold in the registry: {err}",
messages.ERROR,
)
elif err.is_connection_error():
self.message_user(
request,
"Error connecting to the registry",
messages.ERROR,
)
else:
# all other type error messages, display the error
self.message_user(request, err, messages.ERROR)
else: else:
self.message_user( self.message_user(
request, request,

View file

@ -634,13 +634,25 @@ class Domain(TimeStampedModel, DomainHelper):
"""This domain should not be active. """This domain should not be active.
may raises RegistryError, should be caught or handled correctly by caller""" may raises RegistryError, should be caught or handled correctly by caller"""
request = commands.UpdateDomain(name=self.name, add=[self.clientHoldStatus()]) request = commands.UpdateDomain(name=self.name, add=[self.clientHoldStatus()])
registry.send(request, cleaned=True) try:
registry.send(request, cleaned=True)
self._invalidate_cache()
except RegistryError as err:
# if registry error occurs, log the error, and raise it as well
logger.error(f"registry error placing client hold: {err}")
raise (err)
def _remove_client_hold(self): def _remove_client_hold(self):
"""This domain is okay to be active. """This domain is okay to be active.
may raises RegistryError, should be caught or handled correctly by caller""" may raises RegistryError, should be caught or handled correctly by caller"""
request = commands.UpdateDomain(name=self.name, rem=[self.clientHoldStatus()]) request = commands.UpdateDomain(name=self.name, rem=[self.clientHoldStatus()])
registry.send(request, cleaned=True) try:
registry.send(request, cleaned=True)
self._invalidate_cache()
except RegistryError as err:
# if registry error occurs, log the error, and raise it as well
logger.error(f"registry error removing client hold: {err}")
raise (err)
def _delete_domain(self): def _delete_domain(self):
"""This domain should be deleted from the registry """This domain should be deleted from the registry
@ -773,7 +785,9 @@ class Domain(TimeStampedModel, DomainHelper):
administrative_contact.domain = self administrative_contact.domain = self
administrative_contact.save() administrative_contact.save()
@transition(field="state", source=State.READY, target=State.ON_HOLD) @transition(
field="state", source=[State.READY, State.ON_HOLD], target=State.ON_HOLD
)
def place_client_hold(self): def place_client_hold(self):
"""place a clienthold on a domain (no longer should resolve)""" """place a clienthold on a domain (no longer should resolve)"""
# TODO - ensure all requirements for client hold are made here # TODO - ensure all requirements for client hold are made here
@ -782,7 +796,7 @@ class Domain(TimeStampedModel, DomainHelper):
self._place_client_hold() self._place_client_hold()
# TODO -on the client hold ticket any additional error handling here # TODO -on the client hold ticket any additional error handling here
@transition(field="state", source=State.ON_HOLD, target=State.READY) @transition(field="state", source=[State.READY, State.ON_HOLD], target=State.READY)
def revert_client_hold(self): def revert_client_hold(self):
"""undo a clienthold placed on a domain""" """undo a clienthold placed on a domain"""

View file

@ -20,6 +20,8 @@ from .common import MockEppLib
from epplibwrapper import ( from epplibwrapper import (
commands, commands,
common, common,
RegistryError,
ErrorCode,
) )
@ -702,7 +704,7 @@ class TestRegistrantDNSSEC(TestCase):
raise raise
class TestAnalystClientHold(TestCase): class TestAnalystClientHold(MockEppLib):
"""Rule: Analysts may suspend or restore a domain by using client hold""" """Rule: Analysts may suspend or restore a domain by using client hold"""
def setUp(self): def setUp(self):
@ -711,18 +713,50 @@ class TestAnalystClientHold(TestCase):
Given the analyst is logged in Given the analyst is logged in
And a domain exists in the registry And a domain exists in the registry
""" """
pass super().setUp()
# for the tests, need a domain in the ready state
self.domain, _ = Domain.objects.get_or_create(
name="fake.gov", state=Domain.State.READY
)
# for the tests, need a domain in the on_hold state
self.domain_on_hold, _ = Domain.objects.get_or_create(
name="fake-on-hold.gov", state=Domain.State.ON_HOLD
)
def tearDown(self):
Domain.objects.all().delete()
super().tearDown()
@skip("not implemented yet")
def test_analyst_places_client_hold(self): def test_analyst_places_client_hold(self):
""" """
Scenario: Analyst takes a domain off the internet Scenario: Analyst takes a domain off the internet
When `domain.place_client_hold()` is called When `domain.place_client_hold()` is called
Then `CLIENT_HOLD` is added to the domain's statuses Then `CLIENT_HOLD` is added to the domain's statuses
""" """
raise self.domain.place_client_hold()
self.mockedSendFunction.assert_has_calls(
[
call(
commands.UpdateDomain(
name="fake.gov",
add=[
common.Status(
state=Domain.Status.CLIENT_HOLD,
description="",
lang="en",
)
],
nsset=None,
keyset=None,
registrant=None,
auth_info=None,
),
cleaned=True,
)
]
)
self.assertEquals(self.domain.state, Domain.State.ON_HOLD)
@skip("not implemented yet")
def test_analyst_places_client_hold_idempotent(self): def test_analyst_places_client_hold_idempotent(self):
""" """
Scenario: Analyst tries to place client hold twice Scenario: Analyst tries to place client hold twice
@ -730,9 +764,30 @@ class TestAnalystClientHold(TestCase):
When `domain.place_client_hold()` is called When `domain.place_client_hold()` is called
Then Domain returns normally (without error) Then Domain returns normally (without error)
""" """
raise self.domain_on_hold.place_client_hold()
self.mockedSendFunction.assert_has_calls(
[
call(
commands.UpdateDomain(
name="fake-on-hold.gov",
add=[
common.Status(
state=Domain.Status.CLIENT_HOLD,
description="",
lang="en",
)
],
nsset=None,
keyset=None,
registrant=None,
auth_info=None,
),
cleaned=True,
)
]
)
self.assertEquals(self.domain_on_hold.state, Domain.State.ON_HOLD)
@skip("not implemented yet")
def test_analyst_removes_client_hold(self): def test_analyst_removes_client_hold(self):
""" """
Scenario: Analyst restores a suspended domain Scenario: Analyst restores a suspended domain
@ -740,9 +795,30 @@ class TestAnalystClientHold(TestCase):
When `domain.remove_client_hold()` is called When `domain.remove_client_hold()` is called
Then `CLIENT_HOLD` is no longer in the domain's statuses Then `CLIENT_HOLD` is no longer in the domain's statuses
""" """
raise self.domain_on_hold.revert_client_hold()
self.mockedSendFunction.assert_has_calls(
[
call(
commands.UpdateDomain(
name="fake-on-hold.gov",
rem=[
common.Status(
state=Domain.Status.CLIENT_HOLD,
description="",
lang="en",
)
],
nsset=None,
keyset=None,
registrant=None,
auth_info=None,
),
cleaned=True,
)
]
)
self.assertEquals(self.domain_on_hold.state, Domain.State.READY)
@skip("not implemented yet")
def test_analyst_removes_client_hold_idempotent(self): def test_analyst_removes_client_hold_idempotent(self):
""" """
Scenario: Analyst tries to remove client hold twice Scenario: Analyst tries to remove client hold twice
@ -750,16 +826,54 @@ class TestAnalystClientHold(TestCase):
When `domain.remove_client_hold()` is called When `domain.remove_client_hold()` is called
Then Domain returns normally (without error) Then Domain returns normally (without error)
""" """
raise self.domain.revert_client_hold()
self.mockedSendFunction.assert_has_calls(
[
call(
commands.UpdateDomain(
name="fake.gov",
rem=[
common.Status(
state=Domain.Status.CLIENT_HOLD,
description="",
lang="en",
)
],
nsset=None,
keyset=None,
registrant=None,
auth_info=None,
),
cleaned=True,
)
]
)
self.assertEquals(self.domain.state, Domain.State.READY)
@skip("not implemented yet")
def test_update_is_unsuccessful(self): def test_update_is_unsuccessful(self):
""" """
Scenario: An update to place or remove client hold is unsuccessful Scenario: An update to place or remove client hold is unsuccessful
When an error is returned from epplibwrapper When an error is returned from epplibwrapper
Then a user-friendly error message is returned for displaying on the web Then a user-friendly error message is returned for displaying on the web
""" """
raise
def side_effect(_request, cleaned):
raise RegistryError(code=ErrorCode.OBJECT_STATUS_PROHIBITS_OPERATION)
patcher = patch("registrar.models.domain.registry.send")
mocked_send = patcher.start()
mocked_send.side_effect = side_effect
# if RegistryError is raised, admin formats user-friendly
# error message if error is_client_error, is_session_error, or
# is_server_error; so test for those conditions
with self.assertRaises(RegistryError) as err:
self.domain.place_client_hold()
self.assertTrue(
err.is_client_error() or err.is_session_error() or err.is_server_error()
)
patcher.stop()
class TestAnalystLock(TestCase): class TestAnalystLock(TestCase):