| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License |
| # |
| |
| """System tests for management of qdrouter""" |
| |
| import json |
| import os |
| import re |
| from itertools import chain |
| |
| from proton import Message |
| from proton.handlers import MessagingHandler |
| from proton.reactor import Container |
| |
| from qpid_dispatch.management.client import Node, Url |
| from qpid_dispatch.management.error import ManagementError, BadRequestStatus, NotImplementedStatus, NotFoundStatus |
| from qpid_dispatch_internal.management.qdrouter import QdSchema |
| from qpid_dispatch_internal.compat import dictify |
| from qpid_dispatch_internal.compat import BINARY |
| |
| import system_test |
| from system_test import Qdrouterd, Process |
| from system_test import unittest |
| |
| PREFIX = 'org.apache.qpid.dispatch.' |
| MANAGEMENT = PREFIX + 'management' |
| CONFIGURATION = PREFIX + 'configurationEntity' |
| OPERATIONAL = PREFIX + 'operationalEntity' |
| LISTENER = PREFIX + 'listener' |
| CONNECTOR = PREFIX + 'connector' |
| DUMMY = PREFIX + 'dummy' |
| ROUTER = PREFIX + 'router' |
| LINK = ROUTER + '.link' |
| ADDRESS = ROUTER + '.address' |
| NODE = ROUTER + '.node' |
| CONFIG_ADDRESS = ROUTER + '.config.address' |
| |
| |
| def short_name(name): |
| if name.startswith(PREFIX): |
| return name[len(PREFIX):] |
| return name |
| |
| |
| class ManagementTest(system_test.TestCase): |
| |
| @classmethod |
| def setUpClass(cls): |
| super(ManagementTest, cls).setUpClass() |
| # Stand-alone router |
| conf0 = Qdrouterd.Config([ |
| ('router', {'mode': 'standalone', 'id': 'solo', 'metadata': 'selftest;solo'}), |
| ('listener', {'name': 'l0', 'port': cls.tester.get_port(), 'role': 'normal'}), |
| # Extra listeners to exercise managment query |
| ('listener', {'name': 'l1', 'port': cls.tester.get_port(), 'role': 'normal'}), |
| ('listener', {'name': 'l2', 'port': cls.tester.get_port(), 'role': 'normal'}) |
| ]) |
| cls._router = cls.tester.qdrouterd(config=conf0, wait=False) |
| |
| # Trio of interior routers linked in a line so we can see some next-hop values. |
| conf0 = Qdrouterd.Config([ |
| ('router', {'mode': 'interior', 'id': 'router0'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'normal'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'inter-router'}) |
| ]) |
| conf1 = Qdrouterd.Config([ |
| ('router', {'mode': 'interior', 'id': 'router1'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'normal'}), |
| ('connector', {'port': conf0.sections('listener')[1]['port'], 'role':'inter-router'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'inter-router'}) |
| ]) |
| conf2 = Qdrouterd.Config([ |
| ('router', {'mode': 'interior', 'id': 'router2'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'normal'}), |
| ('connector', {'port': conf1.sections('listener')[1]['port'], 'role':'inter-router'}) |
| ]) |
| cls._routers = [cls.tester.qdrouterd(config=c, wait=False) for c in [conf0, conf1, conf2]] |
| |
| # Standalone router for logging tests (avoid interfering with logging for other tests.) |
| conflog = Qdrouterd.Config([ |
| ('router', {'mode': 'standalone', 'id': 'logrouter'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'normal'}), |
| ]) |
| cls._logrouter = cls.tester.qdrouterd(config=conflog, wait=False) |
| |
| @property |
| def router(self): |
| return self.__class__._router.wait_ready() |
| |
| @property |
| def logrouter(self): |
| return self.__class__._logrouter.wait_ready() |
| |
| @property |
| def routers(self): |
| """Wait on demand and return the linked interior routers""" |
| if not self._routers: |
| self._routers = self.__class__._routers |
| self._routers[0].wait_router_connected('router1') |
| self._routers[1].wait_router_connected('router2') |
| self._routers[2].wait_router_connected('router0') |
| return self._routers |
| |
| def setUp(self): |
| super(ManagementTest, self).setUp() |
| self._routers = None # Wait on demand |
| self.maxDiff = None |
| self.longMessage = True |
| self.node = self.cleanup(Node.connect(self.router.addresses[0])) |
| |
| def test_bad_query(self): |
| """Test that various badly formed queries get the proper response""" |
| # No operation attribute |
| self.assertRaises(BadRequestStatus, self.node.call, self.node.request()) |
| self.assertRaises(NotImplementedStatus, self.node.call, |
| self.node.request(operation="nosuch", type="org.amqp.management")) |
| |
| def test_metadata(self): |
| """Query with type only""" |
| response = self.node.query(type=ROUTER) |
| for attr in ['type', 'metadata']: |
| self.assertIn(attr, response.attribute_names) |
| self.assertEqual(response.get_entities()[0]['metadata'], 'selftest;solo') |
| |
| def test_query_type(self): |
| """Query with type only""" |
| response = self.node.query(type=LISTENER) |
| for attr in ['type', 'name', 'identity', 'host', 'port']: |
| self.assertIn(attr, response.attribute_names) |
| for r in response.get_dicts(): |
| self.assertEqual(len(response.attribute_names), len(r)) |
| self.assertEqual(r['type'], LISTENER) |
| self.assertTrue( |
| {'l0', 'l1', 'l2'} <= set(r['name'] for r in response.get_entities())) |
| |
| def test_query_type_attributes(self): |
| """Query with type and attribute names""" |
| attribute_names = ['type', 'name', 'port'] |
| response = self.node.query(type=LISTENER, attribute_names=attribute_names) |
| self.assertEqual(attribute_names, response.attribute_names) |
| expect = [[LISTENER, 'l%s' % i, str(self.router.ports[i])] for i in range(3)] |
| for r in expect: # We might have extras in results due to create tests |
| self.assertIn(r, response.results) |
| self.assertIn(dict(zip(attribute_names, r)), response.get_dicts()) |
| |
| def test_query_attributes(self): |
| """Query with attributes only""" |
| attribute_names = ['type', 'name', 'port'] |
| response = self.node.query(attribute_names=attribute_names) |
| self.assertEqual(attribute_names, response.attribute_names) |
| expect = [[LISTENER, 'l%s' % i, str(self.router.ports[i])] for i in range(3)] |
| for r in expect: # We might have extras in results due to create tests |
| self.assertIn(r, response.results) |
| for name in ['router/' + self.router.name, 'log/DEFAULT']: |
| self.assertTrue([r for r in response.get_dicts() if r['name'] == name], |
| msg="Can't find result with name '%s'" % name) |
| |
| def assertMapSubset(self, small, big): |
| """Assert that mapping small is a subset of mapping big""" |
| missing = [(k, v) for k, v in small.items() if (k, v) not in big.items()] |
| assert not missing, "Not a subset, missing %s, sub=%s, super=%s" % (missing, small, big) |
| |
| def assert_create_ok(self, type, name, attributes): |
| entity = self.node.create(attributes, type, name) |
| self.assertMapSubset(attributes, entity.attributes) |
| return entity |
| |
| def assert_read_ok(self, type, name, attributes): |
| entity = self.node.read(type, name) |
| self.assertMapSubset(attributes, entity.attributes) |
| return entity |
| |
| def test_create_listener(self): |
| """Create a new listener on a running router""" |
| |
| port = self.get_port() |
| # Note qdrouter schema defines port as string not int, since it can be a service name. |
| attributes = {'name': 'foo', 'port': str(port), 'role': 'normal', 'saslMechanisms': 'ANONYMOUS', 'authenticatePeer': False} |
| entity = self.assert_create_ok(LISTENER, 'foo', attributes) |
| self.assertEqual(entity['name'], 'foo') |
| self.assertEqual(entity['host'], '') |
| |
| # Connect via the new listener |
| node3 = self.cleanup(Node.connect(Url(port=port))) |
| router = node3.query(type=ROUTER).get_entities() |
| self.assertEqual(self.router.name, router[0]['id']) |
| |
| # Delete the listener |
| entity.delete() |
| response = self.node.query(type=LISTENER, attribute_names=['name']) |
| for l in response.get_dicts(): |
| self.assertTrue(l['name'] != 'foo') |
| |
| def test_log(self): |
| """Create, update and query log entities""" |
| |
| node = self.cleanup(Node.connect(self.logrouter.addresses[0])) |
| default = node.read(identity='log/DEFAULT') |
| self.assertEqual(default.attributes, |
| {'identity': 'log/DEFAULT', |
| 'enable': 'trace+', |
| 'module': 'DEFAULT', |
| 'name': 'log/DEFAULT', |
| 'outputFile': 'logrouter.log', |
| 'includeSource': True, |
| 'includeTimestamp': True, |
| 'type': 'org.apache.qpid.dispatch.log'}) |
| |
| def check_log(log, error=True, debug=False): |
| """Cause an error and check for expected error and debug logs""" |
| bad_type = "nosuch" |
| self.assertRaises(ManagementError, node.create, type=bad_type, name=bad_type) |
| f = self.cleanup(open(log)) |
| logstr = f.read() |
| |
| def assert_expected(expect, regex, logstr): |
| match = re.search(regex, logstr) |
| assert bool(expect) == bool(match), "%s %s:\n%s" % ( |
| ((match and "Found") or "Not found"), regex, logstr) |
| assert_expected(error, r'AGENT \(error\).*%s' % bad_type, logstr) |
| assert_expected(debug, r'AGENT \(debug\)', logstr) |
| |
| log_count = [0] # In list to work-around daft python scoping rules. |
| |
| def update_check_log(attributes, error=True, debug=False): |
| log_count[0] += 1 |
| log = os.path.abspath("test_log.log%s" % log_count[0]) |
| attributes["outputFile"] = log |
| attributes["identity"] = "log/AGENT" |
| node.update(attributes) |
| check_log(log, error, debug) |
| |
| # Expect error but no debug |
| update_check_log(dict(enable="warning+")) |
| update_check_log(dict(enable="error")) |
| update_check_log(dict(enable="TRACE , Error info")) # Case and space insensitive |
| |
| # Expect no error if not enabled. |
| update_check_log(dict(enable="info,critical"), error=False) |
| update_check_log(dict(enable="none"), error=False) |
| update_check_log(dict(enable=""), error=False) |
| |
| # Expect debug |
| update_check_log(dict(enable="Debug"), error=False, debug=True) |
| update_check_log(dict(enable="trace+"), debug=True) |
| |
| # Check defaults are picked up |
| update_check_log(dict(enable="default"), error=True, debug=True) |
| node.update(dict(identity="log/DEFAULT", enable="debug")) |
| update_check_log(dict(enable="DEFAULT"), error=False, debug=True) |
| node.update(dict(identity="log/DEFAULT", enable="error")) |
| update_check_log(dict(enable="default"), error=True, debug=False) |
| |
| # Invalid values |
| self.assertRaises(ManagementError, node.update, dict(identity="log/AGENT", enable="foo")) |
| self.assertRaises(ManagementError, node.update, |
| dict(identity="log/AGENT", enable="default", |
| includeTimestamp="blahblah")) |
| self.assertRaises(ManagementError, node.update, |
| dict(identity="log/AGENT", enable="default", |
| includeSource="blahblah")) |
| self.assertRaises(ManagementError, node.update, |
| dict(identity="BOGUS", enable="default")) |
| |
| def test_create_config_address(self): |
| self.assert_create_ok(CONFIG_ADDRESS, 'myConfigAddr', dict(prefix='prefixA')) |
| self.assert_read_ok(CONFIG_ADDRESS, 'myConfigAddr', |
| dict(prefix='prefixA', pattern=None)) |
| simple_send_receive_test = SimpleSndRecv(self.router.addresses[0], '/prefixA/other') |
| simple_send_receive_test.run() |
| self.assertTrue(simple_send_receive_test.message_received) |
| |
| self.node.delete(CONFIG_ADDRESS, name='myConfigAddr') |
| self.assertRaises(NotFoundStatus, self.node.read, |
| type=CONFIG_ADDRESS, name='myConfigAddr') |
| |
| def test_create_config_address_pattern(self): |
| self.assert_create_ok(CONFIG_ADDRESS, 'patternAddr', dict(pattern='a.*.b')) |
| self.assert_read_ok(CONFIG_ADDRESS, 'patternAddr', |
| dict(prefix=None, pattern='a.*.b')) |
| simple_send_receive_test = SimpleSndRecv(self.router.addresses[0], '/a.HITHERE.b') |
| simple_send_receive_test.run() |
| self.assertTrue(simple_send_receive_test.message_received) |
| |
| self.node.delete(CONFIG_ADDRESS, name='patternAddr') |
| self.assertRaises(NotFoundStatus, self.node.read, |
| type=CONFIG_ADDRESS, name='patternAddr') |
| |
| def test_dummy(self): |
| """Test all operations on the dummy test entity""" |
| entity = self.node.read(type=LISTENER, name='l0') |
| self.assertEqual('l0', entity.name) |
| self.assertEqual(str(self.router.ports[0]), entity.port) |
| |
| entity = self.node.read( |
| type=LISTENER, identity='listener/0.0.0.0:%s:l1' % self.router.ports[1]) |
| self.assertEqual('l1', entity.name) |
| self.assertEqual(str(self.router.ports[1]), entity.port) |
| |
| # Bad type |
| self.assertRaises(BadRequestStatus, self.node.read, type=CONNECTOR, name='l0') |
| |
| # Unknown entity |
| self.assertRaises(NotFoundStatus, self.node.read, type=LISTENER, name='nosuch') |
| |
| # Update is not allowed by the schema |
| self.assertRaises(NotImplementedStatus, entity.update) |
| |
| # Non-standard request is not allowed by schema. |
| self.assertRaises(NotImplementedStatus, entity.call, 'nosuchop', foo="bar") |
| |
| # Dummy entity supports all CRUD operations |
| dummy = self.node.create({'arg1': 'START'}, type=DUMMY, name='MyDummy', ) |
| self.assertEqual(dummy.type, DUMMY) |
| self.assertEqual(dummy.name, 'MyDummy') |
| self.assertEqual(dummy.arg1, 'START') |
| identity = dummy.identity |
| self.assertEqual( |
| dict(type=DUMMY, identity=identity, name='MyDummy', arg1='START'), |
| dummy.attributes) |
| |
| dummy.attributes['num1'] = 42 |
| dummy.arg1 = 'one' |
| self.assertEqual( |
| dict(type=DUMMY, identity=identity, name='MyDummy', arg1='one', num1=42), |
| dummy.attributes) |
| dummy.update() |
| |
| dummy.attributes.update(dict(arg1='x', num1=0)) |
| dummy.read() |
| self.assertEqual( |
| dict(type=DUMMY, name='MyDummy', identity=identity, arg1='one', num1=42), |
| dummy.attributes) |
| |
| dummy2 = self.node.read(type=DUMMY, name='MyDummy') |
| self.assertEqual(dummy.attributes, dummy2.attributes) |
| |
| integers = [0, 1, 42, (2**63) - 1, -1, -42, -(2**63)] |
| test_data = [BINARY("bytes"), "string"] + integers |
| for data in test_data: |
| try: |
| self.assertEqual( |
| {'operation': 'callme', 'type': DUMMY, 'identity': identity, 'data': data}, |
| dummy.call('callme', data=data)) |
| except TypeError as exc: |
| raise TypeError("data=%r: %s" % (data, exc)) |
| |
| dummy.badattribute = 'Bad' |
| self.assertRaises(BadRequestStatus, dummy.update) |
| |
| dummy.delete() |
| self.assertRaises(NotFoundStatus, self.node.read, type=DUMMY, name='MyDummy') |
| |
| def test_link(self): |
| """Verify we can find our own reply-to address in links""" |
| response = self.node.query(type=LINK) |
| path = self.node.reply_to.split('/')[-1] |
| mylink = [l for l in response.get_dicts() |
| if l['owningAddr'] and l['owningAddr'].endswith(path)] |
| self.assertTrue(mylink) |
| |
| def test_connection(self): |
| """Verify there is at least one connection""" |
| response = self.node.query(type='org.apache.qpid.dispatch.connection') |
| self.assertTrue(response.results) |
| |
| def test_router(self): |
| """Verify router counts match entity counts""" |
| entities = self.node.query().get_entities() |
| routers = [e for e in entities if e.type == ROUTER] |
| self.assertEqual(1, len(routers)) |
| router = routers[0] |
| self.assertEqual(router.linkCount, len([e for e in entities if e.type == LINK])) |
| self.assertEqual(router.addrCount, len([e for e in entities if e.type == ADDRESS])) |
| |
| def test_router_node(self): |
| """Test node entity in a trio of linked routers""" |
| nodes = [self.cleanup(Node.connect(Url(r.addresses[0]))) for r in self.routers] |
| rnode_lists = [n.query(type=NODE).get_dicts() for n in nodes] |
| |
| def check(attrs): |
| name = attrs['id'] |
| self.assertEqual(attrs['identity'], 'router.node/%s' % name) |
| self.assertEqual(attrs['name'], 'router.node/%s' % name) |
| self.assertEqual(attrs['type'], 'org.apache.qpid.dispatch.router.node') |
| self.assertEqual(attrs['address'], 'amqp:/_topo/0/%s' % name) |
| return name |
| |
| self.assertEqual({"router0", "router1", "router2"}, {check(n) for n in rnode_lists[0]}) |
| self.assertEqual({"router0", "router1", "router2"}, {check(n) for n in rnode_lists[1]}) |
| self.assertEqual({"router0", "router1", "router2"}, {check(n) for n in rnode_lists[2]}) |
| |
| def test_entity_names(self): |
| nodes = [self.cleanup(Node.connect(Url(r.addresses[0]))) for r in self.routers] |
| # Test that all entities have a consitent identity format: type/name |
| entities = list(chain( |
| *[n.query(attribute_names=['type', 'identity', 'name']).iter_entities() for n in nodes])) |
| for e in entities: |
| if e.type == MANAGEMENT: |
| self.assertEqual(e.identity, "self") |
| else: |
| if e.type == 'org.apache.qpid.dispatch.connection': |
| # This will make sure that the identity of the connection object is always numeric |
| self.assertRegex(str(e.identity), "[1-9]+", e) |
| else: |
| self.assertRegex(e.identity, "^%s/" % short_name(e.type), e) |
| |
| def test_remote_node(self): |
| """Test that we can access management info of remote nodes using get_mgmt_nodes addresses""" |
| nodes = [self.cleanup(Node.connect(Url(r.addresses[0]))) for r in self.routers] |
| remotes = sum([n.get_mgmt_nodes() for n in nodes], []) |
| self.assertEqual({'amqp:/_topo/0/router%s/$management' % i for i in [0, 1, 2]}, |
| set(remotes)) |
| self.assertEqual(9, len(remotes)) |
| # Query router2 indirectly via router1 |
| remote_url = Url(self.routers[0].addresses[0], path=Url(remotes[0]).path) |
| remote = self.cleanup(Node.connect(remote_url)) |
| router_id = remotes[0].split("/")[3] |
| assert router_id in ['router0', 'router1', 'router2'] |
| self.assertEqual([router_id], [r.id for r in remote.query(type=ROUTER).get_entities()]) |
| |
| def test_get_types(self): |
| types = self.node.get_types() |
| self.assertIn(CONFIGURATION, types[LISTENER]) |
| self.assertIn(OPERATIONAL, types[LINK]) |
| |
| def test_get_operations(self): |
| result = self.node.get_operations(type=DUMMY) |
| self.assertEqual({DUMMY: ["CREATE", "READ", "UPDATE", "DELETE", "CALLME"]}, result) |
| result = self.node.get_operations() |
| for type in LISTENER, LINK: |
| self.assertIn(type, result) |
| self.assertEqual(["UPDATE", "READ"], result[LINK]) |
| |
| def test_get_attributes(self): |
| result = self.node.get_attributes(type=DUMMY) |
| self.assertEqual({'arg1', 'arg2', 'num1', 'num2', 'name', 'identity', 'type'}, |
| set(result[DUMMY])) |
| result = self.node.get_attributes() |
| for type in LISTENER, LINK: |
| self.assertIn(type, result) |
| for a in ['linkType', 'linkDir', 'owningAddr']: |
| self.assertIn(a, result[LINK]) |
| |
| def test_standalone_no_inter_router(self): |
| """Verify that we do not allow inter-router connectors or listeners in standalone mode""" |
| |
| attrs = dict(role="inter-router", saslMechanisms="ANONYMOUS") |
| self.assertRaises( |
| BadRequestStatus, |
| self.node.create, dict(attrs, type=LISTENER, name="bad1", port=str(self.get_port()))) |
| |
| self.assertRaises( |
| BadRequestStatus, |
| self.node.create, dict(attrs, type=CONNECTOR, name="bad2", port=str(self.get_port()))) |
| |
| conf = Qdrouterd.Config([ |
| ('router', {'mode': 'standalone', 'id': 'all_by_myself1'}), |
| ('listener', {'port': self.get_port(), 'role': 'inter-router'}) |
| ]) |
| r = self.qdrouterd('routerX', conf, wait=False) |
| r.expect = Process.EXIT_FAIL |
| self.assertTrue(r.wait() != 0) |
| |
| conf = Qdrouterd.Config([ |
| ('router', {'mode': 'standalone', 'id': 'all_by_myself2'}), |
| ('listener', {'port': self.get_port(), 'role': 'normal'}), |
| ('connector', {'port': self.get_port(), 'role': 'inter-router'}) |
| ]) |
| r = self.qdrouterd('routerY', conf, wait=False) |
| r.expect = Process.EXIT_FAIL |
| self.assertTrue(r.wait() != 0) |
| |
| def test_get_schema(self): |
| schema = dictify(QdSchema().dump()) |
| got = self.node.call(self.node.request(operation="GET-JSON-SCHEMA", identity="self")).body |
| self.assertEqual(schema, dictify(json.loads(got))) |
| got = self.node.call(self.node.request(operation="GET-SCHEMA", identity="self")).body |
| self.assertEqual(schema, got) |
| |
| |
| class SimpleSndRecv(MessagingHandler): |
| def __init__(self, conn_address, address): |
| super(SimpleSndRecv, self).__init__() |
| self.conn_address = conn_address |
| self.address = address |
| self.sender = None |
| self.receiver = None |
| self.conn = None |
| self.message_received = False |
| |
| def on_start(self, event): |
| self.conn = event.container.connect(self.conn_address) |
| self.receiver = event.container.create_receiver(self.conn, self.address) |
| self.sender = event.container.create_sender(self.conn, self.address) |
| |
| def on_sendable(self, event): |
| msg = Message(body="Hello World") |
| event.sender.send(msg) |
| |
| def on_message(self, event): |
| if "Hello World" == event.message.body: |
| self.message_received = True |
| self.conn.close() |
| |
| def run(self): |
| Container(self).run() |
| |
| |
| if __name__ == '__main__': |
| unittest.main(system_test.main_module()) |
