625 feat vpc version management

otcextensions/sdk/apig/v2/_proxy.py

@@ -47,6 +47,7 @@
 from otcextensions.sdk.apig.v2 import ssl_domain as _ssl_domain
 from otcextensions.sdk.apig.v2 import tag as _tag
 from otcextensions.sdk.apig.v2 import config as _config
+from otcextensions.sdk.apig.v2 import vpc_endpoint as _vpc_endpoint
 
 
 class Proxy(proxy.Proxy):
@@ -3469,3 +3470,27 @@ def configs_for_gateway(self, gateway_id, **attrs):
             paginated=False,
             base_path=base_path,
             **attrs,)
+
+    # ======== VPC Endpoint Management Methods ========
+
+    def vpc_endpoints(self, gateway, **attrs):
+        """List all VPC endpoints
+
+        This method retrieves all VPC endpoints associated with the specified
+        API Gateway instance.
+
+        :param gateway: The ID or an instance of
+            :class:`~otcextensions.sdk.apig.v2.gateway.Gateway`
+        :param attrs: Optional query parameters for filtering the list,
+            such as limit, offset, id, marker_id, status
+
+        :returns: A generator of
+            :class:`~otcextensions.sdk.apig.v2.vpc_endpoint.VpcEndpoint`
+            instances
+        """
+        gateway = self._get_resource(_gateway.Gateway, gateway)
+        return self._list(
+            _vpc_endpoint.VpcEndpoint,
+            gateway_id=gateway.id,
+            **attrs
+        )

otcextensions/sdk/apig/v2/vpc_endpoint.py

@@ -0,0 +1,35 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+from openstack import resource
+
+
+class VpcEndpoint(resource.Resource):
+    resources_key = "connections"
+    base_path = f'apigw/instances/%(gateway_id)s/vpc-endpoint/connections'
+
+    _query_mapping = resource.QueryParameters('limit', 'offset', 'id',
+                                              'marker_id', 'status')
+
+    # capabilities
+    allow_fetch = True
+    allow_commit = True
+    allow_list = True
+
+    gateway_id = resource.URI('gateway_id')
+
+    # Properties
+    id = resource.Body("id")
+    marker_id = resource.Body("marker_id", type=int)
+    created_at = resource.Body("created_at")
+    updated_at = resource.Body("updated_at")
+    domain_id = resource.Body("domain_id")
+    status = resource.Body("status")

otcextensions/sdk/vlb/v3/_proxy.py

@@ -133,6 +133,21 @@ def wait_for_load_balancer(self, name_or_id, status='ACTIVE',
         return resource.wait_for_status(self, lb, status, failures, interval,
                                         wait, attribute='provisioning_status')
 
+    def wait_for_delete_load_balancer(self, name_or_id, interval=2, wait=300):
+        """Wait for a load balancer to be deleted.
+
+        :param name_or_id: The name or ID of a load balancer.
+        :param interval: Number of seconds to wait between checks.
+            Default is 2.
+        :param wait: Maximum number of seconds to wait for deletion.
+            Default is 300.
+        :returns: True if the resource is deleted, False if timeout.
+        :raises: :class:`~openstack.exceptions.ResourceTimeout` if transition
+            to delete failed to occur in the specified seconds.
+        """
+        lb = self._get_resource(_lb.LoadBalancer, name_or_id)
+        return resource.wait_for_delete(self, lb, interval, wait)
+
     def get_load_balancer_statuses(self, loadbalancer_id):
         """Get specific load balancer statuses by load balancer id.
 

otcextensions/sdk/vpcep/v1/_proxy.py

@@ -210,10 +210,8 @@ def manage_service_connections(self, service, action, endpoints=[]):
         :param action: action can be ``accept`` or ``reject``.
         :param endpoints: List of VPC Endpoints Id.
 
-        :returns: A generator of connection objects.
-        :rtype: :class:`~otcextensions.sdk.vpcep.connection.Connection`
+        :returns: List of connection objects.
         """
-
         endpoint_service = self._get_resource(_service.Service, service)
         connection = self._get_resource(
             _connection.Connection,

otcextensions/sdk/vpcep/v1/connection.py

@@ -54,8 +54,10 @@ def _action(self, session, action, endpoints=[]):
         body = {'endpoints': endpoints, 'action': action}
         response = session.post(uri, json=body)
         exceptions.raise_from_response(response)
-        for raw_resource in response.json()[self.resources_key]:
-            yield Connection.existing(**raw_resource)
+        return [
+            Connection.existing(**raw_resource)
+            for raw_resource in response.json()[self.resources_key]
+        ]
 
     def accept(self, session, endpoints=[]):
         """Accept connections."""

otcextensions/tests/functional/base.py

@@ -11,6 +11,7 @@
 # under the License.
 
 import os
+import uuid
 
 from keystoneauth1 import exceptions as _exceptions
 
@@ -72,3 +73,86 @@ def setUp(self):
         except _exceptions.EndpointNotFound:
             self.skipTest('Service {service_type} not found in cloud'.format(
                 service_type=service_type))
+
+
+class NetworkBaseFunctionalTest(BaseFunctionalTest):
+    def create_network(self, prefix='sdk-test'):
+        cidr = '192.168.0.0/16'
+        ipv4 = 4
+        uuid_v4 = uuid.uuid4().hex[:8]
+        router_name = prefix + '-router-' + uuid_v4
+        net_name = prefix + '-net-' + uuid_v4
+        subnet_name = prefix + '-subnet-' + uuid_v4
+
+        network = self.conn.network.create_network(name=net_name)
+        self.assertEqual(net_name, network.name)
+        net_id = network.id
+        subnet = self.conn.network.create_subnet(
+            name=subnet_name,
+            ip_version=ipv4,
+            network_id=net_id,
+            cidr=cidr
+        )
+        self.assertEqual(subnet_name, subnet.name)
+        subnet_id = subnet.id
+
+        router = self.conn.network.create_router(name=router_name)
+        self.assertEqual(router_name, router.name)
+        router_id = router.id
+        interface = router.add_interface(
+            self.conn.network,
+            subnet_id=subnet_id
+        )
+        self.assertEqual(interface['subnet_id'], subnet_id)
+        self.assertIn('port_id', interface)
+        return {
+            'router_id': router_id,
+            'subnet_id': subnet_id,
+            'network_id': net_id
+        }
+
+    def destroy_network(self, params: dict):
+        router_id = params.get('router_id')
+        subnet_id = params.get('subnet_id')
+        network_id = params.get('network_id')
+        router = self.conn.network.get_router(router_id)
+
+        interface = router.remove_interface(
+            self.conn.network,
+            subnet_id=subnet_id
+        )
+        self.assertEqual(interface['subnet_id'], subnet_id)
+        self.assertIn('port_id', interface)
+        sot = self.conn.network.delete_router(
+            router_id,
+            ignore_missing=False
+        )
+        self.assertIsNone(sot)
+        sot = self.conn.network.delete_subnet(
+            subnet_id,
+            ignore_missing=False
+        )
+        self.assertIsNone(sot)
+        sot = self.conn.network.delete_network(
+            network_id,
+            ignore_missing=False
+        )
+        self.assertIsNone(sot)
+
+    def create_port(self, network_id, prefix='sdk-test'):
+        uuid_v4 = uuid.uuid4().hex[:8]
+        port_name = prefix + '-port-' + uuid_v4
+
+        port = self.conn.network.create_port(
+            name=port_name,
+            network_id=network_id
+        )
+        self.assertEqual(port_name, port.name)
+        return port
+
+    def destroy_port(self, port):
+        sot = self.conn.network.delete_port(
+            port,
+            ignore_missing=False
+        )
+        self.assertIsNone(sot)

otcextensions/tests/functional/sdk/apig/v2/test_vpc_endpoint.py

@@ -0,0 +1,30 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from otcextensions.tests.functional.sdk.apig import TestApiG
+
+
+class TestVpcEndpoint(TestApiG):
+    gateway = ""
+
+    def setUp(self):
+        super(TestVpcEndpoint, self).setUp()
+
+    def test_01_list_vpc_endpoint(self):
+        self.conn.vpcep.create_endpoint(
+            subnet_id='',
+            endpoint_service_id='',
+            vpc_id='',
+        )
+        list(self.client.vpc_endpoints(
+            gateway=TestVpcEndpoint.gateway,
+        ))