fixup: add tcp and edge test cases

kgiusti · Sep 24, 2023 · edb9d3d · edb9d3d
1 parent cb3e9f6
commit edb9d3d
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 0 deletions.
diff --git a/tests/system_tests_edge_router.py b/tests/system_tests_edge_router.py
@@ -29,11 +29,13 @@
 from proton.utils import BlockingConnection
 
 from skupper_router.management.client import Node
+from skupper_router.management.error import ForbiddenStatus
 
 from system_test import TestCase, Qdrouterd, main_module, TIMEOUT, MgmtMsgProxy, TestTimeout
 from system_test import QdManager
 from system_test import unittest
 from system_test import Process
+from system_test import CONNECTION_TYPE
 from test_broker import FakeBroker
 
 from message_tests import DynamicAddressTest, MobileAddressAnonymousTest, MobileAddressTest
@@ -1499,6 +1501,25 @@ def test_73_skmanage_query_interior_from_edge(self):
 
         self.assertTrue(int_a_inter_router_conn_found and eb1_conn_found and eb2_conn_found)
 
+    def test_74_ensure_no_admin_delete(self):
+        """
+        This test tries to delete edge uplink connections but that operation
+        is forbidden
+        """
+        int_a = self.routers[0]
+        ea_1 = self.routers[2]
+
+        for r in (int_a, ea_1):
+            conns = [d for d in
+                     r.management.query(type=CONNECTION_TYPE).get_dicts() if
+                     d['role'] == 'edge']
+            self.assertNotEqual(0, len(conns), f"Expected at least one connection: {conns}")
+            for conn in conns:
+                with self.assertRaises(ForbiddenStatus):
+                    r.management.update(attributes={'adminStatus': 'deleted'},
+                                        type=CONNECTION_TYPE,
+                                        identity=conn['identity'])
+
 
 class ConnectivityTest(MessagingHandler):
     def __init__(self, interior_host, edge_host, edge_id):

diff --git a/tests/system_tests_tcp_adaptor.py b/tests/system_tests_tcp_adaptor.py
@@ -32,6 +32,8 @@
 from proton.handlers import MessagingHandler
 from proton.reactor import Container
 
+from skupper_router.management.error import ForbiddenStatus
+
 from system_test import Logger
 from system_test import Process
 from system_test import Qdrouterd
@@ -1733,6 +1735,17 @@ def _wait_for_close():
             self.assertEqual(2, c_stats['connectionsOpened'])
             self.assertEqual(1, c_stats['connectionsClosed'])
 
+            # Attempt to delete the special tcp-dispatcher pseudo connection.
+            # Doing so is forbidden so expect this to fail
+
+            conns = [d for d in mgmt.query(type=CONNECTION_TYPE).get_dicts()
+                     if d['host'] == 'egress-dispatch']
+            for conn in conns:
+                with self.assertRaises(ForbiddenStatus):
+                    mgmt.update(attributes={'adminStatus': 'deleted'},
+                                type=CONNECTION_TYPE,
+                                identity=conn['identity'])
+
             # splendid!  Not delete all the things
 
             mgmt.delete(type=LISTENER_TYPE, name=listener_name)