Skip to content

Commit

Permalink
Fixes #1707: Added READ management tests to increase auto link code c…
Browse files Browse the repository at this point in the history
…overage (#1708)
  • Loading branch information
ganeshmurthy authored Dec 23, 2024
1 parent 745ca5b commit 024b63d
Show file tree
Hide file tree
Showing 2 changed files with 30 additions and 2 deletions.
16 changes: 14 additions & 2 deletions tests/system_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
16 changes: 16 additions & 0 deletions tests/system_tests_autolinks.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down

0 comments on commit 024b63d

Please sign in to comment.