test/vpp_object.py

   1 """ abstract vpp object and object registry """
   2
   3 import abc
   4
   5
   6 class VppObject(metaclass=abc.ABCMeta):
   7     """ Abstract vpp object """
   8
   9     @abc.abstractmethod
  10     def add_vpp_config(self) -> None:
  11         """ Add the configuration for this object to vpp. """
  12         pass
  13
  14     @abc.abstractmethod
  15     def query_vpp_config(self) -> bool:
  16         """Query the vpp configuration.
  17
  18         :return: True if the object is configured"""
  19         pass
  20
  21     @abc.abstractmethod
  22     def remove_vpp_config(self) -> None:
  23         """ Remove the configuration for this object from vpp. """
  24         pass
  25
  26     def object_id(self) -> str:
  27         """ Return a unique string representing this object. """
  28         return "Undefined. for <%s %s>" % (self.__class__.__name__, id(self))
  29
  30     def __str__(self) -> str:
  31         return self.object_id()
  32
  33     def __repr__(self) -> str:
  34         return '<%s>' % self.object_id()
  35
  36     def __hash__(self) -> int:
  37         return hash(self.object_id())
  38
  39     def __eq__(self, other) -> bool:
  40         if not isinstance(other, self.__class__):
  41             return NotImplemented
  42         if other.object_id() == self.object_id():
  43             return True
  44         return False
  45
  46     # This can be removed when python2 support is dropped.
  47     def __ne__(self, other):
  48         return not self.__eq__(other)
  49
  50
  51 class VppObjectRegistry:
  52     """ Class which handles automatic configuration cleanup. """
  53     _shared_state = {}
  54
  55     def __init__(self) -> None:
  56         self.__dict__ = self._shared_state
  57         if not hasattr(self, "_object_registry"):
  58             self._object_registry = []
  59         if not hasattr(self, "_object_dict"):
  60             self._object_dict = dict()
  61
  62     def register(self, obj: VppObject, logger) -> None:
  63         """ Register an object in the registry. """
  64         if obj.object_id() not in self._object_dict:
  65             self._object_registry.append(obj)
  66             self._object_dict[obj.object_id()] = obj
  67             logger.debug("REG: registering %s" % obj)
  68         else:
  69             logger.debug("REG: duplicate add, ignoring (%s)" % obj)
  70
  71     def unregister_all(self, logger) -> None:
  72         """ Remove all object registrations from registry. """
  73         logger.debug("REG: removing all object registrations")
  74         self._object_registry = []
  75         self._object_dict = dict()
  76
  77     def remove_vpp_config(self, logger) -> None:
  78         """
  79         Remove configuration (if present) from vpp and then remove all objects
  80         from the registry.
  81         """
  82         if not self._object_registry:
  83             logger.info("REG: No objects registered for auto-cleanup.")
  84             return
  85         logger.info("REG: Removing VPP configuration for registered objects")
  86         # remove the config in reverse order as there might be dependencies
  87         failed = []
  88         for obj in reversed(self._object_registry):
  89             if obj.query_vpp_config():
  90                 logger.info("REG: Removing configuration for %s" % obj)
  91                 obj.remove_vpp_config()
  92                 if obj.query_vpp_config():
  93                     failed.append(obj)
  94             else:
  95                 logger.info(
  96                     "REG: Skipping removal for %s, configuration not present" %
  97                     obj)
  98         self.unregister_all(logger)
  99         if failed:
 100             logger.error("REG: Couldn't remove configuration for object(s):")
 101             for obj in failed:
 102                 logger.error(repr(obj))
 103             raise Exception("Couldn't remove configuration for object(s): %s" %
 104                             (", ".join(str(x) for x in failed)))