cloudvision/cvlib: adding IdAllocator class

- this class is already being used in multiple action scripts,
  as well as needed in multiple templates
- instead of duplicating the code in multiple places, adding it to cvlib
- once it's in cvlib the duplicates can be removed and import this from cvlib

Change-Id: If3580e477d0ed7ef85e03660a406e44d348225c2

cloudvision/cvlib/__init__.py

@@ -25,5 +25,6 @@
 from .topology import Connection, Topology
 from .user import User
 from .workspace import Workspace
+from .id_allocator import IdAllocator
 
 __version__ = "1.7.1"

cloudvision/cvlib/id_allocator.py

@@ -0,0 +1,70 @@
+# Copyright (c) 2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the COPYING file.
+
+from typing import Set, Dict, List, Optional
+
+
+class IdAllocator:
+    '''
+    Object to generate unique integer ids, eg. used for generating nodeIds.
+    Can also be used for checking manually entered ids for duplicates.
+    - start:        starting value of id range
+    - end:          ending value of id range
+    The following are only used if checking duplicate id errors:
+    - idNames:      optional name associated with ids
+    - idLabel:      name describing id type
+    - groupLabel:   name describing what is being id'd
+    '''
+    def __init__(self, start: int = 1,
+                 end: int = 65000,
+                 idLabel: str = 'id',
+                 groupLabel: str = 'devices'):
+        self.rangeStart = start
+        self.rangeEnd = end
+        self.available = set(range(start, end + 1))
+        self.allocated: Set[int] = set()
+        self.idNames: Dict[int, Optional[str]] = {}
+        self.idLabel = idLabel
+        self.groupLabel = groupLabel
+
+    def allocate(self, allocId: int = None, name: str = None) -> int:
+        if allocId is not None:
+            if self.rangeStart <= allocId <= self.rangeEnd:
+                if allocId not in self.allocated:
+                    self.allocated.add(allocId)
+                    self.idNames[allocId] = name
+                    self.available.remove(allocId)
+                elif name:
+                    assert name == self.getIdNames().get(allocId), (
+                        f"The same {self.idLabel}, {allocId}, can not be "
+                        f"applied to both of these {self.groupLabel}: "
+                        f"{self.getIdNames().get(allocId)}, "
+                        f"{name}")
+                return allocId
+            raise ValueError(f"Id {allocId} is outside the available range")
+        if not self.available:
+            raise ValueError("no more Ids available")
+        allocatedId = min(self.available)
+        self.available.remove(allocatedId)
+        self.allocated.add(allocatedId)
+        self.idNames[allocatedId] = name
+        return allocatedId
+
+    def free(self, freeId: int):
+        if self.rangeStart <= freeId <= self.rangeEnd:
+            self.available.add(freeId)
+            if freeId in self.allocated:
+                self.allocated.remove(freeId)
+                self.idNames.pop(freeId, None)
+        else:
+            raise ValueError(f"Id {freeId} is outside the available range")
+
+    def getAvailable(self) -> List:
+        return list(self.available)
+
+    def getAllocated(self) -> List:
+        return list(self.allocated)
+
+    def getIdNames(self) -> Dict:
+        return self.idNames

test/cvlib/id_allocator/test_allocator.py

@@ -0,0 +1,120 @@
+# Copyright (c) 2023 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the COPYING file.
+
+import pytest
+import re
+
+from cloudvision.cvlib import (
+    IdAllocator
+)
+
+checkNodeIdCases = [
+    # name
+    # allocations
+    # spine ids
+    # leaf ids
+    # mleaf ids
+    # expected error
+    [
+        'L3 valid',
+        'L3',
+        [
+            ('spine1', 1), ('spine2', 2),
+            ('leaf1', 1), ('leaf2', 2), ('leaf3', 3),
+            ('memberleaf1', 1), ('memberleaf2', 2), ('memberleaf3', 3),
+            ('memberleaf4', 4),
+        ],
+        [1, 2],
+        [1, 2, 3],
+        [1, 2, 3, 4],
+        None
+    ],
+    [
+        'L2 valid',
+        'L2',
+        [
+            ('spine1', 1), ('spine2', 2),
+            ('leaf1', 1), ('leaf2', 2), ('leaf3', 3),
+            ('memberleaf1', 4), ('memberleaf2', 5), ('memberleaf3', 6),
+            ('memberleaf4', 7),
+        ],
+        [1, 2],
+        [1, 2, 3, 4, 5, 6, 7],
+        [1, 2, 3, 4, 5, 6, 7],
+        None
+    ],
+    [
+        'L3 invalid spine duplicate id',
+        'L3',
+        [
+            ('spine1', 2), ('spine2', 2),
+            ('leaf1', 1), ('leaf2', 2), ('leaf3', 3),
+            ('memberleaf1', 1), ('memberleaf2', 2), ('memberleaf3', 3),
+            ('memberleaf4', 4),
+        ],
+        [1, 2],
+        [1, 2, 3],
+        [1, 2, 3, 4],
+        AssertionError("The same nodeID, 2, can not be applied to both "
+                       "of these spines: spine1, spine2")
+    ],
+    [
+        'L3 invalid leaf duplicate id',
+        'L3',
+        [
+            ('spine1', 1), ('spine2', 2),
+            ('leaf1', 1), ('leaf2', 1), ('leaf3', 3),
+            ('memberleaf1', 1), ('memberleaf2', 2), ('memberleaf3', 3),
+            ('memberleaf4', 4),
+        ],
+        [1, 2],
+        [1, 2, 3],
+        [1, 2, 3, 4],
+        AssertionError("The same nodeID, 1, can not be applied to both "
+                       "of these leafs: leaf1, leaf2")
+    ],
+    [
+        'L2 invalid leaf duplicate id',
+        'L2',
+        [
+            ('spine1', 1), ('spine2', 2),
+            ('leaf1', 1), ('leaf2', 2), ('leaf3', 3),
+            ('memberleaf1', 4), ('memberleaf2', 3), ('memberleaf3', 6),
+            ('memberleaf4', 7),
+        ],
+        [1, 2],
+        [1, 2, 3, 4, 5, 6, 7],
+        [1, 2, 3, 4, 5, 6, 7],
+        AssertionError("The same nodeID, 3, can not be applied to both "
+                       "of these leafs: leaf3, memberleaf2")
+    ],
+]
+
+
+@pytest.mark.parametrize('name, campusType, allocations, exp_spines, exp_leafs, '
+                         + 'exp_memberleafs, expectedError',
+                         checkNodeIdCases)
+def test_getAllDeviceTags(name, campusType, allocations, exp_spines, exp_leafs,
+                          exp_memberleafs, expectedError):
+    error = None
+    id_checkers = {}
+    id_checkers['spine'] = IdAllocator(idLabel='nodeID', groupLabel='spines')
+    if campusType.lower() == "l2":
+        id_checkers['leaf'] = IdAllocator(idLabel='nodeID', groupLabel='leafs')
+        id_checkers['memberleaf'] = id_checkers['leaf']
+    else:
+        id_checkers['leaf'] = IdAllocator(idLabel='nodeID', groupLabel='leafs')
+        id_checkers['memberleaf'] = IdAllocator(idLabel='nodeID', groupLabel='leafs')
+    try:
+        for (deviceId, nodeId) in allocations:
+            devtype = re.findall(r'(\D+)', deviceId)[0]
+            id_checkers[devtype].allocate(nodeId, deviceId)
+    except Exception as e:
+        error = e
+    if error or expectedError:
+        assert str(error) == str(expectedError)
+    else:
+        assert exp_spines == id_checkers['spine'].getAllocated()
+        assert exp_leafs == id_checkers['leaf'].getAllocated()
+        assert exp_memberleafs == id_checkers['memberleaf'].getAllocated()