[APP-5754] Python GetFragmentHistory SDK changes (viamrobotics#701)

src/viam/app/app_client.py

@@ -12,6 +12,7 @@
     AddRoleRequest,
     APIKeyWithAuthorizations,
     AppServiceStub,
+    AuthenticatorInfo,
     Authorization,
     AuthorizedPermissions,
     ChangeRoleRequest,
@@ -49,8 +50,11 @@
     DeleteRobotRequest,
 )
 from viam.proto.app import Fragment as FragmentPB
+from viam.proto.app import FragmentHistoryEntry as FragmentHistoryEntryPB
 from viam.proto.app import FragmentVisibility as FragmentVisibilityPB
 from viam.proto.app import (
+    GetFragmentHistoryRequest,
+    GetFragmentHistoryResponse,
     GetFragmentRequest,
     GetFragmentResponse,
     GetLocationRequest,
@@ -383,6 +387,44 @@ def proto(self) -> FragmentPB:
         )
 
 
+class FragmentHistoryEntry:
+    """A class that mirrors the `FragmentHistoryEntry` proto message.
+
+    Use this class to make the attributes of a `viam.proto.app.FragmentHistoryEntry` more accessible and easier to read/interpret.
+    """
+
+    @classmethod
+    def from_proto(cls, fragment_history_entry: FragmentHistoryEntryPB) -> Self:
+        """Create a `FragmentHistoryEntry` from the .proto defined `FragmentHistoryEntry`.
+
+        Args:
+            fragment_history_entry (viam.proto.app.FragmentHistoryEntry): The object to copy from.
+
+        Returns:
+            FragmentHistoryEntry: The `FragmentHistoryEntry`.
+        """
+        self = cls()
+        self.fragment = fragment_history_entry.fragment
+        self.edited_on = fragment_history_entry.edited_on.ToDatetime()
+        self.old = Fragment.from_proto(fragment_history_entry.old)
+        self.edited_by = fragment_history_entry.edited_by
+        return self
+
+    fragment: str
+    edited_on: datetime
+    old: Fragment
+    edited_by: AuthenticatorInfo
+
+    @property
+    def proto(self) -> FragmentHistoryEntryPB:
+        return FragmentHistoryEntryPB(
+            fragment=self.fragment,
+            edited_on=datetime_to_timestamp(self.edited_on),
+            edited_by=self.edited_by,
+            old=self.old.proto if self.old else None,
+        )
+
+
 class RobotPartHistoryEntry:
     """A class that mirrors the `RobotPartHistoryEntry` proto message.
 
@@ -1781,6 +1823,38 @@ async def delete_fragment(self, fragment_id: str) -> None:
         request = DeleteFragmentRequest(id=fragment_id)
         await self._app_client.DeleteFragment(request, metadata=self._metadata)
 
+    async def get_fragment_history(
+        self, id: str, page_token: Optional[str] = "", page_limit: Optional[int] = 10
+    ) -> List[FragmentHistoryEntry]:
+        """Get fragment history.
+
+        ::
+
+            fragment_history = await cloud.get_fragment_history(
+                id = "12a12ab1-1234-5678-abcd-abcd01234567",
+                page_token = "pg-token",
+                page_limit = 10
+            )
+
+        Args:
+            id (str): ID of the fragment to fetch history for.
+            page_token (Optional[str]): the page token for the fragment history collection
+            page_limit (Optional[int]): the number of fragment history documents to return in the result.
+                The default page limit is 10.
+
+        Raises:
+            GRPCError: if an invalid fragment id, page token or page limit is passed.
+
+        Returns:
+            viam.app.app_client.FragmentHistoryResponse: The fragment history document(s).
+
+        For more information, see `Fleet Management API <https://docs.viam.com/appendix/apis/fleet/>`_.
+        """
+
+        request = GetFragmentHistoryRequest(id=id, page_token=page_token, page_limit=page_limit)
+        response: GetFragmentHistoryResponse = await self._app_client.GetFragmentHistory(request, metadata=self._metadata)
+        return [FragmentHistoryEntry.from_proto(fragment_history) for fragment_history in response.history]
+
     async def add_role(
         self,
         org_id: str,

tests/mocks/services.py

@@ -5,6 +5,7 @@
 from numpy.typing import NDArray
 
 from viam.app.data_client import DataClient
+from viam.gen.app.v1.app_pb2 import FragmentHistoryEntry, GetFragmentHistoryRequest, GetFragmentHistoryResponse
 from viam.media.video import ViamImage
 from viam.proto.app import (
     AddRoleRequest,
@@ -1173,6 +1174,7 @@ def __init__(
         available: bool,
         location_auth: LocationAuth,
         robot_part_history: List[RobotPartHistoryEntry],
+        fragment_history: List[FragmentHistoryEntry],
         authorizations: List[Authorization],
         url: str,
         module: Module,
@@ -1195,6 +1197,7 @@ def __init__(
         self.available = available
         self.location_auth = location_auth
         self.robot_part_history = robot_part_history
+        self.fragment_history = fragment_history
         self.authorizations = authorizations
         self.url = url
         self.module = module
@@ -1490,6 +1493,14 @@ async def GetFragment(self, stream: Stream[GetFragmentRequest, GetFragmentRespon
         self.fragment_id = request.id
         await stream.send_message(GetFragmentResponse(fragment=self.fragment))
 
+    async def GetFragmentHistory(self, stream: Stream[GetFragmentHistoryRequest, GetFragmentHistoryResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        self.id = request.id
+        self.page_token = request.page_token
+        self.page_limit = request.page_limit
+        await stream.send_message(GetFragmentHistoryResponse(history=self.fragment_history))
+
     async def CreateFragment(self, stream: Stream[CreateFragmentRequest, CreateFragmentResponse]) -> None:
         request = await stream.recv_message()
         assert request is not None

tests/test_app_client.py

@@ -4,9 +4,10 @@
 from grpclib.testing import ChannelFor
 
 from viam.app.app_client import APIKeyAuthorization, AppClient, Fragment, FragmentVisibilityPB
-from viam.proto.app import APIKey, APIKeyWithAuthorizations, Authorization, AuthorizationDetails, AuthorizedPermissions
+from viam.proto.app import APIKey, APIKeyWithAuthorizations, AuthenticatorInfo, Authorization, AuthorizationDetails, AuthorizedPermissions
 from viam.proto.app import Fragment as FragmentPB
 from viam.proto.app import (
+    FragmentHistoryEntry,
     Location,
     LocationAuth,
     Model,
@@ -36,6 +37,8 @@
 IDS = [ID]
 NAME = "name"
 CID = "cid"
+PAGE_TOKEN = "123"
+PAGE_LIMIT = 20
 TIME = datetime_to_timestamp(datetime.now())
 PUBLIC_NAMESPACE = "public_namespace"
 DEFAULT_REGION = "default_region"
@@ -122,6 +125,9 @@
 PART = "part"
 ROBOT_PART_HISTORY_ENTRY = RobotPartHistoryEntry(part=PART, robot=ID, when=TIME, old=None)
 ROBOT_PART_HISTORY = [ROBOT_PART_HISTORY_ENTRY]
+AUTHENTICATOR_INFO = AuthenticatorInfo(value="value", is_deactivated=True, type=1)
+FRAGMENT_HISTORY_ENTRY = FragmentHistoryEntry(fragment=ID, edited_by=AUTHENTICATOR_INFO, old=FRAGMENT, edited_on=TIME)
+FRAGMENT_HISTORY = [FRAGMENT_HISTORY_ENTRY]
 TYPE = "robot"
 ROLE = "operator"
 API_KEY = "key"
@@ -210,6 +216,7 @@ def service() -> MockApp:
         available=AVAILABLE,
         location_auth=LOCATION_AUTH,
         robot_part_history=ROBOT_PART_HISTORY,
+        fragment_history=FRAGMENT_HISTORY,
         authorizations=AUTHORIZATIONS,
         url=URL,
         module=MODULE,
@@ -630,6 +637,19 @@ async def test_delete_fragment(self, service: MockApp):
             await client.delete_fragment(fragment_id=ID)
             assert service.id == ID
 
+    @pytest.mark.asyncio
+    async def test_get_fragment_history(self, service: MockApp):
+        async with ChannelFor([service]) as channel:
+            client = AppClient(channel, METADATA, ID)
+            fragment_history = await client.get_fragment_history(id=ID, page_token=PAGE_TOKEN, page_limit=PAGE_LIMIT)
+            assert service.fragment.id == ID
+            assert len(fragment_history) == len(FRAGMENT_HISTORY)
+            assert service.id == ID
+            assert service.page_token == PAGE_TOKEN
+            assert service.page_limit == PAGE_LIMIT
+            for i in range(len(FRAGMENT_HISTORY)):
+                assert fragment_history[i].proto == FRAGMENT_HISTORY[i]
+
     @pytest.mark.asyncio
     async def test_add_role(self, service: MockApp):
         async with ChannelFor([service]) as channel: