diff --git a/tests/system_test.py b/tests/system_test.py index 8d047e035..1747e0739 100755 --- a/tests/system_test.py +++ b/tests/system_test.py @@ -1823,8 +1823,20 @@ def create(self, long_type, kwargs): cmd += " %s=%s" % (k, v) return json.loads(self(cmd)) - def read(self, name): - return json.loads(self(f"READ --name {name}")) + # According to the AMQP spec, the READ operation does not require a type, requires either a name or identity. + # But the C management agent requires a long type to be specified (see management_agent.c) + def read(self, long_type=None, name=None, identity=None): + if name is not None and identity is not None: + return json.loads(self(f"READ --type={long_type} --name {name} --identity {identity}")) + if long_type is not None: + if name is not None: + return json.loads(self(f"READ --type={long_type} --name {name}")) + if identity is not None: + return json.loads(self(f"READ --type={long_type} --identity {identity}")) + else: + if identity is not None: + return json.loads(self(f"READ --identity {identity}")) + return json.loads(self(f"READ --name {name}")) def update(self, long_type, kwargs, name=None, identity=None): cmd = 'UPDATE --type=%s' % long_type diff --git a/tests/system_tests_autolinks.py b/tests/system_tests_autolinks.py index 776637d4e..a191b0916 100644 --- a/tests/system_tests_autolinks.py +++ b/tests/system_tests_autolinks.py @@ -116,6 +116,22 @@ def test_name_collision(self): args = {"name": "autoLink", "address": "autoLink1", "connection": "broker", "direction": "in"} # Add autoLink with the same name as the one already present. mgmt = SkManager(address=self.router.addresses[0]) + + not_found = False + try: + mgmt.read(long_type=CONFIG_AUTOLINK_TYPE, name="autoLink10") + except Exception as e: + error_message = str(e) + if "NotFoundStatus: Not Found" in error_message: + not_found = True + self.assertTrue(not_found) + + auto_link = mgmt.read(long_type=CONFIG_AUTOLINK_TYPE, name="autoLink") + self.assertIsNotNone(auto_link) + + auto_link = mgmt.read(long_type=CONFIG_AUTOLINK_TYPE, identity="2") + self.assertIsNotNone(auto_link) + test_pass = False try: mgmt.create(CONFIG_AUTOLINK_TYPE, args)