fix: openstack needs to use neutron for floating ip association

VERSION

@@ -1 +1 @@
-1!10.11.0
+1!10.12.0

examples/openstack_example.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+# This file is part of pycloudlib. See LICENSE file for license information.
+"""Basic examples of various lifecycles with a Openstack instance."""
+
+import logging
+import os
+import sys
+
+import pycloudlib
+
+REQUIRED_ENV_VARS = ("OS_AUTH_URL", "OS_PASSWORD", "OS_USERNAME")
+
+
+def basic_lifecycle(image_id: str):
+    """Demonstrate basic set of lifecycle operations with OpenStack."""
+    with pycloudlib.Openstack("pycloudlib-test") as os_cloud:
+        with os_cloud.launch(image_id=image_id) as inst:
+            inst.wait()
+
+            result = inst.execute("uptime")
+            print(result)
+            inst.console_log()
+            inst.delete(wait=False)
+
+
+def demo(image_id: str):
+    """Show examples of using the Openstack module."""
+    basic_lifecycle(image_id)
+
+
+def assert_openstack_config():
+    """Assert any required OpenStack env variables and args needed for demo."""
+    if len(sys.argv) != 2:
+        sys.stderr.write(
+            f"Usage: {sys.argv[0]} <openstack_image_id>\n"
+            "Must provide an image id from openstack image list\n\n"
+        )
+        sys.exit(1)
+    for env_name in REQUIRED_ENV_VARS:
+        assert os.environ.get(
+            env_name
+        ), f"Missing required Openstack environment variable: {env_name}"
+
+
+if __name__ == "__main__":
+    assert_openstack_config()
+    logging.basicConfig(level=logging.DEBUG)
+    image_id = sys.argv[1]
+    demo(image_id=sys.argv[1])

pycloudlib/openstack/instance.py

@@ -66,13 +66,20 @@ def _get_existing_floating_ip(self):
 
     def _create_and_attach_floating_ip(self):
         floating_ip = self.conn.create_floating_ip(wait=True)
-        tries = 30
-        for _ in range(tries):
+        for _ in range(30):
             try:
-                self.conn.compute.add_floating_ip_to_server(
-                    self.server, floating_ip.floating_ip_address
-                )
+                ports = [p for p in self.conn.network.ports(device_id=self.server.id)]
+                if not ports:
+                    self._log.debug(f"Server {self.name} ports not yet available; sleeping")
+                    time.sleep(1)
+                    continue
+                # Assign IP to first port on the server
+                self.conn.network.update_ip(floating_ip, port_id=ports[0].id)
                 break
+            except ResourceNotFound as e:
+                if "Floating IP" in str(e):
+                    time.sleep(1)
+                    continue
             except BadRequestException as e:
                 if "Instance network is not ready yet" in str(e):
                     time.sleep(1)

tests/unit_tests/openstack/test_instance.py

@@ -1,5 +1,7 @@
 """Openstack instance tests."""
 
+import pytest
+
 from unittest import mock
 
 from pycloudlib.openstack.instance import OpenstackInstance
@@ -37,38 +39,45 @@
 ]
 
 
-@mock.patch("pycloudlib.openstack.instance.OpenstackInstance._create_and_attach_floating_ip")
 class TestAttachFloatingIp:
     """Ensure we create/use floating IPs accordingly."""
 
-    def test_existing_floating_ip(self, m_create):
-        """Test that if a server has an existing floating IP, we use it."""
-        m_connection = mock.Mock()
-        m_server = m_connection.compute.get_server.return_value
+    @pytest.fixture(autouse=True)
+    def setup_connection(self):
+        self.conn = mock.Mock()
+        m_server = self.conn.compute.get_server.return_value
         m_server.addresses = SERVER_ADDRESSES
-        m_connection.network.ips.return_value = NETWORK_IPS
+        m_create_floating_ip = self.conn.create_floating_ip.return_value
+        m_create_floating_ip.floating_ip_address = "10.42.42.42"
+        self.conn.network.ports.return_value = [
+            mock.Mock(id="port1"), mock.Mock(id="port2")
+        ]
+
+    def test_existing_floating_ip(self):
+        """Test that if a server has an existing floating IP, we use it."""
+        self.conn.network.ips.return_value = NETWORK_IPS
 
         instance = OpenstackInstance(
             key_pair=None,
             instance_id=None,
             network_id=None,
-            connection=m_connection,
+            connection=self.conn,
         )
         assert "10.0.0.3" == instance.floating_ip["floating_ip_address"]
-        assert 0 == m_create.call_count
+        assert 0 == self.conn.create_floating_ip.call_count
 
-    def test_no_matching_floating_ip(self, m_create):
+    def test_no_matching_floating_ip(self):
         """Test that if a server doesn't have a floating IP, we create it."""
-        m_connection = mock.Mock()
-        m_server = m_connection.compute.get_server.return_value = mock.Mock()
-        m_server.addresses = SERVER_ADDRESSES
-        m_connection.network.ips.return_value = []
+        self.conn.network.ips.return_value = []
 
         instance = OpenstackInstance(
             key_pair=None,
             instance_id=None,
             network_id=None,
-            connection=m_connection,
+            connection=self.conn,
+        )
+        assert instance.floating_ip is self.conn.create_floating_ip.return_value
+        assert 1 == self.conn.create_floating_ip.call_count
+        self.conn.network.update_ip.assert_called_once_with(
+            self.conn.create_floating_ip.return_value, port_id='port1'
         )
-        assert instance.floating_ip is m_create.return_value
-        assert 1 == m_create.call_count