Add get_security_group_by_id to network driver

Change-Id: I76f00ef9234a615c5ea275532f5fe831345060b8

octavia/network/base.py

@@ -13,10 +13,15 @@
 #    under the License.
 
 import abc
+import typing
 
 from octavia.common import constants
 from octavia.common import exceptions
 
+if typing.TYPE_CHECKING:
+    from octavia.common import context
+    import octavia.network.data_models as n_data_models
+
 
 class NetworkException(exceptions.OctaviaException):
     pass
@@ -294,13 +299,25 @@ class AbstractNetworkDriver(metaclass=abc.ABCMeta):
 
     @abc.abstractmethod
     def get_security_group(self, sg_name):
-        """Retrieves the security group by it's name.
+        """Retrieves the security group by its name.
 
         :param sg_name: The security group name.
         :return: octavia.network.data_models.SecurityGroup, None if not enabled
         :raises: NetworkException, SecurityGroupNotFound
         """
 
+    @abc.abstractmethod
+    def get_security_group_by_id(self, sg_id: str,
+                                 context: 'context.RequestContext' = None) -> (
+            'n_data_models.SecurityGroup'):
+        """Retrieves the security group by its id.
+
+        :param sg_id: The security group ID.
+        :param context: A request context
+        :return: octavia.network.data_models.SecurityGroup, None if not enabled
+        :raises: NetworkException, SecurityGroupNotFound
+        """
+
     @abc.abstractmethod
     def failover_preparation(self, amphora):
         """Prepare an amphora for failover.

octavia/network/drivers/neutron/allowed_address_pairs.py

@@ -876,7 +876,7 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
             raise base.CreatePortException(message)
 
     def get_security_group(self, sg_name):
-        """Retrieves the security group by it's name.
+        """Retrieves the security group by its name.
 
         :param sg_name: The security group name.
         :return: octavia.network.data_models.SecurityGroup, None if not enabled

octavia/network/drivers/neutron/base.py

@@ -12,6 +12,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import typing
+
 from openstack.connection import Connection
 import openstack.exceptions as os_exceptions
 from openstack.network.v2._proxy import Proxy
@@ -25,6 +27,9 @@ from octavia.network import base
 from octavia.network import data_models as network_models
 from octavia.network.drivers.neutron import utils
 
+if typing.TYPE_CHECKING:
+    from octavia.common import context
+
 LOG = logging.getLogger(__name__)
 DNS_INT_EXT_ALIAS = 'dns-integration'
 SEC_GRP_EXT_ALIAS = 'security-group'
@@ -252,6 +257,11 @@ class BaseNeutronDriver(base.AbstractNetworkDriver):
     def get_port(self, port_id, context=None):
         return self._get_resource('port', port_id, context=context)
 
+    def get_security_group_by_id(self, sg_id: str,
+                                 context: 'context.RequestContext' = None) -> (
+            'network_models.SecurityGroup'):
+        return self._get_resource('security_group', sg_id, context=context)
+
     def get_network_by_name(self, network_name):
         return self._get_resources_by_filters(
             'network', unique_item=True, name=network_name)

octavia/network/drivers/noop_driver/driver.py

@@ -12,6 +12,8 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import typing
+
 from oslo_log import log as logging
 from oslo_utils import uuidutils
 
@@ -20,6 +22,9 @@ from octavia.common import data_models
 from octavia.network import base as driver_base
 from octavia.network import data_models as network_models
 
+if typing.TYPE_CHECKING:
+    from octavia.common import context
+
 LOG = logging.getLogger(__name__)
 
 _NOOP_MANAGER_VARS = {
@@ -228,6 +233,14 @@ class NoopManager:
         _NOOP_MANAGER_VARS['ports'][port_id] = port
         return port
 
+    def get_security_group_by_id(self, sg_id: str,
+                                 context: 'context.RequestContext' = None) -> (
+            'network_models.SecurityGroup'):
+        LOG.debug("Network %s no-op, get_security_group_by_id %s",
+                  self.__class__.__name__, sg_id)
+        self.networkconfigconfig[sg_id] = (sg_id, 'get_security_group_by_id')
+        return network_models.SecurityGroup(id=sg_id)
+
     def get_network_by_name(self, network_name):
         LOG.debug("Network %s no-op, get_network_by_name network_name %s",
                   self.__class__.__name__, network_name)
@@ -475,6 +488,11 @@ class NoopNetworkDriver(driver_base.AbstractNetworkDriver):
     def get_port(self, port_id, context=None):
         return self.driver.get_port(port_id)
 
+    def get_security_group_by_id(self, sg_id: str,
+                                 context: 'context.RequestContext' = None) -> (
+            'network_models.SecurityGroup'):
+        return self.driver.get_security_group_by_id(sg_id, context=context)
+
     def get_qos_policy(self, qos_policy_id):
         return self.driver.get_qos_policy(qos_policy_id)