diff --git a/test/unit/console_tool/test_notification_rules.py b/test/unit/console_tool/test_notification_rules.py index 2940d17a..afe87f14 100644 --- a/test/unit/console_tool/test_notification_rules.py +++ b/test/unit/console_tool/test_notification_rules.py @@ -12,6 +12,8 @@ import pytest +COMMANDS = (["bucket", "notification-rule"], ["notification-rules"]) + @pytest.fixture() def bucket_notification_rule(b2_cli, bucket): rule = { @@ -46,14 +48,14 @@ def bucket_notification_rule(b2_cli, bucket): def test_notification_rules__list_all(b2_cli, bucket, bucket_notification_rule): - _, stdout, _ = b2_cli.run([ - "bucket", - "notification-rule", - "list", - f"b2://{bucket}", - ]) - assert ( - stdout == f"""\ + for command in COMMANDS: + _, stdout, _ = b2_cli.run([ + *command, + "list", + f"b2://{bucket}", + ]) + assert ( + stdout == f"""\ Notification rules for b2://{bucket}/ : - name: test-rule eventTypes: @@ -66,80 +68,79 @@ def test_notification_rules__list_all(b2_cli, bucket, bucket_notification_rule): targetType: webhook url: https://example.com/webhook """ - ) - + ) def test_notification_rules__list_all_json(b2_cli, bucket, bucket_notification_rule): - _, stdout, _ = b2_cli.run([ - "bucket", - "notification-rule", - "list", - "--json", - f"b2://{bucket}", - ]) - assert json.loads(stdout) == [bucket_notification_rule] + for command in COMMANDS: + _, stdout, _ = b2_cli.run([ + *command, + "list", + "--json", + f"b2://{bucket}", + ]) + assert json.loads(stdout) == [bucket_notification_rule] def test_notification_rules__update(b2_cli, bucket, bucket_notification_rule): - bucket_notification_rule["isEnabled"] = False - _, stdout, _ = b2_cli.run( - [ - "bucket", - "notification-rule", - "update", - "--json", - f"b2://{bucket}", - bucket_notification_rule["name"], - "--disable", - "--custom-header", - "X-Custom-Header=value=1", - ], - ) - bucket_notification_rule["targetConfiguration"]["customHeaders"] = { - "X-Custom-Header": "value=1" - } - assert json.loads(stdout) == bucket_notification_rule + for command in COMMANDS: + bucket_notification_rule["isEnabled"] = False + _, stdout, _ = b2_cli.run( + [ + *command, + "update", + "--json", + f"b2://{bucket}", + bucket_notification_rule["name"], + "--disable", + "--custom-header", + "X-Custom-Header=value=1", + ], + ) + bucket_notification_rule["targetConfiguration"]["customHeaders"] = { + "X-Custom-Header": "value=1" + } + assert json.loads(stdout) == bucket_notification_rule def test_notification_rules__update__no_such_rule(b2_cli, bucket, bucket_notification_rule): - b2_cli.run( - [ - "bucket", - "notification-rule", - "update", - f"b2://{bucket}", - f'{bucket_notification_rule["name"]}-unexisting', - "--disable", - ], - expected_stderr=( - "ERROR: rule with name 'test-rule-unexisting' does not exist on bucket " - "'my-bucket', available rules: ['test-rule']\n" - ), - expected_status=1, - ) + for command in COMMANDS: + b2_cli.run( + [ + *command, + "update", + f"b2://{bucket}", + f'{bucket_notification_rule["name"]}-unexisting', + "--disable", + ], + expected_stderr=( + "ERROR: rule with name 'test-rule-unexisting' does not exist on bucket " + "'my-bucket', available rules: ['test-rule']\n" + ), + expected_status=1, + ) def test_notification_rules__update__custom_header_malformed( b2_cli, bucket, bucket_notification_rule ): - bucket_notification_rule["isEnabled"] = False - _, stdout, _ = b2_cli.run( - [ - "bucket", - "notification-rule", - "update", - "--json", - f"b2://{bucket}", - bucket_notification_rule["name"], - "--disable", - "--custom-header", - "X-Custom-Header: value", - ], - ) - bucket_notification_rule["targetConfiguration"]["customHeaders"] = { - "X-Custom-Header: value": "" - } - assert json.loads(stdout) == bucket_notification_rule + for command in COMMANDS: + bucket_notification_rule["isEnabled"] = False + _, stdout, _ = b2_cli.run( + [ + *command, + "update", + "--json", + f"b2://{bucket}", + bucket_notification_rule["name"], + "--disable", + "--custom-header", + "X-Custom-Header: value", + ], + ) + bucket_notification_rule["targetConfiguration"]["customHeaders"] = { + "X-Custom-Header: value": "" + } + assert json.loads(stdout) == bucket_notification_rule def test_notification_rules__delete(b2_cli, bucket, bucket_notification_rule): @@ -156,20 +157,20 @@ def test_notification_rules__delete(b2_cli, bucket, bucket_notification_rule): def test_notification_rules__delete_no_such_rule(b2_cli, bucket, bucket_notification_rule): - b2_cli.run( - [ - "bucket", - "notification-rule", - "delete", - f"b2://{bucket}", - f'{bucket_notification_rule["name"]}-unexisting', - ], - expected_stderr=( - "ERROR: no such rule to delete: 'test-rule-unexisting', available rules: ['test-rule'];" - " No rules have been deleted.\n" - ), - expected_status=1, - ) + for command in COMMANDS: + b2_cli.run( + [ + *command, + "delete", + f"b2://{bucket}", + f'{bucket_notification_rule["name"]}-unexisting', + ], + expected_stderr=( + "ERROR: no such rule to delete: 'test-rule-unexisting', available rules: ['test-rule'];" + " No rules have been deleted.\n" + ), + expected_status=1, + ) @pytest.mark.parametrize( @@ -181,91 +182,90 @@ def test_notification_rules__delete_no_such_rule(b2_cli, bucket, bucket_notifica ], ) def test_notification_rules__no_rules(b2_cli, bucket, args, expected_stdout): - b2_cli.run( - ["bucket", "notification-rule", "list", f"b2://{bucket}", *args], - expected_stdout=expected_stdout, - ) + for command in COMMANDS: + b2_cli.run( + [*command, "list", f"b2://{bucket}", *args], + expected_stdout=expected_stdout, + ) def test_notification_rules__disable_enable(b2_cli, bucket, bucket_notification_rule): - _, stdout, _ = b2_cli.run( - [ - "bucket", - "notification-rule", - "disable", - "--json", - f"b2://{bucket}", - bucket_notification_rule["name"], - ], - ) - assert json.loads(stdout) == {**bucket_notification_rule, "isEnabled": False} + for command in COMMANDS: + _, stdout, _ = b2_cli.run( + [ + *command, + "disable", + "--json", + f"b2://{bucket}", + bucket_notification_rule["name"], + ], + ) + assert json.loads(stdout) == {**bucket_notification_rule, "isEnabled": False} - _, stdout, _ = b2_cli.run( - [ - "bucket", - "notification-rule", - "enable", - "--json", - f"b2://{bucket}", - bucket_notification_rule["name"], - ], - ) - assert json.loads(stdout) == {**bucket_notification_rule, "isEnabled": True} + _, stdout, _ = b2_cli.run( + [ + *command, + "enable", + "--json", + f"b2://{bucket}", + bucket_notification_rule["name"], + ], + ) + assert json.loads(stdout) == {**bucket_notification_rule, "isEnabled": True} @pytest.mark.parametrize( - "command", + "subcommand", ["disable", "enable"], ) def test_notification_rules__disable_enable__no_such_rule( - b2_cli, bucket, bucket_notification_rule, command + b2_cli, bucket, bucket_notification_rule, subcommand ): - b2_cli.run( - [ - "bucket", - "notification-rule", - command, - f"b2://{bucket}", - f'{bucket_notification_rule["name"]}-unexisting', - ], - expected_stderr=( - "ERROR: rule with name 'test-rule-unexisting' does not exist on bucket " - "'my-bucket', available rules: ['test-rule']\n" - ), - expected_status=1, - ) + for command in COMMANDS: + b2_cli.run( + [ + *command, + subcommand, + f"b2://{bucket}", + f'{bucket_notification_rule["name"]}-unexisting', + ], + expected_stderr=( + "ERROR: rule with name 'test-rule-unexisting' does not exist on bucket " + "'my-bucket', available rules: ['test-rule']\n" + ), + expected_status=1, + ) def test_notification_rules__sign_secret(b2_cli, bucket, bucket_notification_rule): - b2_cli.run( - [ - "bucket", - "notification-rule", - "update", - "--json", - f"b2://{bucket}", - bucket_notification_rule["name"], - "--sign-secret", - "new-secret", - ], - expected_status=2, - ) + for command in COMMANDS: + b2_cli.run( + [ + *command, + "update", + "--json", + f"b2://{bucket}", + bucket_notification_rule["name"], + "--sign-secret", + "new-secret", + ], + expected_status=2, + ) - _, stdout, _ = b2_cli.run( - [ - "bucket", - "notification-rule", - "update", - "--json", - f"b2://{bucket}", - bucket_notification_rule["name"], - "--sign-secret", - "7" * 32, - ], - ) - bucket_notification_rule["targetConfiguration"]["hmacSha256SigningSecret"] = "7" * 32 - assert json.loads(stdout) == bucket_notification_rule + _, stdout, _ = b2_cli.run( + [ + *command, + "update", + "--json", + f"b2://{bucket}", + bucket_notification_rule["name"], + "--sign-secret", + "7" * 32, + ], + ) + bucket_notification_rule["targetConfiguration"]["hmacSha256SigningSecret"] = "7" * 32 + assert json.loads(stdout) == bucket_notification_rule - assert json.loads( - b2_cli.run(["bucket", "notification-rule", "list", "--json", f"b2://{bucket}"],)[1] - ) == [bucket_notification_rule] + assert json.loads( + b2_cli.run([*command, "list", "--json", f"b2://{bucket}"],)[1] + ) == [bucket_notification_rule]