cloudvision/cvlibs: add methods for interface tag operations

Change-Id: I72f91778731a8ba330f0ede0d66362b4b7706269

cloudvision/cvlib/context.py

@@ -26,7 +26,7 @@
     STUDIO_IDS_ARG,
     WORKSPACE_ID_ARG,
 )
-from .device import Device
+from .device import Device, Interface
 from .execution import Execution
 from .exceptions import (
     ConnectionFailed,
@@ -803,3 +803,28 @@ def getDevicesByTag(self, tag: Tag, inTopology: bool = True):
                 elif not inTopology:
                     devices.append(Device(deviceId=devId))
         return devices
+
+    def getInterfacesByTag(self, tag: Tag, inTopology: bool = True):
+        '''
+        Returns list of interfaces that have the user tag assigned to them.
+        If tag.value is unspecified then returns interfaces having that label assigned.
+        By default only interfaces in the topology are returned.
+        '''
+        interfaces = []
+        # Note use list instead of .items()
+        # parallel thread might add/delete tags
+        for devId in list(allTags := self.tags._getAllInterfaceTags()):
+            for intfId in list(devIntfTags := allTags.get(devId, {})):
+                tags = devIntfTags.get(intfId, {})
+                if tags.get(tag.label) and (
+                        not tag.value or tag.value in tags.get(tag.label, [])):
+                    if dev := self.topology._deviceMap.get(devId) if self.topology else None:
+                        if intf := dev.getInterface(intfId):
+                            interfaces.append(intf)
+                        elif not inTopology:
+                            interfaces.append(Interface(name=intfId, device=dev))
+                    elif not inTopology:
+                        interfaces.append(
+                            Interface(name=intfId,
+                                      device=Device(deviceId=devId)))
+        return interfaces

cloudvision/cvlib/device.py

@@ -110,6 +110,25 @@ def _unassignTag(self, ctx, tag: Tag):
         else:
             ctx.tags._unassignDeviceTagLabel(self.id, tag.label)
 
+    def getInterfacesByTag(self, ctx, tag: Tag, inTopology: bool = True):
+        '''
+        Returns list of interfaces that have the user tag assigned to them.
+        If tag.value is unspecified then returns interfaces having that label assigned.
+        By default only interfaces in the topology are returned.
+        '''
+        interfaces = []
+        # Note use list instead of .items()
+        # parallel thread might add/delete tags
+        for intfId in list(devIntfTags := ctx.tags._getAllInterfaceTags().get(self.id, {})):
+            tags = devIntfTags.get(intfId, {})
+            if tags.get(tag.label) and (
+                    not tag.value or tag.value in tags.get(tag.label, [])):
+                if intf := self.getInterface(intfId):
+                    interfaces.append(intf)
+                elif not inTopology:
+                    interfaces.append(Interface(name=intfId, device=self))
+        return interfaces
+
 
 # Interfaces and devices are defined together to avoid circular imports
 class Interface:
@@ -143,3 +162,52 @@ def setPeerInfo(self, device: Device, interface):
 
     def getPeerInfo(self):
         return self._peerDevice, self._peerInterface
+
+    def getSingleTag(self, ctx, label: str, required: bool = True):
+        '''
+        Returns a Tag of the label assigned to the interface.
+        Raises TagTooManyValuesException if there are multiple tags of the label assigned.
+        Raises TagMissingException if required is True and the tag is missing.
+        Returns None if required is False and the tag is missing.
+        '''
+        devName = str(self._device.hostName) if self._device.hostName else str(self._device.id)
+        values = ctx.tags._getInterfaceTags(self._device.id, self.name).get(label)
+        if values and len(values) > 1:
+            raise TagTooManyValuesException(label, devName, values, self.name)
+        if required and not values:
+            raise TagMissingException(label, devName, self.name)
+        return Tag(label, values[0]) if values else None
+
+    def getTags(self, ctx, label: str = None):
+        '''
+        Returns a list of Tags matching the specified label assigned to the interface.
+        If label is unspecified then it returns all Tags assigned to the interface.
+        '''
+        tags: List[Tag] = []
+        if not (ctxTags := ctx.tags._getInterfaceTags(self._device.id, self.name)):
+            return tags
+        # Note use list instead of .items()
+        # parallel thread might add/delete tags
+        for tagLabel in list(ctxTags):
+            if label and label != tagLabel:
+                continue
+            for value in ctxTags.get(tagLabel, []):
+                tags.append(Tag(tagLabel, value))
+        return tags
+
+    def _assignTag(self, ctx, tag: Tag, replaceValue: bool = True):
+        '''
+        Assign a Tag to an interface.
+        If replaceValue is True ensures only one value of label is assigned.
+        '''
+        ctx.tags._assignInterfaceTag(self._device.id, self.name, tag.label, tag.value, replaceValue)
+
+    def _unassignTag(self, ctx, tag: Tag):
+        '''
+        Unassign a Tag from an interface.
+        If tag.value is unspecified unassign all tags of label.
+        '''
+        if tag.value:
+            ctx.tags._unassignInterfaceTag(self._device.id, self.name, tag.label, tag.value)
+        else:
+            ctx.tags._unassignInterfaceTagLabel(self._device.id, self.name, tag.label)

cloudvision/cvlib/exceptions.py

@@ -311,11 +311,14 @@ class TagOperationException(TagErrorException):
     Exception raised when an attempted tag operation is invalid
     """
 
-    def __init__(self, label: str, value: str, operation: str, devId: str = None):
+    def __init__(self, label: str, value: str, operation: str,
+                 devId: str = None, intfId: str = None):
         message = (f"invalid attempt to {operation} tag "
                    f"{self.expTagField(label)}:{self.expTagField(value)}")
         if devId:
             message = message + " for device " + devId
+        if intfId:
+            message = message + " on interface " + intfId
         super().__init__(message)
 
 
@@ -324,9 +327,11 @@ class TagMissingException(TagErrorException):
     Exception raised when a tag is missing from a device
     """
 
-    def __init__(self, label: str, devId: str):
+    def __init__(self, label: str, devId: str, intfId: str = None):
         message = (f"{self.expTagField(label)} tag missing"
                    f" for device {devId}")
+        if intfId:
+            message = message + " on interface " + intfId
         super().__init__(message)
 
 
@@ -335,9 +340,12 @@ class TagTooManyValuesException(TagErrorException):
     Exception raised when a tag has too many values assigned to a device
     """
 
-    def __init__(self, label: str, devId: str, currVals: List[str] = None):
+    def __init__(self, label: str, devId: str, currVals: List[str] = None,
+                 intfId: str = None):
         message = (f"{self.expTagField(label)} tag has too many values"
                    f" assigned to device {devId}")
+        if intfId:
+            message = message + " on interface " + intfId
         if currVals:
             message = message + ", assigned values: " + ", ".join(currVals)
         super().__init__(message)