Fix #92 - Add contact representatives

regal/__init__.py

@@ -30,6 +30,7 @@ from .registrar import (
     RegistrarCertificationId,
     RegistrarClient,
 )
+from .representative import Representative, RepresentativeClient
 from .search import SearchContactClient, SearchDomainClient, SearchKeysetClient, SearchNssetClient, SearchResults
 
 __version__ = "1.1.4"
@@ -60,6 +61,10 @@ __all__ = [
     "RegistrarCertification",
     "RegistrarCertificationId",
     "RegistrarClient",
+    "Representative",
+    "RepresentativeClient",
+    "RepresentativeHistoryId",
+    "RepresentativeId",
     "SearchContactClient",
     "SearchDomainClient",
     "SearchKeysetClient",

regal/exceptions.py

@@ -65,6 +65,14 @@ class EppCredentialsDoesNotExist(RegistryException):
     """Registrar EPP credentials does not exist."""
 
 
+class RepresentativeDoesNotExist(ObjectDoesNotExist):
+    """Contact representative does not exist."""
+
+
+class RepresentativeAlreadyExists(RegistryException):
+    """Contact representative already exists."""
+
+
 class NotCurrentVersion(RegistryException):
     """Contact history identification not valid."""
 

regal/representative.py

+"""Contact representatives and their client API."""
+from typing import Optional, Union, cast
+
+from fred_api.registry.contact_representative import service_contact_representative_grpc_pb2_grpc
+from fred_api.registry.contact_representative.contact_representative_common_types_pb2 import ContactRepresentativeInfo
+from fred_api.registry.contact_representative.service_contact_representative_grpc_pb2 import (
+    ContactRepresentativeInfoReply,
+    ContactRepresentativeInfoRequest,
+    CreateContactRepresentativeReply,
+    CreateContactRepresentativeRequest,
+    DeleteContactRepresentativeReply,
+    DeleteContactRepresentativeRequest,
+    UpdateContactRepresentativeReply,
+    UpdateContactRepresentativeRequest,
+)
+from fred_types import BaseModel, ContactId, LogEntryId, RepresentativeHistoryId, RepresentativeId
+from frgal.aio import AsyncGrpcClient
+
+from .common import Address, RegistryDecoder
+from .exceptions import ContactDoesNotExist, InvalidData, RepresentativeAlreadyExists, RepresentativeDoesNotExist
+
+
+class Representative(BaseModel):
+    """Represents a contact representative."""
+
+    id: RepresentativeId
+    history_id: Optional[RepresentativeHistoryId] = None
+    contact_id: ContactId
+    name: str = ""
+    organization: str = ""
+    place: Optional[Address] = None
+    telephone: str = ""
+    email: str = ""
+
+
+class RepresentativeDecoder(RegistryDecoder):
+    """Contact representative service decoder."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.set_decoder(ContactRepresentativeInfo, self._decode_contact_representative_info)
+
+        self.set_exception_decoder(
+            ContactRepresentativeInfoReply.Exception.ContactRepresentativeDoesNotExist, RepresentativeDoesNotExist
+        )
+        self.set_exception_decoder(
+            CreateContactRepresentativeReply.Exception.ContactRepresentativeAlreadyExists, RepresentativeAlreadyExists
+        )
+        self.set_exception_decoder(CreateContactRepresentativeReply.Exception.ContactDoesNotExist, ContactDoesNotExist)
+        self.set_exception_decoder(CreateContactRepresentativeReply.Exception.InvalidData, InvalidData)
+        self.set_exception_decoder(
+            UpdateContactRepresentativeReply.Exception.ContactRepresentativeDoesNotExist, RepresentativeDoesNotExist
+        )
+        self.set_exception_decoder(UpdateContactRepresentativeReply.Exception.InvalidData, InvalidData)
+        self.set_exception_decoder(
+            DeleteContactRepresentativeReply.Exception.ContactRepresentativeDoesNotExist, RepresentativeDoesNotExist
+        )
+
+    def _decode_contact_representative_info(self, value: ContactRepresentativeInfo) -> Representative:
+        """Decode ContactRepresentativeInfo."""
+        data = self._decode_message(value)
+        data["telephone"] = data["telephone"] or ""
+        data["email"] = data["email"] or ""
+        return Representative(**data)
+
+
+class RepresentativeClient(AsyncGrpcClient):
+    """Contact representative service gRPC client."""
+
+    service = "ContactRepresentative"
+    decoder_cls = RepresentativeDecoder
+    grpc_modules = (service_contact_representative_grpc_pb2_grpc,)
+
+    async def get(
+        self,
+        value: Union[RepresentativeId, RepresentativeHistoryId, ContactId],
+        /,
+    ) -> Representative:
+        """Return contact representative based on its identifier.
+
+        Args:
+            value: Contact representative ID, contact representative history ID or contact ID.
+
+        Returns:
+            Contact representative info.
+
+        Raises:
+            RepresentativeDoesNotExist: Identifier doesn't reference any known representative.
+        """
+        request = ContactRepresentativeInfoRequest()
+        if isinstance(value, ContactId):
+            request.contact_id.uuid.value = value
+        elif isinstance(value, RepresentativeHistoryId):
+            request.history_id.value = value
+        else:
+            request.id.value = value
+        return cast(Representative, await self.call(self.service, "get_contact_representative_info", request))
+
+    async def create(
+        self, representative: Representative, *, log_entry_id: Optional[LogEntryId] = None
+    ) -> Representative:
+        """Create and return contact representative.
+
+        Args:
+            representative: Contact representative.
+            log_entry_id: An identifier of the related log entry.
+
+        Returns:
+            Contact representative info.
+
+        Raises:
+            ContactDoesNotExist: Referenced contact does not exist.
+            RepresentativeAlreadyExists: Contact representative already exists.
+            InvalidData: Contact representative data are invalid.
+        """
+        request = CreateContactRepresentativeRequest()
+        request.contact_id.uuid.value = representative.contact_id
+        request.name = representative.name
+        request.organization = representative.organization
+        if representative.place:
+            request.place.street.extend(representative.place.street)
+            request.place.city = representative.place.city
+            request.place.state_or_province = representative.place.state_or_province
+            request.place.postal_code.value = representative.place.postal_code
+            request.place.country_code.value = representative.place.country_code
+        if representative.telephone:
+            request.telephone.value = representative.telephone
+        if representative.email:
+            request.email.value = representative.email
+        if log_entry_id:
+            request.log_entry_id.value = log_entry_id
+        return cast(Representative, await self.call(self.service, "create_contact_representative", request))
+
+    async def update(
+        self, representative: Representative, *, log_entry_id: Optional[LogEntryId] = None
+    ) -> Representative:
+        """Update and return contact representative.
+
+        Args:
+            representative: Contact representative.
+            log_entry_id: An identifier of the related log entry.
+
+        Returns:
+            Contact representative info.
+
+        Raises:
+            RepresentativeDoesNotExist: Contact representative does not exist.
+            InvalidData: Contact representative data are invalid.
+        """
+        request = UpdateContactRepresentativeRequest()
+        request.id.value = representative.id
+        request.name = representative.name
+        request.organization = representative.organization
+        if representative.place:
+            request.place.street.extend(representative.place.street)
+            request.place.city = representative.place.city
+            request.place.state_or_province = representative.place.state_or_province
+            request.place.postal_code.value = representative.place.postal_code
+            request.place.country_code.value = representative.place.country_code
+        if representative.telephone:
+            request.telephone.value = representative.telephone
+        if representative.email:
+            request.email.value = representative.email
+        if log_entry_id:
+            request.log_entry_id.value = log_entry_id
+        return cast(Representative, await self.call(self.service, "update_contact_representative", request))
+
+    async def delete(self, id: RepresentativeId, *, log_entry_id: Optional[LogEntryId] = None) -> None:
+        """Delete contact representative.
+
+        Args:
+            id: Contact representative ID.
+            log_entry_id: An identifier of the related log entry.
+
+        Raises:
+            RepresentativeDoesNotExist: Contact representative does not exist.
+        """
+        request = DeleteContactRepresentativeRequest()
+        request.id.value = id
+        if log_entry_id:
+            request.log_entry_id.value = log_entry_id
+        await self.call(self.service, "delete_contact_representative", request)

regal/tests/test_representative.py

+from typing import Any
+from unittest import IsolatedAsyncioTestCase
+from unittest.mock import call, sentinel
+
+from fred_api.registry.contact_representative.contact_representative_common_types_pb2 import ContactRepresentativeInfo
+from fred_api.registry.contact_representative.service_contact_representative_grpc_pb2 import (
+    ContactRepresentativeInfoReply,
+    ContactRepresentativeInfoRequest,
+    CreateContactRepresentativeReply,
+    CreateContactRepresentativeRequest,
+    DeleteContactRepresentativeReply,
+    DeleteContactRepresentativeRequest,
+    UpdateContactRepresentativeReply,
+    UpdateContactRepresentativeRequest,
+)
+from fred_types import ContactId, LogEntryId
+from frgal.utils import AsyncTestClientMixin, make_awaitable
+
+from regal.common import Address
+from regal.representative import (
+    Representative,
+    RepresentativeClient,
+    RepresentativeDecoder,
+    RepresentativeHistoryId,
+    RepresentativeId,
+)
+
+
+class RepresentativeDecoderTest(IsolatedAsyncioTestCase):
+    """Test RepresentativeDecoder."""
+
+    def setUp(self):
+        self.decoder = RepresentativeDecoder()
+
+    def test_decode_info(self):
+        info_1 = ContactRepresentativeInfo()
+        info_1.id.value = "RIMMER"
+        info_1.contact_id.uuid.value = "SMEGHEAD"
+
+        info_2 = ContactRepresentativeInfo()
+        info_2.id.value = "RIMMER"
+        info_2.contact_id.uuid.value = "SMEGHEAD"
+        info_2.history_id.value = "ACE"
+
+        info_3 = ContactRepresentativeInfo(name="Arnold Rimmer")
+        info_3.id.value = "RIMMER"
+        info_3.contact_id.uuid.value = "SMEGHEAD"
+
+        info_4 = ContactRepresentativeInfo()
+        info_4.id.value = "RIMMER"
+        info_4.contact_id.uuid.value = "SMEGHEAD"
+        info_4.name = "Arnold Rimmer"
+        info_4.organization = "Jupiter Mining Corporation"
+        info_4.place.street.append("Deck 16")
+        info_4.place.city = "Red Dwarf"
+        info_4.place.state_or_province = "Deep Space"
+        info_4.place.postal_code.value = "JMC"
+        info_4.place.country_code.value = "JU"
+        info_4.telephone.value = "Chief!"
+        info_4.email.value = "arnold@example.org"
+        place = Address(
+            street=["Deck 16"], city="Red Dwarf", state_or_province="Deep Space", postal_code="JMC", country_code="JU"
+        )
+        result_4 = Representative(
+            id=RepresentativeId("RIMMER"),
+            contact_id=ContactId("SMEGHEAD"),
+            name="Arnold Rimmer",
+            organization="Jupiter Mining Corporation",
+            place=place,
+            telephone="Chief!",
+            email="arnold@example.org",
+        )
+        data = (
+            # info, result
+            (info_1, Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))),
+            (
+                info_2,
+                Representative(
+                    id=RepresentativeId("RIMMER"),
+                    history_id=RepresentativeHistoryId("ACE"),
+                    contact_id=ContactId("SMEGHEAD"),
+                ),
+            ),
+            (
+                info_3,
+                Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"), name="Arnold Rimmer"),
+            ),
+            (info_4, result_4),
+        )
+        for info, result in data:
+            with self.subTest(info=info):
+                self.assertEqual(self.decoder.decode(info), result)
+
+
+class TestRepresentativeClient(AsyncTestClientMixin, RepresentativeClient):
+    """Test RepresentativeClient."""
+
+
+class RegistrarClientTest(IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.client = TestRepresentativeClient(netloc=sentinel.netloc)
+
+    async def _test_get(
+        self,
+        value: Any,
+        request: ContactRepresentativeInfoRequest,
+    ) -> None:
+        reply = ContactRepresentativeInfoReply()
+        reply.data.contact_representative.id.value = "RIMMER"
+        reply.data.contact_representative.contact_id.uuid.value = "SMEGHEAD"
+        self.client.mock.return_value = make_awaitable(reply)
+        result = Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))
+
+        self.assertEqual(await self.client.get(value), result)
+
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/get_contact_representative_info",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_get_plain(self):
+        request = ContactRepresentativeInfoRequest()
+        request.id.value = "RIMMER"
+        await self._test_get("RIMMER", request)
+
+    async def test_get_id(self):
+        request = ContactRepresentativeInfoRequest()
+        request.id.value = "RIMMER"
+        await self._test_get(RepresentativeId("RIMMER"), request)
+
+    async def test_get_history_id(self):
+        request = ContactRepresentativeInfoRequest()
+        request.history_id.value = "ACE"
+        await self._test_get(RepresentativeHistoryId("ACE"), request)
+
+    async def test_get_contact_id(self):
+        request = ContactRepresentativeInfoRequest()
+        request.contact_id.uuid.value = "SMEGHEAD"
+        await self._test_get(ContactId("SMEGHEAD"), request)
+
+    async def test_create_min(self):
+        reply = CreateContactRepresentativeReply()
+        reply.data.contact_representative.id.value = "RIMMER"
+        reply.data.contact_representative.contact_id.uuid.value = "SMEGHEAD"
+        self.client.mock.return_value = make_awaitable(reply)
+        representative = Representative(id=RepresentativeId("PLACEHOLDER"), contact_id=ContactId("SMEGHEAD"))
+        result = Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))
+
+        self.assertEqual(await self.client.create(representative), result)
+
+        request = CreateContactRepresentativeRequest()
+        request.contact_id.uuid.value = "SMEGHEAD"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/create_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_create_full(self):
+        reply = CreateContactRepresentativeReply()
+        reply.data.contact_representative.id.value = "RIMMER"
+        reply.data.contact_representative.contact_id.uuid.value = "SMEGHEAD"
+        self.client.mock.return_value = make_awaitable(reply)
+        place = Address(
+            street=["Deck 16"], city="Red Dwarf", state_or_province="Deep Space", postal_code="JMC", country_code="JU"
+        )
+        representative = Representative(
+            id=RepresentativeId("PLACEHOLDER"),
+            contact_id=ContactId("SMEGHEAD"),
+            name="Arnold Rimmer",
+            organization="Jupiter Mining Corporation",
+            place=place,
+            telephone="Chief!",
+            email="arnold@example.org",
+        )
+        result = Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))
+
+        self.assertEqual(await self.client.create(representative), result)
+
+        request = CreateContactRepresentativeRequest()
+        request.contact_id.uuid.value = "SMEGHEAD"
+        request.name = "Arnold Rimmer"
+        request.organization = "Jupiter Mining Corporation"
+        request.place.street.append("Deck 16")
+        request.place.city = "Red Dwarf"
+        request.place.state_or_province = "Deep Space"
+        request.place.postal_code.value = "JMC"
+        request.place.country_code.value = "JU"
+        request.telephone.value = "Chief!"
+        request.email.value = "arnold@example.org"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/create_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_create_log_entry_id(self):
+        reply = CreateContactRepresentativeReply()
+        reply.data.contact_representative.id.value = "RIMMER"
+        reply.data.contact_representative.contact_id.uuid.value = "SMEGHEAD"
+        self.client.mock.return_value = make_awaitable(reply)
+        representative = Representative(
+            id=RepresentativeId("PLACEHOLDER"), contact_id=ContactId("SMEGHEAD"), name="Arnold Rimmer"
+        )
+        result = Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))
+
+        self.assertEqual(await self.client.create(representative, log_entry_id=LogEntryId("QUAGAARS")), result)
+
+        request = CreateContactRepresentativeRequest()
+        request.contact_id.uuid.value = "SMEGHEAD"
+        request.name = "Arnold Rimmer"
+        request.log_entry_id.value = "QUAGAARS"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/create_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_update_min(self):
+        reply = UpdateContactRepresentativeReply()
+        reply.data.contact_representative.id.value = "RIMMER"
+        reply.data.contact_representative.contact_id.uuid.value = "SMEGHEAD"
+        self.client.mock.return_value = make_awaitable(reply)
+        representative = Representative(id=RepresentativeId("SMEGHEAD"), contact_id=ContactId("PLACEHOLDER"))
+        result = Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))
+
+        self.assertEqual(await self.client.update(representative), result)
+
+        request = UpdateContactRepresentativeRequest()
+        request.id.value = "SMEGHEAD"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/update_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_update_full(self):
+        reply = UpdateContactRepresentativeReply()
+        reply.data.contact_representative.id.value = "RIMMER"
+        reply.data.contact_representative.contact_id.uuid.value = "SMEGHEAD"
+        self.client.mock.return_value = make_awaitable(reply)
+        place = Address(
+            street=["Deck 16"], city="Red Dwarf", state_or_province="Deep Space", postal_code="JMC", country_code="JU"
+        )
+        representative = Representative(
+            id=RepresentativeId("SMEGHEAD"),
+            contact_id=ContactId("PLACEHOLDER"),
+            name="Arnold Rimmer",
+            organization="Jupiter Mining Corporation",
+            place=place,
+            telephone="Chief!",
+            email="arnold@example.org",
+        )
+        result = Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))
+
+        self.assertEqual(await self.client.update(representative), result)
+
+        request = UpdateContactRepresentativeRequest()
+        request.id.value = "SMEGHEAD"
+        request.name = "Arnold Rimmer"
+        request.organization = "Jupiter Mining Corporation"
+        request.place.street.append("Deck 16")
+        request.place.city = "Red Dwarf"
+        request.place.state_or_province = "Deep Space"
+        request.place.postal_code.value = "JMC"
+        request.place.country_code.value = "JU"
+        request.telephone.value = "Chief!"
+        request.email.value = "arnold@example.org"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/update_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_update_log_entry_id(self):
+        reply = UpdateContactRepresentativeReply()
+        reply.data.contact_representative.id.value = "RIMMER"
+        reply.data.contact_representative.contact_id.uuid.value = "SMEGHEAD"
+        self.client.mock.return_value = make_awaitable(reply)
+        representative = Representative(
+            id=RepresentativeId("SMEGHEAD"), contact_id=ContactId("PLACEHOLDER"), name="Arnold Rimmer"
+        )
+        result = Representative(id=RepresentativeId("RIMMER"), contact_id=ContactId("SMEGHEAD"))
+
+        self.assertEqual(await self.client.update(representative, log_entry_id=LogEntryId("QUAGAARS")), result)
+
+        request = UpdateContactRepresentativeRequest()
+        request.id.value = "SMEGHEAD"
+        request.name = "Arnold Rimmer"
+        request.log_entry_id.value = "QUAGAARS"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/update_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_delete(self):
+        self.client.mock.return_value = make_awaitable(DeleteContactRepresentativeReply())
+
+        await self.client.delete(RepresentativeId("SMEGHEAD"))
+
+        request = DeleteContactRepresentativeRequest()
+        request.id.value = "SMEGHEAD"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/delete_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])
+
+    async def test_delete_log_entry_id(self):
+        self.client.mock.return_value = make_awaitable(DeleteContactRepresentativeReply())
+
+        await self.client.delete(RepresentativeId("SMEGHEAD"), log_entry_id=LogEntryId("QUAGAARS"))
+
+        request = DeleteContactRepresentativeRequest()
+        request.id.value = "SMEGHEAD"
+        request.log_entry_id.value = "QUAGAARS"
+        a_call = call(
+            request,
+            method="/Fred.Registry.Api.ContactRepresentative.ContactRepresentative/delete_contact_representative",
+            timeout=None,
+        )
+        self.assertEqual(self.client.mock.mock_calls, [a_call])

setup.cfg

@@ -30,9 +30,9 @@ python_requires = ~=3.8
 install_requires =
     aioitertools >= 0.10, < 0.12
     cryptography >= 38
-    fred-api-registry ~= 5.3.0
+    fred-api-registry ~= 5.4.0
     frgal ~= 3.12
-    fred-types ~=1.0
+    fred-types ~=1.1
     protobuf
     pydantic ~= 1.9
     # NotRequired added in python 3.11