diff --git a/CHANGES.rst b/CHANGES.rst index f6628cd5..d986132c 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -10,6 +10,8 @@ Unreleased Changes * Stop referencing deprecated ``botocore.vendored.requests.exceptions.ConnectTimeout`` in favor of new, and higher-level, ``botocore.exceptions.ConnectionError`` * In :py:meth:`awslimitchecker.utils._get_latest_version`, replace use of ``botocore.vendored.requests`` with ``urllib3``. +* `Issue #324 `__ - Support loading :ref:`limit overrides ` and/or :ref:`threshold overrides ` from a JSON file either stored locally or in S3 via new ``--limit-override-json`` and ``--threshold-override-json`` CLI options. + 7.0.0 (2019-08-13) ------------------ diff --git a/awslimitchecker/runner.py b/awslimitchecker/runner.py index a4696255..6bb3e830 100644 --- a/awslimitchecker/runner.py +++ b/awslimitchecker/runner.py @@ -42,11 +42,17 @@ import logging import json import termcolor +import boto3 from .checker import AwsLimitChecker from .utils import StoreKeyValuePair, dict2cols from .limit import SOURCE_TA, SOURCE_API +try: + from urllib.parse import urlparse +except ImportError: + from urlparse import urlparse + logging.basicConfig(level=logging.WARNING) logger = logging.getLogger() @@ -125,6 +131,16 @@ def parse_args(self, argv): help='override a single AWS limit, specified in ' '"service_name/limit_name=value" format; can be ' 'specified multiple times.') + p.add_argument('--limit-override-json', action='store', type=str, + default=None, + help='Absolute or relative path, or s3:// URL, to a ' + 'JSON file specifying limit overrides. See docs ' + 'for expected format.') + p.add_argument('--threshold-override-json', action='store', type=str, + default=None, + help='Absolute or relative path, or s3:// URL, to a ' + 'JSON file specifying threshold overrides. See ' + 'docs for expected format.') p.add_argument('-u', '--show-usage', action='store_true', default=False, help='find and print the current usage of all AWS ' @@ -349,6 +365,38 @@ def set_limit_overrides(self, overrides): svc, limit = key.split('/') self.checker.set_limit_override(svc, limit, int(overrides[key])) + def load_json(self, path): + """Load JSON from either a local file or S3""" + if path.startswith('s3://'): + parsed = urlparse(path) + s3key = parsed.path.lstrip('/') + logger.debug( + 'Reading JSON from S3 bucket "%s" key "%s"', + parsed.netloc, s3key + ) + client = boto3.client('s3') + resp = client.get_object(Bucket=parsed.netloc, Key=s3key) + data = resp['Body'].read() + else: + logger.debug('Reading JSON from: %s', path) + with open(path, 'r') as fh: + data = fh.read() + if isinstance(data, type(b'')): + data = data.decode() + return json.loads(data) + + def set_limit_overrides_from_json(self, path): + j = self.load_json(path) + logger.debug('Limit overrides: %s', j) + self.checker.set_limit_overrides(j) + logger.debug('Done setting limit overrides from JSON.') + + def set_threshold_overrides_from_json(self, path): + j = self.load_json(path) + logger.debug('Threshold overrides: %s', j) + self.checker.set_threshold_overrides(j) + logger.debug('Done setting threshold overrides from JSON.') + def console_entry_point(self): args = self.parse_args(sys.argv[1:]) self.service_name = args.service @@ -398,6 +446,14 @@ def console_entry_point(self): for check in args.skip_check: self.skip_check.append(check) + if args.limit_override_json is not None: + self.set_limit_overrides_from_json(args.limit_override_json) + + if args.threshold_override_json is not None: + self.set_threshold_overrides_from_json( + args.threshold_override_json + ) + if len(args.limit) > 0: self.set_limit_overrides(args.limit) diff --git a/awslimitchecker/tests/test_runner.py b/awslimitchecker/tests/test_runner.py index 700b12f7..1a060d45 100644 --- a/awslimitchecker/tests/test_runner.py +++ b/awslimitchecker/tests/test_runner.py @@ -56,9 +56,9 @@ sys.version_info[0] < 3 or sys.version_info[0] == 3 and sys.version_info[1] < 4 ): - from mock import patch, call, Mock + from mock import patch, call, Mock, mock_open else: - from unittest.mock import patch, call, Mock + from unittest.mock import patch, call, Mock, mock_open def red(s): @@ -73,7 +73,7 @@ def yellow(s): pb = 'awslimitchecker.runner' -class TestAwsLimitCheckerRunner(object): +class RunnerTester(object): def setup(self): self.cls = Runner() @@ -85,6 +85,9 @@ def setup(self): version_str='1.2.3@mytag' ) + +class TestModule(RunnerTester): + def test_module_entry_point(self): with patch('%s.Runner' % pb) as mock_runner: console_entry_point() @@ -93,6 +96,9 @@ def test_module_entry_point(self): call().console_entry_point(), ] + +class TestInit(RunnerTester): + def test_init(self): assert self.cls.colorize is True assert self.cls.checker is None @@ -100,14 +106,20 @@ def test_init(self): assert self.cls.service_name is None assert len(self.cls.skip_check) == 0 - def test_parse_args(self): + +class TestParseArgs(RunnerTester): + + def test_simple(self): argv = ['-V'] res = self.cls.parse_args(argv) assert isinstance(res, argparse.Namespace) assert res.version is True assert res.ta_refresh_mode is None + assert res.limit == {} + assert res.limit_override_json is None + assert res.threshold_override_json is None - def test_parse_args_parser(self): + def test_parser(self): argv = ['-V'] desc = 'Report on AWS service limits and usage via boto3, optionally ' \ 'warn about any services with usage nearing or exceeding ' \ @@ -153,6 +165,18 @@ def test_parse_args_parser(self): help='override a single AWS limit, specified in' ' "service_name/limit_name=value" format; can ' 'be specified multiple times.'), + call().add_argument('--limit-override-json', action='store', + type=str, default=None, + help='Absolute or relative path, or s3:// URL, ' + 'to a JSON file specifying limit ' + 'overrides. See docs for expected format.' + ), + call().add_argument('--threshold-override-json', action='store', + type=str, default=None, + help='Absolute or relative path, or s3:// URL,' + ' to a JSON file specifying threshold ' + 'overrides. See docs for expected format.' + ), call().add_argument('-u', '--show-usage', action='store_true', default=False, help='find and print the current usage of ' @@ -248,40 +272,40 @@ def test_parse_args_parser(self): call().parse_args(argv) ] - def test_parse_args_multiple_ta(self): + def test_multiple_ta(self): argv = ['--ta-refresh-wait', '--ta-refresh-older=100'] with pytest.raises(SystemExit): self.cls.parse_args(argv) - def test_parse_args_ta_refresh_wait(self): + def test_ta_refresh_wait(self): argv = ['--ta-refresh-wait'] res = self.cls.parse_args(argv) assert isinstance(res, argparse.Namespace) assert res.ta_refresh_mode == 'wait' - def test_parse_args_ta_refresh_trigger(self): + def test_ta_refresh_trigger(self): argv = ['--ta-refresh-trigger'] res = self.cls.parse_args(argv) assert isinstance(res, argparse.Namespace) assert res.ta_refresh_mode == 'trigger' - def test_parse_args_ta_refresh_older(self): + def test_ta_refresh_older(self): argv = ['--ta-refresh-older=123'] res = self.cls.parse_args(argv) assert isinstance(res, argparse.Namespace) assert res.ta_refresh_mode == 123 - def test_parse_args_skip_service_none(self): + def test_skip_service_none(self): argv = [] res = self.cls.parse_args(argv) assert res.skip_service == [] - def test_parse_args_skip_service_one(self): + def test_skip_service_one(self): argv = ['--skip-service', 'foo'] res = self.cls.parse_args(argv) assert res.skip_service == ['foo'] - def test_parse_args_skip_service_multiple(self): + def test_skip_service_multiple(self): argv = [ '--skip-service', 'foo', '--skip-service', 'bar', @@ -290,14 +314,14 @@ def test_parse_args_skip_service_multiple(self): res = self.cls.parse_args(argv) assert res.skip_service == ['foo', 'bar', 'baz'] - def test_parse_args_skip_check(self): + def test_skip_check(self): argv = [ '--skip-check', 'EC2/Running On-Demand x1e.8xlarge instances', ] res = self.cls.parse_args(argv) assert res.skip_check == ['EC2/Running On-Demand x1e.8xlarge instances'] - def test_parse_args_skip_check_multiple(self): + def test_skip_check_multiple(self): argv = [ '--skip-check', 'EC2/Running On-Demand x1e.8xlarge instances', '--skip-check', 'EC2/Running On-Demand c5.9xlarge instances', @@ -308,51 +332,10 @@ def test_parse_args_skip_check_multiple(self): 'EC2/Running On-Demand c5.9xlarge instances', ] - def test_entry_version(self, capsys): - argv = ['awslimitchecker', '-V'] - expected = 'awslimitchecker ver (see for source code)\n' - with patch.object(sys, 'argv', argv): - with patch('%s.AwsLimitChecker' % pb, - spec_set=AwsLimitChecker) as mock_alc: - mock_alc.return_value.get_project_url.return_value = 'foo' - mock_alc.return_value.get_version.return_value = 'ver' - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == expected - assert excinfo.value.code == 0 - assert mock_alc.mock_calls == [ - call( - warning_threshold=80, - critical_threshold=99, - account_id=None, - account_role=None, - region=None, - external_id=None, - mfa_serial_number=None, - mfa_token=None, - profile_name=None, - ta_refresh_mode=None, - ta_refresh_timeout=None, - check_version=True - ), - call().get_project_url(), - call().get_version() - ] - def test_entry_list_services(self): - argv = ['awslimitchecker', '-s'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.list_services' % pb, - autospec=True) as mock_list: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 0 - assert mock_list.mock_calls == [ - call(self.cls) - ] +class TestListServices(RunnerTester): - def test_list_services(self, capsys): + def test_happy_path(self, capsys): expected = 'Bar\nFoo\n' mock_checker = Mock(spec_set=AwsLimitChecker) mock_checker.get_service_names.return_value = [ @@ -367,18 +350,10 @@ def test_list_services(self, capsys): call.get_service_names() ] - def test_entry_iam_policy(self): - argv = ['awslimitchecker', '--iam-policy'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.iam_policy' % pb, autospec=True) as mock_iam: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 0 - assert mock_iam.mock_calls == [ - call(self.cls) - ] - def test_iam_policy(self, capsys): +class TestIamPolicy(RunnerTester): + + def test_happy_path(self, capsys): expected = {"baz": "blam", "foo": "bar"} mock_checker = Mock(spec_set=AwsLimitChecker) mock_checker.get_required_iam_policy.return_value = { @@ -393,19 +368,10 @@ def test_iam_policy(self, capsys): call.get_required_iam_policy() ] - def test_entry_list_defaults(self): - argv = ['awslimitchecker', '--list-defaults'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.list_defaults' % pb, - autospec=True) as mock_list: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 0 - assert mock_list.mock_calls == [ - call(self.cls) - ] - def test_list_defaults(self, capsys): +class TestListDefaults(RunnerTester): + + def test_simple(self, capsys): mock_checker = Mock(spec_set=AwsLimitChecker) mock_checker.get_limits.return_value = sample_limits() self.cls.checker = mock_checker @@ -426,7 +392,7 @@ def test_list_defaults(self, capsys): }) ] - def test_list_defaults_one_service(self, capsys): + def test_one_service(self, capsys): mock_checker = Mock(spec_set=AwsLimitChecker) mock_checker.get_limits.return_value = { 'SvcFoo': sample_limits()['SvcFoo'], @@ -448,19 +414,10 @@ def test_list_defaults_one_service(self, capsys): }) ] - def test_entry_list_limits(self): - argv = ['awslimitchecker', '-l'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.list_limits' % pb, - autospec=True) as mock_list: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 0 - assert mock_list.mock_calls == [ - call(self.cls) - ] - def test_list_limits(self, capsys): +class TestListLimits(RunnerTester): + + def test_simple(self, capsys): mock_checker = Mock(spec_set=AwsLimitChecker) mock_checker.get_limits.return_value = sample_limits_api() self.cls.checker = mock_checker @@ -482,7 +439,7 @@ def test_list_limits(self, capsys): }) ] - def test_list_limits_one_service(self, capsys): + def test_one_service(self, capsys): mock_checker = Mock(spec_set=AwsLimitChecker) mock_checker.get_limits.return_value = { 'SvcFoo': sample_limits_api()['SvcFoo'], @@ -505,146 +462,10 @@ def test_list_limits_one_service(self, capsys): }) ] - def test_entry_skip_service_none(self): - argv = ['awslimitchecker'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_check: - mock_check.return_value = 2 - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 2 - assert mock_c.mock_calls == [ - call(account_id=None, account_role=None, critical_threshold=99, - external_id=None, mfa_serial_number=None, mfa_token=None, - profile_name=None, region=None, ta_refresh_mode=None, - ta_refresh_timeout=None, warning_threshold=80, - check_version=True) - ] - - def test_entry_skip_service(self): - argv = ['awslimitchecker', '--skip-service=foo'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_check: - mock_check.return_value = 2 - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 2 - assert mock_c.mock_calls == [ - call(account_id=None, account_role=None, critical_threshold=99, - external_id=None, mfa_serial_number=None, mfa_token=None, - profile_name=None, region=None, ta_refresh_mode=None, - ta_refresh_timeout=None, warning_threshold=80, - check_version=True), - call().remove_services(['foo']) - ] - - def test_entry_skip_service_multi(self): - argv = [ - 'awslimitchecker', - '--skip-service=foo', - '--skip-service', 'bar' - ] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_check: - mock_check.return_value = 2 - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 2 - assert mock_c.mock_calls == [ - call(account_id=None, account_role=None, critical_threshold=99, - external_id=None, mfa_serial_number=None, mfa_token=None, - profile_name=None, region=None, ta_refresh_mode=None, - ta_refresh_timeout=None, warning_threshold=80, - check_version=True), - call().remove_services(['foo', 'bar']) - ] - - def test_entry_skip_check(self): - argv = [ - 'awslimitchecker', - '--skip-check=EC2/Max launch specifications per spot fleet' - ] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_check: - mock_check.return_value = 2 - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 2 - assert mock_c.mock_calls == [ - call(account_id=None, account_role=None, critical_threshold=99, - external_id=None, mfa_serial_number=None, mfa_token=None, - profile_name=None, region=None, ta_refresh_mode=None, - ta_refresh_timeout=None, warning_threshold=80, - check_version=True), - ] - assert self.cls.skip_check == [ - 'EC2/Max launch specifications per spot fleet', - ] - - def test_entry_skip_check_multi(self): - argv = [ - 'awslimitchecker', - '--skip-check=EC2/Max launch specifications per spot fleet', - '--skip-check', 'EC2/Running On-Demand i3.large instances', - ] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_check: - mock_check.return_value = 2 - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 2 - assert mock_c.mock_calls == [ - call(account_id=None, account_role=None, critical_threshold=99, - external_id=None, mfa_serial_number=None, mfa_token=None, - profile_name=None, region=None, ta_refresh_mode=None, - ta_refresh_timeout=None, warning_threshold=80, - check_version=True), - ] - assert self.cls.skip_check == [ - 'EC2/Max launch specifications per spot fleet', - 'EC2/Running On-Demand i3.large instances', - ] - - def test_entry_limit(self): - argv = ['awslimitchecker', '-L', 'foo=bar'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb) as mock_ct: - with patch('%s.Runner.set_limit_overrides' - '' % pb, autospec=True) as mock_slo: - mock_ct.return_value = 0 - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 0 - assert mock_slo.mock_calls == [ - call(self.cls, {'foo': 'bar'}) - ] - def test_entry_limit_multi(self): - argv = ['awslimitchecker', '--limit=foo=bar', '--limit=baz=blam'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with patch('%s.Runner.set_limit_overrides' - '' % pb, autospec=True) as mock_slo: - mock_ct.return_value = 0 - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 0 - assert mock_slo.mock_calls == [ - call(self.cls, {'foo': 'bar', 'baz': 'blam'}) - ] +class TestSetLimitOverride(RunnerTester): - def test_set_limit_overrides(self): + def test_simple(self): overrides = { 'EC2/Foo bar': "2", 'ElastiCache/Cache cluster subnet groups': "100", @@ -661,7 +482,7 @@ def test_set_limit_overrides(self): ) ] - def test_set_limit_overrides_error(self): + def test_error(self): overrides = { 'EC2': 2, } @@ -677,18 +498,171 @@ def test_set_limit_overrides_error(self): assert msg == "Limit names must be in " \ "'service/limit' format; EC2 is invalid." - def test_entry_show_usage(self): - argv = ['awslimitchecker', '-u'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.show_usage' % pb, autospec=True) as mock_show: - with pytest.raises(SystemExit) as excinfo: - self.cls.console_entry_point() - assert excinfo.value.code == 0 - assert mock_show.mock_calls == [ - call(self.cls) + +class TestLoadJson(RunnerTester): + + def test_local_file_py27(self): + data = u'{"Foo": {"bar": 23, "baz": 6}, "Blam": {"Blarg": 73}}' + mock_body = Mock() + mock_body.read.return_value = '{"Foo": {"bar": 23, "baz": 6}}' + mock_client = Mock() + mock_client.get_object.return_value = {'Body': mock_body} + with patch( + '%s.open' % pb, mock_open(read_data=data), create=True + ) as m_open: + with patch('%s.boto3.client' % pb) as m_client: + m_client.return_value = mock_client + res = self.cls.load_json('/foo/bar/baz.json') + assert m_open.mock_calls == [ + call('/foo/bar/baz.json', 'r'), + call().__enter__(), + call().read(), + call().__exit__(None, None, None) + ] + assert m_client.mock_calls == [] + assert res == { + 'Foo': {'bar': 23, 'baz': 6}, + 'Blam': {'Blarg': 73} + } + + def test_s3_py27(self): + data = '{"Foo": {"bar": 23, "baz": 6}, "Blam": {"Blarg": 73}}' + mock_body = Mock() + mock_body.read.return_value = data + mock_client = Mock() + mock_client.get_object.return_value = {'Body': mock_body} + with patch( + '%s.open' % pb, mock_open(read_data=data), create=True + ) as m_open: + with patch('%s.boto3.client' % pb) as m_client: + m_client.return_value = mock_client + res = self.cls.load_json( + 's3://bucketname/key/foo/bar/baz.json' + ) + assert m_open.mock_calls == [] + assert m_client.mock_calls == [ + call('s3'), + call().get_object(Bucket='bucketname', Key='key/foo/bar/baz.json') + ] + assert res == { + 'Foo': {'bar': 23, 'baz': 6}, + 'Blam': {'Blarg': 73} + } + + def test_local_file_py37(self): + data = '{"Foo": {"bar": 23, "baz": 6}, "Blam": {"Blarg": 73}}' + mock_body = Mock() + mock_body.read.return_value = '{"Foo": {"bar": 23, "baz": 6}}' + mock_client = Mock() + mock_client.get_object.return_value = {'Body': mock_body} + with patch( + '%s.open' % pb, mock_open(read_data=data), create=True + ) as m_open: + with patch('%s.boto3.client' % pb) as m_client: + m_client.return_value = mock_client + res = self.cls.load_json('/foo/bar/baz.json') + assert m_open.mock_calls == [ + call('/foo/bar/baz.json', 'r'), + call().__enter__(), + call().read(), + call().__exit__(None, None, None) + ] + assert m_client.mock_calls == [] + assert res == { + 'Foo': {'bar': 23, 'baz': 6}, + 'Blam': {'Blarg': 73} + } + + def test_s3_py37(self): + data = b'{"Foo": {"bar": 23, "baz": 6}, "Blam": {"Blarg": 73}}' + mock_body = Mock() + mock_body.read.return_value = data + mock_client = Mock() + mock_client.get_object.return_value = {'Body': mock_body} + with patch( + '%s.open' % pb, mock_open(read_data=data), create=True + ) as m_open: + with patch('%s.boto3.client' % pb) as m_client: + m_client.return_value = mock_client + res = self.cls.load_json( + 's3://bucketname/key/foo/bar/baz.json' + ) + assert m_open.mock_calls == [] + assert m_client.mock_calls == [ + call('s3'), + call().get_object(Bucket='bucketname', Key='key/foo/bar/baz.json') + ] + assert res == { + 'Foo': {'bar': 23, 'baz': 6}, + 'Blam': {'Blarg': 73} + } + + +class TestSetLimitOverridesFromJson(RunnerTester): + + def test_happy_path(self): + mock_checker = Mock(spec_set=AwsLimitChecker) + self.cls.checker = mock_checker + with patch('%s.Runner.load_json' % pb, autospec=True) as m_load: + m_load.return_value = { + 'Foo': {'bar': 23, 'baz': 6}, + 'Blam': {'Blarg': 73} + } + self.cls.set_limit_overrides_from_json('/foo/bar/baz.json') + assert m_load.mock_calls == [ + call(self.cls, '/foo/bar/baz.json') + ] + assert self.cls.checker.mock_calls == [ + call.set_limit_overrides({ + 'Foo': {'bar': 23, 'baz': 6}, + 'Blam': {'Blarg': 73} + }) + ] + + +class TestSetThresholdOverridesFromJson(RunnerTester): + + def test_happy_path(self): + mock_checker = Mock(spec_set=AwsLimitChecker) + self.cls.checker = mock_checker + with patch('%s.Runner.load_json' % pb, autospec=True) as m_load: + m_load.return_value = { + 'Foo': { + 'bar': { + 'warning': { + 'percent': 90, + 'count': 10 + }, + 'critical': { + 'percent': 95 + } + } + } + } + self.cls.set_threshold_overrides_from_json('/foo/bar/baz.json') + assert m_load.mock_calls == [ + call(self.cls, '/foo/bar/baz.json') + ] + assert self.cls.checker.mock_calls == [ + call.set_threshold_overrides({ + 'Foo': { + 'bar': { + 'warning': { + 'percent': 90, + 'count': 10 + }, + 'critical': { + 'percent': 95 + } + } + } + }) ] - def test_show_usage(self, capsys): + +class TestShowUsage(RunnerTester): + + def test_default(self, capsys): limits = sample_limits() limits['SvcFoo']['foo limit3']._add_current_usage(33) limits['SvcBar']['bar limit2']._add_current_usage(22) @@ -713,7 +687,7 @@ def test_show_usage(self, capsys): }) ] - def test_show_usage_one_service(self, capsys): + def test_one_service(self, capsys): limits = { 'SvcFoo': sample_limits()['SvcFoo'], } @@ -738,255 +712,387 @@ def test_show_usage_one_service(self, capsys): }) ] - def test_entry_skip_ta(self, capsys): - argv = ['awslimitchecker', '--skip-ta'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 6 - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.code == 6 - assert self.cls.skip_ta is True - def test_entry_service_name(self, capsys): - argv = ['awslimitchecker', '-S', 'foo'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 6 - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.code == 6 - assert self.cls.service_name == ['foo'] +class TestCheckThresholds(RunnerTester): - def test_entry_no_service_name(self, capsys): - argv = ['awslimitchecker'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 6 - self.cls.console_entry_point() + def test_ok(self, capsys): + """no problems, return 0 and print nothing""" + mock_checker = Mock(spec_set=AwsLimitChecker) + mock_checker.check_thresholds.return_value = {} + self.cls.checker = mock_checker + with patch('awslimitchecker.runner.dict2cols') as mock_d2c: + mock_d2c.return_value = '' + res = self.cls.check_thresholds() out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.code == 6 - assert self.cls.service_name is None + assert out == '\n' + assert mock_checker.mock_calls == [ + call.check_thresholds(use_ta=True, service=None) + ] + assert res == 0 + + def test_many_problems(self): + """lots of problems""" + mock_limit1 = Mock(spec_set=AwsLimit) + type(mock_limit1).name = 'limit1' + mock_w1 = Mock(spec_set=AwsLimitUsage) + mock_limit1.get_warnings.return_value = [mock_w1] + mock_c1 = Mock(spec_set=AwsLimitUsage) + mock_limit1.get_criticals.return_value = [mock_c1] + + mock_limit2 = Mock(spec_set=AwsLimit) + type(mock_limit2).name = 'limit2' + mock_w2 = Mock(spec_set=AwsLimitUsage) + mock_limit2.get_warnings.return_value = [mock_w2] + mock_limit2.get_criticals.return_value = [] + + mock_limit3 = Mock(spec_set=AwsLimit) + type(mock_limit3).name = 'limit3' + mock_w3 = Mock(spec_set=AwsLimitUsage) + mock_limit3.get_warnings.return_value = [mock_w3] + mock_limit3.get_criticals.return_value = [] + + mock_limit4 = Mock(spec_set=AwsLimit) + type(mock_limit4).name = 'limit4' + mock_limit4.get_warnings.return_value = [] + mock_c2 = Mock(spec_set=AwsLimitUsage) + mock_limit4.get_criticals.return_value = [mock_c2] + + mock_checker = Mock(spec_set=AwsLimitChecker) + mock_checker.check_thresholds.return_value = { + 'svc2': { + 'limit3': mock_limit3, + 'limit4': mock_limit4, + }, + 'svc1': { + 'limit1': mock_limit1, + 'limit2': mock_limit2, + }, + } + + def se_print(cls, s, l, c, w): + return ('{s}/{l}'.format(s=s, l=l.name), '') + + self.cls.checker = mock_checker + with patch('awslimitchecker.runner.Runner.print_issue', + autospec=True) as mock_print: + mock_print.side_effect = se_print + with patch('awslimitchecker.runner.dict2cols') as mock_d2c: + mock_d2c.return_value = 'd2cval' + res = self.cls.check_thresholds() + assert mock_checker.mock_calls == [ + call.check_thresholds(use_ta=True, service=None) + ] + assert mock_print.mock_calls == [ + call(self.cls, 'svc1', mock_limit1, [mock_c1], [mock_w1]), + call(self.cls, 'svc1', mock_limit2, [], [mock_w2]), + call(self.cls, 'svc2', mock_limit3, [], [mock_w3]), + call(self.cls, 'svc2', mock_limit4, [mock_c2], []), + ] + assert mock_d2c.mock_calls == [ + call({ + 'svc1/limit1': '', + 'svc1/limit2': '', + 'svc2/limit3': '', + 'svc2/limit4': '', + }) + ] + assert res == 2 + + def test_when_skip_check(self): + """lots of problems""" + mock_limit1 = Mock(spec_set=AwsLimit) + type(mock_limit1).name = 'limit1' + mock_w1 = Mock(spec_set=AwsLimitUsage) + mock_limit1.get_warnings.return_value = [mock_w1] + mock_c1 = Mock(spec_set=AwsLimitUsage) + mock_limit1.get_criticals.return_value = [mock_c1] + + mock_limit2 = Mock(spec_set=AwsLimit) + type(mock_limit2).name = 'limit2' + mock_w2 = Mock(spec_set=AwsLimitUsage) + mock_limit2.get_warnings.return_value = [mock_w2] + mock_limit2.get_criticals.return_value = [] + + mock_checker = Mock(spec_set=AwsLimitChecker) + mock_checker.check_thresholds.return_value = { + 'svc1': { + 'limit1': mock_limit1, + 'limit2': mock_limit2, + }, + } + + def se_print(cls, s, l, c, w): + return ('{s}/{l}'.format(s=s, l=l.name), '') + + self.cls.checker = mock_checker + self.cls.skip_check = ['svc1/limit1'] + with patch('awslimitchecker.runner.Runner.print_issue', + autospec=True) as mock_print: + mock_print.side_effect = se_print + with patch('awslimitchecker.runner.dict2cols') as mock_d2c: + mock_d2c.return_value = 'd2cval' + res = self.cls.check_thresholds() + + assert mock_checker.mock_calls == [ + call.check_thresholds(use_ta=True, service=None) + ] + assert mock_print.mock_calls == [ + call(self.cls, 'svc1', mock_limit2, [], [mock_w2]), + ] + assert mock_d2c.mock_calls == [ + call({ + 'svc1/limit2': '', + }) + ] + assert res == 1 + + def test_warn(self): + """just warnings""" + mock_limit1 = Mock(spec_set=AwsLimit) + mock_w1 = Mock(spec_set=AwsLimitUsage) + mock_w2 = Mock(spec_set=AwsLimitUsage) + mock_limit1.get_warnings.return_value = [mock_w1, mock_w2] + mock_limit1.get_criticals.return_value = [] + + mock_limit2 = Mock(spec_set=AwsLimit) + mock_w3 = Mock(spec_set=AwsLimitUsage) + mock_limit2.get_warnings.return_value = [mock_w3] + mock_limit2.get_criticals.return_value = [] + + mock_checker = Mock(spec_set=AwsLimitChecker) + mock_checker.check_thresholds.return_value = { + 'svc2': { + 'limit2': mock_limit2, + }, + 'svc1': { + 'limit1': mock_limit1, + }, + } + + self.cls.checker = mock_checker + with patch('awslimitchecker.runner.Runner.print_issue', + autospec=True) as mock_print: + mock_print.return_value = ('', '') + with patch('awslimitchecker.runner.dict2cols') as mock_d2c: + mock_d2c.return_value = 'd2cval' + res = self.cls.check_thresholds() + assert mock_checker.mock_calls == [ + call.check_thresholds(use_ta=True, service=None) + ] + assert mock_print.mock_calls == [ + call(self.cls, 'svc1', mock_limit1, [], [mock_w1, mock_w2]), + call(self.cls, 'svc2', mock_limit2, [], [mock_w3]), + ] + assert res == 1 + + def test_warn_one_service(self): + """just warnings""" + mock_limit1 = Mock(spec_set=AwsLimit) + mock_w1 = Mock(spec_set=AwsLimitUsage) + mock_w2 = Mock(spec_set=AwsLimitUsage) + mock_limit1.get_warnings.return_value = [mock_w1, mock_w2] + mock_limit1.get_criticals.return_value = [] + + mock_limit2 = Mock(spec_set=AwsLimit) + mock_w3 = Mock(spec_set=AwsLimitUsage) + mock_limit2.get_warnings.return_value = [mock_w3] + mock_limit2.get_criticals.return_value = [] + + mock_checker = Mock(spec_set=AwsLimitChecker) + mock_checker.check_thresholds.return_value = { + 'svc2': { + 'limit2': mock_limit2, + }, + } + + self.cls.checker = mock_checker + self.cls.service_name = ['svc2'] + with patch('awslimitchecker.runner.Runner.print_issue', + autospec=True) as mock_print: + mock_print.return_value = ('', '') + with patch('awslimitchecker.runner.dict2cols') as mock_d2c: + mock_d2c.return_value = 'd2cval' + res = self.cls.check_thresholds() + assert mock_checker.mock_calls == [ + call.check_thresholds(use_ta=True, service=['svc2']) + ] + assert mock_print.mock_calls == [ + call(self.cls, 'svc2', mock_limit2, [], [mock_w3]), + ] + assert res == 1 + + def test_crit(self): + """only critical""" + mock_limit1 = Mock(spec_set=AwsLimit) + mock_limit1.get_warnings.return_value = [] + mock_c1 = Mock(spec_set=AwsLimitUsage) + mock_c2 = Mock(spec_set=AwsLimitUsage) + mock_limit1.get_criticals.return_value = [mock_c1, mock_c2] + + mock_checker = Mock(spec_set=AwsLimitChecker) + mock_checker.check_thresholds.return_value = { + 'svc1': { + 'limit1': mock_limit1, + }, + } + + self.cls.checker = mock_checker + self.cls.skip_ta = True + with patch('awslimitchecker.runner.Runner.print_issue', + autospec=True) as mock_print: + mock_print.return_value = ('', '') + with patch('awslimitchecker.runner.dict2cols') as mock_d2c: + mock_d2c.return_value = 'd2cval' + res = self.cls.check_thresholds() + assert mock_checker.mock_calls == [ + call.check_thresholds(use_ta=False, service=None) + ] + assert mock_print.mock_calls == [ + call(self.cls, 'svc1', mock_limit1, [mock_c1, mock_c2], []), + ] + assert res == 2 + + +class TestPrintIssue(RunnerTester): + + def test_crit_one(self): + mock_limit = Mock(spec_set=AwsLimit) + type(mock_limit).name = 'limitname' + mock_limit.get_limit.return_value = 12 + + c1 = AwsLimitUsage(mock_limit, 56) + + res = self.cls.print_issue( + 'svcname', + mock_limit, + [c1], + [] + ) + assert res == ('svcname/limitname', + '(limit 12) ' + red('CRITICAL: 56')) + + def test_crit_multi(self): + mock_limit = Mock(spec_set=AwsLimit) + type(mock_limit).name = 'limitname' + mock_limit.get_limit.return_value = 5 + + c1 = AwsLimitUsage(mock_limit, 10) + c2 = AwsLimitUsage(mock_limit, 12, resource_id='c2id') + c3 = AwsLimitUsage(mock_limit, 8) + + res = self.cls.print_issue( + 'svcname', + mock_limit, + [c1, c2, c3], + [] + ) + assert res == ('svcname/limitname', + '(limit 5) ' + red('CRITICAL: 8, 10, c2id=12')) + + def test_warn_one(self): + mock_limit = Mock(spec_set=AwsLimit) + type(mock_limit).name = 'limitname' + mock_limit.get_limit.return_value = 12 + + w1 = AwsLimitUsage(mock_limit, 11) + + res = self.cls.print_issue( + 'svcname', + mock_limit, + [], + [w1] + ) + assert res == ('svcname/limitname', '(limit 12) ' + + yellow('WARNING: 11')) + + def test_warn_multi(self): + mock_limit = Mock(spec_set=AwsLimit) + type(mock_limit).name = 'limitname' + mock_limit.get_limit.return_value = 12 + + w1 = AwsLimitUsage(mock_limit, 11) + w2 = AwsLimitUsage(mock_limit, 10, resource_id='w2id') + w3 = AwsLimitUsage(mock_limit, 10, resource_id='w3id') + + res = self.cls.print_issue( + 'svcname', + mock_limit, + [], + [w1, w2, w3] + ) + assert res == ('svcname/limitname', + '(limit 12) ' + yellow('WARNING: ' + 'w2id=10, w3id=10, 11')) + + def test_both_one(self): + mock_limit = Mock(spec_set=AwsLimit) + type(mock_limit).name = 'limitname' + mock_limit.get_limit.return_value = 12 + + c1 = AwsLimitUsage(mock_limit, 10) + w1 = AwsLimitUsage(mock_limit, 10, resource_id='w3id') + + res = self.cls.print_issue( + 'svcname', + mock_limit, + [c1], + [w1] + ) + assert res == ('svcname/limitname', + '(limit 12) ' + + red('CRITICAL: 10') + ' ' + + yellow('WARNING: w3id=10')) - def test_entry_no_service_name_region(self, capsys): - argv = ['awslimitchecker', '-r', 'myregion'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with patch('%s.AwsLimitChecker' % pb, - spec_set=AwsLimitChecker) as mock_alc: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 6 - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.code == 6 - assert mock_alc.mock_calls == [ - call( - warning_threshold=80, - critical_threshold=99, - account_id=None, - account_role=None, - region='myregion', - external_id=None, - mfa_serial_number=None, - mfa_token=None, - profile_name=None, - ta_refresh_mode=None, - ta_refresh_timeout=None, - check_version=True - ) - ] - assert self.cls.service_name is None + def test_both_multi(self): + mock_limit = Mock(spec_set=AwsLimit) + type(mock_limit).name = 'limitname' + mock_limit.get_limit.return_value = 12 - def test_entry_no_service_name_sts(self, capsys): - argv = [ - 'awslimitchecker', - '-r', - 'myregion', - '-A', - '098765432109', - '-R', - 'myrole' - ] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with patch('%s.AwsLimitChecker' % pb, - spec_set=AwsLimitChecker) as mock_alc: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 6 - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.code == 6 - assert mock_alc.mock_calls == [ - call( - warning_threshold=80, - critical_threshold=99, - account_id='098765432109', - account_role='myrole', - region='myregion', - external_id=None, - mfa_serial_number=None, - mfa_token=None, - profile_name=None, - ta_refresh_mode=None, - ta_refresh_timeout=None, - check_version=True - ) - ] - assert self.cls.service_name is None + c1 = AwsLimitUsage(mock_limit, 10) + c2 = AwsLimitUsage(mock_limit, 12, resource_id='c2id') + c3 = AwsLimitUsage(mock_limit, 8) + w1 = AwsLimitUsage(mock_limit, 11) + w2 = AwsLimitUsage(mock_limit, 10, resource_id='w2id') + w3 = AwsLimitUsage(mock_limit, 10, resource_id='w3id') - def test_entry_no_service_name_sts_external_id(self, capsys): - argv = [ - 'awslimitchecker', - '-r', - 'myregion', - '-A', - '098765432109', - '-R', - 'myrole', - '-E', - 'myextid', - '--no-check-version' - ] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with patch('%s.AwsLimitChecker' % pb, - spec_set=AwsLimitChecker) as mock_alc: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 6 - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.code == 6 - assert mock_alc.mock_calls == [ - call( - warning_threshold=80, - critical_threshold=99, - account_id='098765432109', - account_role='myrole', - region='myregion', - external_id='myextid', - mfa_serial_number=None, - mfa_token=None, - profile_name=None, - ta_refresh_mode=None, - ta_refresh_timeout=None, - check_version=False - ) - ] - assert self.cls.service_name is None + res = self.cls.print_issue( + 'svcname', + mock_limit, + [c1, c2, c3], + [w1, w2, w3] + ) + assert res == ('svcname/limitname', + '(limit 12) ' + + red('CRITICAL: 8, 10, c2id=12') + ' ' + + yellow('WARNING: w2id=10, w3id=10, 11')) - def test_entry_verbose(self, capsys): - argv = ['awslimitchecker', '-v'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with patch('awslimitchecker.runner.logger.setLevel' - '') as mock_set_level: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 6 - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.code == 6 - assert mock_set_level.mock_calls == [call(logging.INFO)] - def test_entry_debug(self, capsys): - argv = ['awslimitchecker', '-vv'] - with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with patch('awslimitchecker.runner.logger.setLevel' - '') as mock_set_level: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 7 - self.cls.console_entry_point() - out, err = capsys.readouterr() - assert out == '' - assert excinfo.value.args[0] == 7 - assert mock_set_level.mock_calls == [call(logging.DEBUG)] +class TestColorOutput(RunnerTester): - def test_entry_warning(self): - argv = ['awslimitchecker', '-W', '50'] - with patch.object(sys, 'argv', argv): - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 8 - self.cls.console_entry_point() - assert excinfo.value.code == 8 - assert mock_alc.mock_calls == [ - call( - warning_threshold=50, - critical_threshold=99, - account_id=None, - account_role=None, - region=None, - external_id=None, - mfa_serial_number=None, - mfa_token=None, - profile_name=None, - ta_refresh_mode=None, - ta_refresh_timeout=None, - check_version=True - ) - ] + def test_simple(self): + assert self.cls.color_output('foo', 'yellow') == termcolor.colored( + 'foo', 'yellow') - def test_entry_warning_profile_name(self): - argv = ['awslimitchecker', '-W', '50', '-P', 'myprof'] - with patch.object(sys, 'argv', argv): - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 8 - self.cls.console_entry_point() - assert excinfo.value.code == 8 - assert mock_alc.mock_calls == [ - call( - warning_threshold=50, - critical_threshold=99, - account_id=None, - account_role=None, - region=None, - external_id=None, - mfa_serial_number=None, - mfa_token=None, - profile_name='myprof', - ta_refresh_mode=None, - ta_refresh_timeout=None, - check_version=True - ) - ] - def test_entry_critical(self): - argv = ['awslimitchecker', '-C', '95'] +class TestConsoleEntryPoint(RunnerTester): + + def test_version(self, capsys): + argv = ['awslimitchecker', '-V'] + expected = 'awslimitchecker ver (see for source code)\n' with patch.object(sys, 'argv', argv): - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 9 - self.cls.console_entry_point() - assert excinfo.value.code == 9 + with patch('%s.AwsLimitChecker' % pb, + spec_set=AwsLimitChecker) as mock_alc: + mock_alc.return_value.get_project_url.return_value = 'foo' + mock_alc.return_value.get_version.return_value = 'ver' + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == expected + assert excinfo.value.code == 0 assert mock_alc.mock_calls == [ call( warning_threshold=80, - critical_threshold=95, + critical_threshold=99, account_id=None, account_role=None, region=None, @@ -997,399 +1103,550 @@ def test_entry_critical(self): ta_refresh_mode=None, ta_refresh_timeout=None, check_version=True - ) - ] - - def test_entry_critical_ta_refresh(self): - argv = ['awslimitchecker', '-C', '95', '--ta-refresh-timeout=123', - '--ta-refresh-older=456'] - with patch.object(sys, 'argv', argv): - with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: - with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 9 - self.cls.console_entry_point() - assert excinfo.value.code == 9 - assert mock_alc.mock_calls == [ - call( - warning_threshold=80, - critical_threshold=95, - account_id=None, - account_role=None, - region=None, - external_id=None, - mfa_serial_number=None, - mfa_token=None, - profile_name=None, - ta_refresh_mode=456, - ta_refresh_timeout=123, - check_version=True - ) + ), + call().get_project_url(), + call().get_version() ] - def test_entry_check_thresholds(self): - argv = ['awslimitchecker'] + def test_list_services(self): + argv = ['awslimitchecker', '-s'] with patch.object(sys, 'argv', argv): - with patch('%s.Runner.check_thresholds' % pb, - autospec=True) as mock_ct: + with patch('%s.Runner.list_services' % pb, + autospec=True) as mock_list: with pytest.raises(SystemExit) as excinfo: - mock_ct.return_value = 10 self.cls.console_entry_point() - assert excinfo.value.code == 10 - assert mock_ct.mock_calls == [ + assert excinfo.value.code == 0 + assert mock_list.mock_calls == [ call(self.cls) ] - def test_check_thresholds_ok(self, capsys): - """no problems, return 0 and print nothing""" - mock_checker = Mock(spec_set=AwsLimitChecker) - mock_checker.check_thresholds.return_value = {} - self.cls.checker = mock_checker - with patch('awslimitchecker.runner.dict2cols') as mock_d2c: - mock_d2c.return_value = '' - res = self.cls.check_thresholds() - out, err = capsys.readouterr() - assert out == '\n' - assert mock_checker.mock_calls == [ - call.check_thresholds(use_ta=True, service=None) + def test_iam_policy(self): + argv = ['awslimitchecker', '--iam-policy'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.iam_policy' % pb, autospec=True) as mock_iam: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_iam.mock_calls == [ + call(self.cls) ] - assert res == 0 - - def test_check_thresholds_many_problems(self): - """lots of problems""" - mock_limit1 = Mock(spec_set=AwsLimit) - type(mock_limit1).name = 'limit1' - mock_w1 = Mock(spec_set=AwsLimitUsage) - mock_limit1.get_warnings.return_value = [mock_w1] - mock_c1 = Mock(spec_set=AwsLimitUsage) - mock_limit1.get_criticals.return_value = [mock_c1] - - mock_limit2 = Mock(spec_set=AwsLimit) - type(mock_limit2).name = 'limit2' - mock_w2 = Mock(spec_set=AwsLimitUsage) - mock_limit2.get_warnings.return_value = [mock_w2] - mock_limit2.get_criticals.return_value = [] - - mock_limit3 = Mock(spec_set=AwsLimit) - type(mock_limit3).name = 'limit3' - mock_w3 = Mock(spec_set=AwsLimitUsage) - mock_limit3.get_warnings.return_value = [mock_w3] - mock_limit3.get_criticals.return_value = [] - - mock_limit4 = Mock(spec_set=AwsLimit) - type(mock_limit4).name = 'limit4' - mock_limit4.get_warnings.return_value = [] - mock_c2 = Mock(spec_set=AwsLimitUsage) - mock_limit4.get_criticals.return_value = [mock_c2] - - mock_checker = Mock(spec_set=AwsLimitChecker) - mock_checker.check_thresholds.return_value = { - 'svc2': { - 'limit3': mock_limit3, - 'limit4': mock_limit4, - }, - 'svc1': { - 'limit1': mock_limit1, - 'limit2': mock_limit2, - }, - } - def se_print(cls, s, l, c, w): - return ('{s}/{l}'.format(s=s, l=l.name), '') - - self.cls.checker = mock_checker - with patch('awslimitchecker.runner.Runner.print_issue', - autospec=True) as mock_print: - mock_print.side_effect = se_print - with patch('awslimitchecker.runner.dict2cols') as mock_d2c: - mock_d2c.return_value = 'd2cval' - res = self.cls.check_thresholds() - assert mock_checker.mock_calls == [ - call.check_thresholds(use_ta=True, service=None) - ] - assert mock_print.mock_calls == [ - call(self.cls, 'svc1', mock_limit1, [mock_c1], [mock_w1]), - call(self.cls, 'svc1', mock_limit2, [], [mock_w2]), - call(self.cls, 'svc2', mock_limit3, [], [mock_w3]), - call(self.cls, 'svc2', mock_limit4, [mock_c2], []), - ] - assert mock_d2c.mock_calls == [ - call({ - 'svc1/limit1': '', - 'svc1/limit2': '', - 'svc2/limit3': '', - 'svc2/limit4': '', - }) + def test_list_defaults(self): + argv = ['awslimitchecker', '--list-defaults'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.list_defaults' % pb, + autospec=True) as mock_list: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_list.mock_calls == [ + call(self.cls) ] - assert res == 2 - - def test_check_thresholds_when_skip_check(self): - """lots of problems""" - mock_limit1 = Mock(spec_set=AwsLimit) - type(mock_limit1).name = 'limit1' - mock_w1 = Mock(spec_set=AwsLimitUsage) - mock_limit1.get_warnings.return_value = [mock_w1] - mock_c1 = Mock(spec_set=AwsLimitUsage) - mock_limit1.get_criticals.return_value = [mock_c1] - - mock_limit2 = Mock(spec_set=AwsLimit) - type(mock_limit2).name = 'limit2' - mock_w2 = Mock(spec_set=AwsLimitUsage) - mock_limit2.get_warnings.return_value = [mock_w2] - mock_limit2.get_criticals.return_value = [] - - mock_checker = Mock(spec_set=AwsLimitChecker) - mock_checker.check_thresholds.return_value = { - 'svc1': { - 'limit1': mock_limit1, - 'limit2': mock_limit2, - }, - } - def se_print(cls, s, l, c, w): - return ('{s}/{l}'.format(s=s, l=l.name), '') + def test_list_limits(self): + argv = ['awslimitchecker', '-l'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.list_limits' % pb, + autospec=True) as mock_list: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_list.mock_calls == [ + call(self.cls) + ] - self.cls.checker = mock_checker - self.cls.skip_check = ['svc1/limit1'] - with patch('awslimitchecker.runner.Runner.print_issue', - autospec=True) as mock_print: - mock_print.side_effect = se_print - with patch('awslimitchecker.runner.dict2cols') as mock_d2c: - mock_d2c.return_value = 'd2cval' - res = self.cls.check_thresholds() + def test_skip_service_none(self): + argv = ['awslimitchecker'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_check: + mock_check.return_value = 2 + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 2 + assert mock_c.mock_calls == [ + call(account_id=None, account_role=None, critical_threshold=99, + external_id=None, mfa_serial_number=None, mfa_token=None, + profile_name=None, region=None, ta_refresh_mode=None, + ta_refresh_timeout=None, warning_threshold=80, + check_version=True) + ] - assert mock_checker.mock_calls == [ - call.check_thresholds(use_ta=True, service=None) + def test_skip_service(self): + argv = ['awslimitchecker', '--skip-service=foo'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_check: + mock_check.return_value = 2 + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 2 + assert mock_c.mock_calls == [ + call(account_id=None, account_role=None, critical_threshold=99, + external_id=None, mfa_serial_number=None, mfa_token=None, + profile_name=None, region=None, ta_refresh_mode=None, + ta_refresh_timeout=None, warning_threshold=80, + check_version=True), + call().remove_services(['foo']) ] - assert mock_print.mock_calls == [ - call(self.cls, 'svc1', mock_limit2, [], [mock_w2]), + + def test_skip_service_multi(self): + argv = [ + 'awslimitchecker', + '--skip-service=foo', + '--skip-service', 'bar' ] - assert mock_d2c.mock_calls == [ - call({ - 'svc1/limit2': '', - }) + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_check: + mock_check.return_value = 2 + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 2 + assert mock_c.mock_calls == [ + call(account_id=None, account_role=None, critical_threshold=99, + external_id=None, mfa_serial_number=None, mfa_token=None, + profile_name=None, region=None, ta_refresh_mode=None, + ta_refresh_timeout=None, warning_threshold=80, + check_version=True), + call().remove_services(['foo', 'bar']) ] - assert res == 1 - - def test_check_thresholds_warn(self): - """just warnings""" - mock_limit1 = Mock(spec_set=AwsLimit) - mock_w1 = Mock(spec_set=AwsLimitUsage) - mock_w2 = Mock(spec_set=AwsLimitUsage) - mock_limit1.get_warnings.return_value = [mock_w1, mock_w2] - mock_limit1.get_criticals.return_value = [] - - mock_limit2 = Mock(spec_set=AwsLimit) - mock_w3 = Mock(spec_set=AwsLimitUsage) - mock_limit2.get_warnings.return_value = [mock_w3] - mock_limit2.get_criticals.return_value = [] - - mock_checker = Mock(spec_set=AwsLimitChecker) - mock_checker.check_thresholds.return_value = { - 'svc2': { - 'limit2': mock_limit2, - }, - 'svc1': { - 'limit1': mock_limit1, - }, - } - self.cls.checker = mock_checker - with patch('awslimitchecker.runner.Runner.print_issue', - autospec=True) as mock_print: - mock_print.return_value = ('', '') - with patch('awslimitchecker.runner.dict2cols') as mock_d2c: - mock_d2c.return_value = 'd2cval' - res = self.cls.check_thresholds() - assert mock_checker.mock_calls == [ - call.check_thresholds(use_ta=True, service=None) + def test_skip_check(self): + argv = [ + 'awslimitchecker', + '--skip-check=EC2/Max launch specifications per spot fleet' + ] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_check: + mock_check.return_value = 2 + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 2 + assert mock_c.mock_calls == [ + call(account_id=None, account_role=None, critical_threshold=99, + external_id=None, mfa_serial_number=None, mfa_token=None, + profile_name=None, region=None, ta_refresh_mode=None, + ta_refresh_timeout=None, warning_threshold=80, + check_version=True), ] - assert mock_print.mock_calls == [ - call(self.cls, 'svc1', mock_limit1, [], [mock_w1, mock_w2]), - call(self.cls, 'svc2', mock_limit2, [], [mock_w3]), + assert self.cls.skip_check == [ + 'EC2/Max launch specifications per spot fleet', ] - assert res == 1 - - def test_check_thresholds_warn_one_service(self): - """just warnings""" - mock_limit1 = Mock(spec_set=AwsLimit) - mock_w1 = Mock(spec_set=AwsLimitUsage) - mock_w2 = Mock(spec_set=AwsLimitUsage) - mock_limit1.get_warnings.return_value = [mock_w1, mock_w2] - mock_limit1.get_criticals.return_value = [] - - mock_limit2 = Mock(spec_set=AwsLimit) - mock_w3 = Mock(spec_set=AwsLimitUsage) - mock_limit2.get_warnings.return_value = [mock_w3] - mock_limit2.get_criticals.return_value = [] - - mock_checker = Mock(spec_set=AwsLimitChecker) - mock_checker.check_thresholds.return_value = { - 'svc2': { - 'limit2': mock_limit2, - }, - } - self.cls.checker = mock_checker - self.cls.service_name = ['svc2'] - with patch('awslimitchecker.runner.Runner.print_issue', - autospec=True) as mock_print: - mock_print.return_value = ('', '') - with patch('awslimitchecker.runner.dict2cols') as mock_d2c: - mock_d2c.return_value = 'd2cval' - res = self.cls.check_thresholds() - assert mock_checker.mock_calls == [ - call.check_thresholds(use_ta=True, service=['svc2']) + def test_skip_check_multi(self): + argv = [ + 'awslimitchecker', + '--skip-check=EC2/Max launch specifications per spot fleet', + '--skip-check', 'EC2/Running On-Demand i3.large instances', ] - assert mock_print.mock_calls == [ - call(self.cls, 'svc2', mock_limit2, [], [mock_w3]), + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_check: + mock_check.return_value = 2 + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_c: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 2 + assert mock_c.mock_calls == [ + call(account_id=None, account_role=None, critical_threshold=99, + external_id=None, mfa_serial_number=None, mfa_token=None, + profile_name=None, region=None, ta_refresh_mode=None, + ta_refresh_timeout=None, warning_threshold=80, + check_version=True), + ] + assert self.cls.skip_check == [ + 'EC2/Max launch specifications per spot fleet', + 'EC2/Running On-Demand i3.large instances', ] - assert res == 1 - def test_check_thresholds_crit(self): - """only critical""" - mock_limit1 = Mock(spec_set=AwsLimit) - mock_limit1.get_warnings.return_value = [] - mock_c1 = Mock(spec_set=AwsLimitUsage) - mock_c2 = Mock(spec_set=AwsLimitUsage) - mock_limit1.get_criticals.return_value = [mock_c1, mock_c2] + def test_limit(self): + argv = ['awslimitchecker', '-L', 'foo=bar'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb) as mock_ct: + with patch('%s.Runner.set_limit_overrides' + '' % pb, autospec=True) as mock_slo: + mock_ct.return_value = 0 + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_slo.mock_calls == [ + call(self.cls, {'foo': 'bar'}) + ] - mock_checker = Mock(spec_set=AwsLimitChecker) - mock_checker.check_thresholds.return_value = { - 'svc1': { - 'limit1': mock_limit1, - }, - } + def test_limit_multi(self): + argv = ['awslimitchecker', '--limit=foo=bar', '--limit=baz=blam'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with patch('%s.Runner.set_limit_overrides' + '' % pb, autospec=True) as mock_slo: + mock_ct.return_value = 0 + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_slo.mock_calls == [ + call(self.cls, {'foo': 'bar', 'baz': 'blam'}) + ] - self.cls.checker = mock_checker - self.cls.skip_ta = True - with patch('awslimitchecker.runner.Runner.print_issue', - autospec=True) as mock_print: - mock_print.return_value = ('', '') - with patch('awslimitchecker.runner.dict2cols') as mock_d2c: - mock_d2c.return_value = 'd2cval' - res = self.cls.check_thresholds() - assert mock_checker.mock_calls == [ - call.check_thresholds(use_ta=False, service=None) + def test_limit_json(self): + argv = [ + 'awslimitchecker', + '--limit-override-json=/path/to/file.json' ] - assert mock_print.mock_calls == [ - call(self.cls, 'svc1', mock_limit1, [mock_c1, mock_c2], []), + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb) as mock_ct: + with patch( + '%s.Runner.set_limit_overrides_from_json' % pb, + autospec=True + ) as mock_slo: + mock_ct.return_value = 0 + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_slo.mock_calls == [ + call(self.cls, '/path/to/file.json') ] - assert res == 2 - - def test_print_issue_crit_one(self): - mock_limit = Mock(spec_set=AwsLimit) - type(mock_limit).name = 'limitname' - mock_limit.get_limit.return_value = 12 - - c1 = AwsLimitUsage(mock_limit, 56) - - res = self.cls.print_issue( - 'svcname', - mock_limit, - [c1], - [] - ) - assert res == ('svcname/limitname', - '(limit 12) ' + red('CRITICAL: 56')) - def test_print_issue_crit_multi(self): - mock_limit = Mock(spec_set=AwsLimit) - type(mock_limit).name = 'limitname' - mock_limit.get_limit.return_value = 5 + def test_threshold_override_json(self): + argv = [ + 'awslimitchecker', + '--threshold-override-json=/path/to/file.json' + ] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb) as mock_ct: + with patch( + '%s.Runner.set_threshold_overrides_from_json' % pb, + autospec=True + ) as mock_tlo: + mock_ct.return_value = 0 + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_tlo.mock_calls == [ + call(self.cls, '/path/to/file.json') + ] - c1 = AwsLimitUsage(mock_limit, 10) - c2 = AwsLimitUsage(mock_limit, 12, resource_id='c2id') - c3 = AwsLimitUsage(mock_limit, 8) + def test_show_usage(self): + argv = ['awslimitchecker', '-u'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.show_usage' % pb, autospec=True) as mock_show: + with pytest.raises(SystemExit) as excinfo: + self.cls.console_entry_point() + assert excinfo.value.code == 0 + assert mock_show.mock_calls == [ + call(self.cls) + ] - res = self.cls.print_issue( - 'svcname', - mock_limit, - [c1, c2, c3], - [] - ) - assert res == ('svcname/limitname', - '(limit 5) ' + red('CRITICAL: 8, 10, c2id=12')) + def test_skip_ta(self, capsys): + argv = ['awslimitchecker', '--skip-ta'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 6 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.code == 6 + assert self.cls.skip_ta is True - def test_print_issue_warn_one(self): - mock_limit = Mock(spec_set=AwsLimit) - type(mock_limit).name = 'limitname' - mock_limit.get_limit.return_value = 12 + def test_service_name(self, capsys): + argv = ['awslimitchecker', '-S', 'foo'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 6 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.code == 6 + assert self.cls.service_name == ['foo'] - w1 = AwsLimitUsage(mock_limit, 11) + def test_no_service_name(self, capsys): + argv = ['awslimitchecker'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 6 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.code == 6 + assert self.cls.service_name is None - res = self.cls.print_issue( - 'svcname', - mock_limit, - [], - [w1] - ) - assert res == ('svcname/limitname', '(limit 12) ' + - yellow('WARNING: 11')) + def test_no_service_name_region(self, capsys): + argv = ['awslimitchecker', '-r', 'myregion'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with patch('%s.AwsLimitChecker' % pb, + spec_set=AwsLimitChecker) as mock_alc: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 6 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.code == 6 + assert mock_alc.mock_calls == [ + call( + warning_threshold=80, + critical_threshold=99, + account_id=None, + account_role=None, + region='myregion', + external_id=None, + mfa_serial_number=None, + mfa_token=None, + profile_name=None, + ta_refresh_mode=None, + ta_refresh_timeout=None, + check_version=True + ) + ] + assert self.cls.service_name is None - def test_print_issue_warn_multi(self): - mock_limit = Mock(spec_set=AwsLimit) - type(mock_limit).name = 'limitname' - mock_limit.get_limit.return_value = 12 + def test_no_service_name_sts(self, capsys): + argv = [ + 'awslimitchecker', + '-r', + 'myregion', + '-A', + '098765432109', + '-R', + 'myrole' + ] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with patch('%s.AwsLimitChecker' % pb, + spec_set=AwsLimitChecker) as mock_alc: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 6 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.code == 6 + assert mock_alc.mock_calls == [ + call( + warning_threshold=80, + critical_threshold=99, + account_id='098765432109', + account_role='myrole', + region='myregion', + external_id=None, + mfa_serial_number=None, + mfa_token=None, + profile_name=None, + ta_refresh_mode=None, + ta_refresh_timeout=None, + check_version=True + ) + ] + assert self.cls.service_name is None - w1 = AwsLimitUsage(mock_limit, 11) - w2 = AwsLimitUsage(mock_limit, 10, resource_id='w2id') - w3 = AwsLimitUsage(mock_limit, 10, resource_id='w3id') + def test_no_service_name_sts_external_id(self, capsys): + argv = [ + 'awslimitchecker', + '-r', + 'myregion', + '-A', + '098765432109', + '-R', + 'myrole', + '-E', + 'myextid', + '--no-check-version' + ] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with patch('%s.AwsLimitChecker' % pb, + spec_set=AwsLimitChecker) as mock_alc: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 6 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.code == 6 + assert mock_alc.mock_calls == [ + call( + warning_threshold=80, + critical_threshold=99, + account_id='098765432109', + account_role='myrole', + region='myregion', + external_id='myextid', + mfa_serial_number=None, + mfa_token=None, + profile_name=None, + ta_refresh_mode=None, + ta_refresh_timeout=None, + check_version=False + ) + ] + assert self.cls.service_name is None - res = self.cls.print_issue( - 'svcname', - mock_limit, - [], - [w1, w2, w3] - ) - assert res == ('svcname/limitname', - '(limit 12) ' + yellow('WARNING: ' - 'w2id=10, w3id=10, 11')) + def test_verbose(self, capsys): + argv = ['awslimitchecker', '-v'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with patch('awslimitchecker.runner.logger.setLevel' + '') as mock_set_level: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 6 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.code == 6 + assert mock_set_level.mock_calls == [call(logging.INFO)] - def test_print_issue_both_one(self): - mock_limit = Mock(spec_set=AwsLimit) - type(mock_limit).name = 'limitname' - mock_limit.get_limit.return_value = 12 + def test_debug(self, capsys): + argv = ['awslimitchecker', '-vv'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with patch('awslimitchecker.runner.logger.setLevel' + '') as mock_set_level: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 7 + self.cls.console_entry_point() + out, err = capsys.readouterr() + assert out == '' + assert excinfo.value.args[0] == 7 + assert mock_set_level.mock_calls == [call(logging.DEBUG)] - c1 = AwsLimitUsage(mock_limit, 10) - w1 = AwsLimitUsage(mock_limit, 10, resource_id='w3id') + def test_warning(self): + argv = ['awslimitchecker', '-W', '50'] + with patch.object(sys, 'argv', argv): + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 8 + self.cls.console_entry_point() + assert excinfo.value.code == 8 + assert mock_alc.mock_calls == [ + call( + warning_threshold=50, + critical_threshold=99, + account_id=None, + account_role=None, + region=None, + external_id=None, + mfa_serial_number=None, + mfa_token=None, + profile_name=None, + ta_refresh_mode=None, + ta_refresh_timeout=None, + check_version=True + ) + ] - res = self.cls.print_issue( - 'svcname', - mock_limit, - [c1], - [w1] - ) - assert res == ('svcname/limitname', - '(limit 12) ' + - red('CRITICAL: 10') + ' ' + - yellow('WARNING: w3id=10')) + def test_warning_profile_name(self): + argv = ['awslimitchecker', '-W', '50', '-P', 'myprof'] + with patch.object(sys, 'argv', argv): + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 8 + self.cls.console_entry_point() + assert excinfo.value.code == 8 + assert mock_alc.mock_calls == [ + call( + warning_threshold=50, + critical_threshold=99, + account_id=None, + account_role=None, + region=None, + external_id=None, + mfa_serial_number=None, + mfa_token=None, + profile_name='myprof', + ta_refresh_mode=None, + ta_refresh_timeout=None, + check_version=True + ) + ] - def test_print_issue_both_multi(self): - mock_limit = Mock(spec_set=AwsLimit) - type(mock_limit).name = 'limitname' - mock_limit.get_limit.return_value = 12 + def test_critical(self): + argv = ['awslimitchecker', '-C', '95'] + with patch.object(sys, 'argv', argv): + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 9 + self.cls.console_entry_point() + assert excinfo.value.code == 9 + assert mock_alc.mock_calls == [ + call( + warning_threshold=80, + critical_threshold=95, + account_id=None, + account_role=None, + region=None, + external_id=None, + mfa_serial_number=None, + mfa_token=None, + profile_name=None, + ta_refresh_mode=None, + ta_refresh_timeout=None, + check_version=True + ) + ] - c1 = AwsLimitUsage(mock_limit, 10) - c2 = AwsLimitUsage(mock_limit, 12, resource_id='c2id') - c3 = AwsLimitUsage(mock_limit, 8) - w1 = AwsLimitUsage(mock_limit, 11) - w2 = AwsLimitUsage(mock_limit, 10, resource_id='w2id') - w3 = AwsLimitUsage(mock_limit, 10, resource_id='w3id') + def test_critical_ta_refresh(self): + argv = ['awslimitchecker', '-C', '95', '--ta-refresh-timeout=123', + '--ta-refresh-older=456'] + with patch.object(sys, 'argv', argv): + with patch('%s.AwsLimitChecker' % pb, autospec=True) as mock_alc: + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 9 + self.cls.console_entry_point() + assert excinfo.value.code == 9 + assert mock_alc.mock_calls == [ + call( + warning_threshold=80, + critical_threshold=95, + account_id=None, + account_role=None, + region=None, + external_id=None, + mfa_serial_number=None, + mfa_token=None, + profile_name=None, + ta_refresh_mode=456, + ta_refresh_timeout=123, + check_version=True + ) + ] - res = self.cls.print_issue( - 'svcname', - mock_limit, - [c1, c2, c3], - [w1, w2, w3] - ) - assert res == ('svcname/limitname', - '(limit 12) ' + - red('CRITICAL: 8, 10, c2id=12') + ' ' + - yellow('WARNING: w2id=10, w3id=10, 11')) + def test_check_thresholds(self): + argv = ['awslimitchecker'] + with patch.object(sys, 'argv', argv): + with patch('%s.Runner.check_thresholds' % pb, + autospec=True) as mock_ct: + with pytest.raises(SystemExit) as excinfo: + mock_ct.return_value = 10 + self.cls.console_entry_point() + assert excinfo.value.code == 10 + assert mock_ct.mock_calls == [ + call(self.cls) + ] - def test_entry_no_color(self): + def test_no_color(self): argv = ['awslimitchecker', '--no-color'] with patch.object(sys, 'argv', argv): with patch('%s.Runner.check_thresholds' % pb, @@ -1400,7 +1657,3 @@ def test_entry_no_color(self): assert excinfo.value.code == 0 assert self.cls.color_output('foo', 'red') == 'foo' self.cls.colorize = True - - def test_color_output(self): - assert self.cls.color_output('foo', 'yellow') == termcolor.colored( - 'foo', 'yellow') diff --git a/docs/build_generated_docs.py b/docs/build_generated_docs.py index 0ecd1753..2614051b 100644 --- a/docs/build_generated_docs.py +++ b/docs/build_generated_docs.py @@ -237,6 +237,7 @@ def build_runner_examples(): except subprocess.CalledProcessError as e: output = e.output results[name] = format_cmd_output(cmd_str, output, name) + results['%s-output-only' % name] = format_cmd_output(None, output, name) tmpl = tmpl.format(**results) # write out the final .rst @@ -246,8 +247,11 @@ def build_runner_examples(): def format_cmd_output(cmd, output, name): """format command output for docs""" - formatted = '.. code-block:: console\n\n' - formatted += ' (venv)$ {c}\n'.format(c=cmd) + if cmd is None: + formatted = '' + else: + formatted = '.. code-block:: console\n\n' + formatted += ' (venv)$ {c}\n'.format(c=cmd) lines = output.split("\n") if name != 'help': for idx, line in enumerate(lines): @@ -278,6 +282,11 @@ def format_cmd_output(cmd, output, name): for line in lines: if line.strip() == '': continue + if ( + name == 'check_thresholds_custom' and + 'VPC security groups per elastic network interface' in line + ): + continue formatted += ' ' + line + "\n" formatted += '\n' return formatted diff --git a/docs/source/cli_usage.rst b/docs/source/cli_usage.rst index 8051bab0..5d65c3bc 100644 --- a/docs/source/cli_usage.rst +++ b/docs/source/cli_usage.rst @@ -285,16 +285,20 @@ using their IDs). +.. _cli_usage.limit_overrides: + Overriding Limits +++++++++++++++++ In cases where you've been given a limit increase by AWS Support, you can override the default limits with custom ones. Currently, to do this from the command line, -you must specify each limit that you want to override separately (the +you can either specify each limit that you want to override separately using the +``-L`` or ``--limit`` options, or you can specify a JSON file at either a local path +or an S3 URL using the ``--limit-overrides-json`` option (the :py:meth:`~.AwsLimitChecker.set_limit_overrides` Python method accepts a dict for -easy bulk overrides of limits) using the ``-L`` or ``--limit`` options. Limits are -specified in a ``service_name/limit_name=value`` format, and must be quoted if the -limit name contains spaces. +easy bulk overrides of limits). Limits for the ``-L`` / ``--limit`` option are +specified in a ``service_name/limit_name=value`` format, and must be quoted if +the limit name contains spaces. For example, to override the limits of EC2's "EC2-Classic Elastic IPs" and "EC2-VPC Elastic IPs" from their defaults of 5, to 10 and 20, respestively: @@ -314,9 +318,36 @@ For example, to override the limits of EC2's "EC2-Classic Elastic IPs" and VPC/VPCs 1000 (TA) VPC/Virtual private gateways 5 +This example simply sets the overrides, and then prints the limits for confirmation. +You could also set the same limit overrides using a JSON file stored at ``limit_overrides.json``, following the format documented for :py:meth:`awslimitchecker.checker.AwsLimitChecker.set_limit_overrides`: + +.. code-block:: json + + { + "AutoScaling": { + "Auto Scaling groups": 321, + "Launch configurations": 456 + } + } + +Using a command like: + +.. code-block:: console + + (venv)$ awslimitchecker --limit-overrides-json=limit_overrides.json -l + ApiGateway/API keys per account 500 + ApiGateway/APIs per account 60 + ApiGateway/Client certificates per account 60 + ApiGateway/Custom authorizers per API 10 + ApiGateway/Documentation parts per API 2000 + (...) + CloudFormation/Stacks 2500 (API) + (...) + VPC/Subnets per VPC 200 + VPC/VPCs 1000 (TA) + VPC/Virtual private gateways 5 -This example simply sets the overrides, and then prints the limits for confirmation. Check Limits Against Thresholds +++++++++++++++++++++++++++++++ @@ -354,7 +385,7 @@ threshold only, and another has crossed the critical threshold): VPC/NAT Gateways per AZ (limit 5) CRITICAL: us-east-1d=9, us-east-1c= (...) VPC/Virtual private gateways (limit 5) CRITICAL: 6 - +.. _cli_usage.threshold_overrides: Set Custom Thresholds +++++++++++++++++++++ @@ -374,7 +405,60 @@ To set the warning threshold of 50% and a critical threshold of 75% when checkin VPC/NAT Gateways per AZ (limit 5) CRITICAL: us-east-1d=7, us-east-1c= (...) VPC/Virtual private gateways (limit 5) CRITICAL: 5 +You can also set custom thresholds on a per-limit basis using the +``--threshold-override-json`` CLI option, which accepts the path to a JSON file +(local or an s3:// URL) matching the format described in +:py:meth:`awslimitchecker.checker.AwsLimitChecker.set_threshold_overrides`, for example: + +.. code-block:: json + + { + "S3": { + "Buckets": { + "warning": { + "percent": 97 + }, + "critical": { + "percent": 99 + } + } + }, + "EC2": { + "Security groups per VPC": { + "warning": { + "percent": 80, + "count": 800 + }, + "critical": { + "percent": 90, + "count": 900 + } + }, + "VPC security groups per elastic network interface": { + "warning": { + "percent": 101 + }, + "critical": { + "percent": 101 + } + } + } + } + +Using a command like: +.. code-block:: console + + (venv)$ awslimitchecker -W 97 --critical=98 --no-color --threshold-overrides-json=s3://bucketname/path/overrides.json + ApiGateway/APIs per account (limit 60) CRITICAL: 98 + DynamoDB/Local Secondary Indexes (limit 5) CRITICAL: sale_setup_draft_vehicles (...) + DynamoDB/Tables Per Region (limit 256) WARNING: 250 + EC2/Security groups per VPC (limit 500) CRITICAL: vpc-c89074a9=863 + EC2/VPC security groups per elastic network interface (limit 5) CRITICAL: eni-8226ce61=5 + (...) + S3/Buckets (limit 100) CRITICAL: 657 + VPC/NAT Gateways per AZ (limit 5) CRITICAL: us-east-1d=7, us-east-1c= (...) + VPC/Virtual private gateways (limit 5) CRITICAL: 5 Required IAM Policy +++++++++++++++++++ @@ -397,11 +481,12 @@ permissions for it to perform all limit checks. This can be viewed with the "Version": "2012-10-17" } - - For the current IAM Policy required by this version of awslimitchecker, see :ref:`IAM Policy `. +.. important:: + The required IAM policy output by awslimitchecker includes only the permissions required to check limits and usage. If you are loading :ref:`limit overrides ` and/or :ref:`threshold overrides ` from S3, you will need to run awslimitchecker with additional permissions to access those objects. + Connect to a Specific Region ++++++++++++++++++++++++++++ diff --git a/docs/source/cli_usage.rst.template b/docs/source/cli_usage.rst.template index f6483814..40335d1b 100644 --- a/docs/source/cli_usage.rst.template +++ b/docs/source/cli_usage.rst.template @@ -118,16 +118,20 @@ using their IDs). {show_usage} +.. _cli_usage.limit_overrides: + Overriding Limits +++++++++++++++++ In cases where you've been given a limit increase by AWS Support, you can override the default limits with custom ones. Currently, to do this from the command line, -you must specify each limit that you want to override separately (the +you can either specify each limit that you want to override separately using the +``-L`` or ``--limit`` options, or you can specify a JSON file at either a local path +or an S3 URL using the ``--limit-overrides-json`` option (the :py:meth:`~.AwsLimitChecker.set_limit_overrides` Python method accepts a dict for -easy bulk overrides of limits) using the ``-L`` or ``--limit`` options. Limits are -specified in a ``service_name/limit_name=value`` format, and must be quoted if the -limit name contains spaces. +easy bulk overrides of limits). Limits for the ``-L`` / ``--limit`` option are +specified in a ``service_name/limit_name=value`` format, and must be quoted if +the limit name contains spaces. For example, to override the limits of EC2's "EC2-Classic Elastic IPs" and "EC2-VPC Elastic IPs" from their defaults of 5, to 10 and 20, respestively: @@ -136,6 +140,24 @@ For example, to override the limits of EC2's "EC2-Classic Elastic IPs" and This example simply sets the overrides, and then prints the limits for confirmation. +You could also set the same limit overrides using a JSON file stored at ``limit_overrides.json``, following the format documented for :py:meth:`awslimitchecker.checker.AwsLimitChecker.set_limit_overrides`: + +.. code-block:: json + + { + "AutoScaling": { + "Auto Scaling groups": 321, + "Launch configurations": 456 + } + } + +Using a command like: + +.. code-block:: console + + (venv)$ awslimitchecker --limit-overrides-json=limit_overrides.json -l +{limit_overrides-output-only} + Check Limits Against Thresholds +++++++++++++++++++++++++++++++ @@ -161,6 +183,8 @@ threshold only, and another has crossed the critical threshold): {check_thresholds} +.. _cli_usage.threshold_overrides: + Set Custom Thresholds +++++++++++++++++++++ @@ -168,6 +192,53 @@ To set the warning threshold of 50% and a critical threshold of 75% when checkin {check_thresholds_custom} +You can also set custom thresholds on a per-limit basis using the +``--threshold-override-json`` CLI option, which accepts the path to a JSON file +(local or an s3:// URL) matching the format described in +:py:meth:`awslimitchecker.checker.AwsLimitChecker.set_threshold_overrides`, for example: + +.. code-block:: json + + { + "S3": { + "Buckets": { + "warning": { + "percent": 97 + }, + "critical": { + "percent": 99 + } + } + }, + "EC2": { + "Security groups per VPC": { + "warning": { + "percent": 80, + "count": 800 + }, + "critical": { + "percent": 90, + "count": 900 + } + }, + "VPC security groups per elastic network interface": { + "warning": { + "percent": 101 + }, + "critical": { + "percent": 101 + } + } + } + } + +Using a command like: + +.. code-block:: console + + (venv)$ awslimitchecker -W 97 --critical=98 --no-color --threshold-overrides-json=s3://bucketname/path/overrides.json +{check_thresholds_custom-output-only} + Required IAM Policy +++++++++++++++++++ @@ -180,6 +251,9 @@ permissions for it to perform all limit checks. This can be viewed with the For the current IAM Policy required by this version of awslimitchecker, see :ref:`IAM Policy `. +.. important:: + The required IAM policy output by awslimitchecker includes only the permissions required to check limits and usage. If you are loading :ref:`limit overrides ` and/or :ref:`threshold overrides ` from S3, you will need to run awslimitchecker with additional permissions to access those objects. + Connect to a Specific Region ++++++++++++++++++++++++++++ diff --git a/docs/source/iam_policy.rst b/docs/source/iam_policy.rst index 1a3ffca2..f10ef94b 100644 --- a/docs/source/iam_policy.rst +++ b/docs/source/iam_policy.rst @@ -9,6 +9,9 @@ Required IAM Permissions ======================== +.. important:: + The required IAM policy output by awslimitchecker includes only the permissions required to check limits and usage. If you are loading :ref:`limit overrides ` and/or :ref:`threshold overrides ` from S3, you will need to run awslimitchecker with additional permissions to access those objects. + Below is the sample IAM policy from this version of awslimitchecker, listing the IAM permissions required for it to function correctly. Please note that in some cases awslimitchecker may cause AWS services to make additional API calls on your behalf