From edb9d3d7f6dd604de45ee1242dd13822f7e2f7c0 Mon Sep 17 00:00:00 2001 From: Kenneth Giusti Date: Sun, 24 Sep 2023 10:44:16 -0400 Subject: [PATCH] fixup: add tcp and edge test cases --- tests/system_tests_edge_router.py | 21 +++++++++++++++++++++ tests/system_tests_tcp_adaptor.py | 13 +++++++++++++ 2 files changed, 34 insertions(+) diff --git a/tests/system_tests_edge_router.py b/tests/system_tests_edge_router.py index f7477b306..684694c8c 100644 --- 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 index 1d4209e0f..70d0ebc5d 100644 --- 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)