Browse Source

ut: Test port operations

Port update and port delete are covered. There is added a test for port
update where binding in networking-ansible is triggered. Test simulates
a negative case where an error is coming up from ansible-runner. In such
situation the port binding should be set to failed status.

Change-Id: Ib64676499b23078ebf1f3cfecc46b1ebd5da7d6f
Jakub Libosvar 8 months ago
parent
commit
51eebc2797
1 changed files with 75 additions and 0 deletions
  1. 75
    0
      networking_ansible/tests/unit/ml2/test_mech_driver.py

+ 75
- 0
networking_ansible/tests/unit/ml2/test_mech_driver.py View File

@@ -13,6 +13,7 @@
13 13
 #    License for the specific language governing permissions and limitations
14 14
 #    under the License.
15 15
 
16
+import contextlib
16 17
 import mock
17 18
 import tempfile
18 19
 
@@ -25,6 +26,7 @@ from neutron_lib.api.definitions import provider_net
25 26
 import webob.exc
26 27
 
27 28
 from networking_ansible import ansible_networking as anet
29
+from networking_ansible import exceptions
28 30
 from networking_ansible.tests.unit import base
29 31
 
30 32
 
@@ -323,3 +325,76 @@ class TestML2PluginIntegration(NetAnsibleML2Base):
323 325
         self.assertItemsEqual(
324 326
             expected_calls,
325 327
             m_run_task.call_args_list)
328
+
329
+    @contextlib.contextmanager
330
+    def _create_unbound_port(self):
331
+        """Create a bound port in a network.
332
+
333
+        Network is created using self.network_spec defined in the setUp()
334
+        method of this class. Port attributes are defined in the
335
+        UNBOUND_PORT_SPEC.
336
+
337
+        """
338
+        with self.network('tenant', **self.network_spec) as n:
339
+            with self.subnet(network=n, cidr=self.CIDR) as s:
340
+                with self.port(
341
+                        subnet=s,
342
+                        **self.UNBOUND_PORT_SPEC
343
+                        ) as p:
344
+                    yield p
345
+
346
+    def test_update_port_unbound(self, m_run_task):
347
+        with self._create_unbound_port() as p:
348
+            req = self.new_update_request(
349
+                'ports',
350
+                self.BIND_PORT_UPDATE,
351
+                p['port']['id'])
352
+            m_run_task.reset_mock()
353
+
354
+            port = self.deserialize(
355
+                self.fmt, req.get_response(self.api))
356
+            m_run_task.called_once_with(
357
+                'update_port',
358
+                self.HOSTS[0],
359
+                self.network_spec[provider_net.SEGMENTATION_ID],
360
+                self.LOCAL_LINK_INFORMATION[0]['port_id'])
361
+            self.assertNotEqual(
362
+                portbindings.VIF_TYPE_BINDING_FAILED,
363
+                port['port'][portbindings.VIF_TYPE])
364
+
365
+    def test_delete_port(self, m_run_task):
366
+        with self._create_unbound_port() as p:
367
+            req = self.new_update_request(
368
+                'ports',
369
+                self.BIND_PORT_UPDATE,
370
+                p['port']['id'])
371
+            port = self.deserialize(
372
+                self.fmt, req.get_response(self.api))
373
+            m_run_task.reset_mock()
374
+
375
+            self._delete('ports', port['port']['id'])
376
+            m_run_task.called_once_with(
377
+                'delete_port',
378
+                self.HOSTS[0],
379
+                self.network_spec[provider_net.SEGMENTATION_ID],
380
+                self.LOCAL_LINK_INFORMATION[0]['port_id'])
381
+
382
+    def test_update_port_error(self, m_run_task):
383
+        with self._create_unbound_port() as p:
384
+            m_run_task.side_effect = exceptions.AnsibleRunnerException('foo')
385
+            self.assertEqual(
386
+                portbindings.VIF_TYPE_UNBOUND,
387
+                p['port'][portbindings.VIF_TYPE])
388
+
389
+            req = self.new_update_request(
390
+                'ports',
391
+                self.BIND_PORT_UPDATE,
392
+                p['port']['id'])
393
+            res = req.get_response(self.api)
394
+            port = self.deserialize(self.fmt, res)
395
+            # NOTE(jlibosva): The port update was successful as the binding was
396
+            # changed. The way how to check the error is via its binding value
397
+            self.assertEqual(
398
+                portbindings.VIF_TYPE_BINDING_FAILED,
399
+                port['port'][portbindings.VIF_TYPE])
400
+            self.assertEqual(webob.exc.HTTPOk.code, res.status_int)

Loading…
Cancel
Save