| # |
| # 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. |
| # |
| |
| import argparse, sys |
| from itertools import combinations |
| |
| from qpid_dispatch_internal.tools.command import (main, |
| UsageError, |
| parse_args_qdstat, |
| parse_args_qdmanage, |
| _qdmanage_parser, |
| _qdstat_parser, |
| version_supports_mutually_exclusive_arguments) |
| |
| from system_test import unittest |
| |
| def mock_error(self, message): |
| raise ValueError(message) |
| |
| argparse.ArgumentParser.error = mock_error |
| |
| # Since BusManager file is definded in tools/qdmanage.in -> tools/qdmanage |
| # otherwise it could be just imported |
| class FakeBusManager: |
| def displayGeneral(self): pass |
| def displayConnections(self): pass |
| def displayRouterLinks(self): pass |
| def displayRouterNodes(self): pass |
| def displayEdges(self): pass |
| def displayAddresses(self): pass |
| def displayMemory(self): pass |
| def displayAutolinks(self): pass |
| def displayLinkRoutes(self): pass |
| def displayLog(self): pass |
| def show_all(self): pass |
| |
| FBM = FakeBusManager |
| |
| class TestParseArgsQdstat(unittest.TestCase): |
| def setUp(self): |
| self.parser = _qdstat_parser(BusManager=FBM) |
| |
| def test_parse_args_qdstat_print_help(self): |
| self.parser.print_help() |
| |
| def test_parse_args_qdstat_mutually_exclusive(self): |
| if not version_supports_mutually_exclusive_arguments(): |
| #unittest.skip also not supported |
| print("skipping: Mutually_exclusive options not working on python 2.6") |
| return |
| |
| options1 = ["-g", "-c", |
| "-l","-n","-e","-a","-m","--autolinks","--linkroutes","--log", |
| "--all-entities"] |
| options2 = ["-r", "--all-routers"] |
| |
| def _call_pairs(options): |
| for options_pair in combinations(options, 2): |
| with self.assertRaises(ValueError): |
| self.parser.parse_args(options_pair) |
| |
| _call_pairs(options1) |
| _call_pairs(options2) |
| |
| def test_parse_args_qdstat_default(self): |
| args = parse_args_qdstat(FBM, argv = []) |
| self.assertEqual(FBM.displayGeneral.__name__, args.show) |
| |
| def test_parse_args_qdstat_method_show_matching(self): |
| matching = [("-g", FBM.displayGeneral.__name__), |
| ("-c", FBM.displayConnections.__name__), |
| ("-l", FBM.displayRouterLinks.__name__), |
| ("-n", FBM.displayRouterNodes.__name__), |
| ("-e", FBM.displayEdges.__name__), |
| ("-a", FBM.displayAddresses.__name__), |
| ("-m", FBM.displayMemory.__name__), |
| ("--autolinks", FBM.displayAutolinks.__name__), |
| ("--linkroutes", FBM.displayLinkRoutes.__name__), |
| ("--log", FBM.displayLog.__name__), |
| ("--all-entities", FBM.show_all.__name__), |
| ] |
| for option, expected in matching: |
| args = self.parser.parse_args([option]) |
| self.assertEqual(expected, args.show) |
| |
| def test_parse_args_qdstat_limit(self): |
| args = self.parser.parse_args([]) |
| self.assertEqual(None, args.limit) |
| |
| args = self.parser.parse_args(["--limit", "1"]) |
| self.assertEqual(1, args.limit) |
| |
| class TestParseArgsQdmanage(unittest.TestCase): |
| def setUp(self): |
| self.operations = ["HERE", "SOME", "OPERATIONS"] |
| self.parser = _qdmanage_parser(operations=self.operations) |
| |
| def test_parse_args_qdmanage_print_help(self): |
| self.parser.print_help() |
| |
| def test_parse_args_qdmanage_operation_no_args(self): |
| argv = "-r r1 QUERY --type some --name the_name -b 127.0.0.1:5672" |
| opts, args = self.parser.parse_known_args(argv.split()) |
| self.assertEqual("QUERY", args[0]) |
| |
| def test_parse_args_qdmanage_operation_and_args(self): |
| argv = "-r r1 QUERY arg1=val1 --type some other=argument --name the_name -b 127.0.0.1:5672" |
| opts, args = self.parser.parse_known_args(argv.split()) |
| self.assertEqual(["QUERY", "arg1=val1", "other=argument"], args) |
| |
| |
| class TestMain(unittest.TestCase): |
| def test_main(self): |
| def run_success(argv): |
| self.assertEqual(sys.argv, argv) |
| |
| def run_raises(argv, _Exception): |
| run_success(argv) |
| raise _Exception("some") |
| |
| def run_raises_UsageError(argv): |
| run_raises(argv, UsageError) |
| |
| def run_raises_Exception(argv): |
| run_raises(argv, Exception) |
| |
| def run_raises_KeyboardInterrupt(argv): |
| run_raises(argv, KeyboardInterrupt) |
| |
| self.assertEqual(0, main(run_success)) |
| failed_runs = [ |
| #run_raises_UsageError, ##uncomment this exposes bug |
| run_raises_Exception, |
| run_raises_KeyboardInterrupt, |
| ] |
| for run in failed_runs: |
| self.assertEqual(1, main(run)) |
| |
| if __name__ == '__main__': |
| unittest.main() |
| |