Skip to content

Commit

Permalink
unit tests for deprecated notification-rules command
Browse files Browse the repository at this point in the history
  • Loading branch information
@adal-chiriliuc-reef
adal-chiriliuc-reef committed Apr 24, 2024
1 parent 31bef99 commit 2f60683
Showing 1 changed file with 155 additions and 155 deletions.
310 changes: 155 additions & 155 deletions test/unit/console_tool/test_notification_rules.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
import pytest


COMMANDS = (["bucket", "notification-rule"], ["notification-rules"])

@pytest.fixture()
def bucket_notification_rule(b2_cli, bucket):
rule = {
Expand Down Expand Up @@ -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:
Expand All @@ -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):
Expand All @@ -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(
Expand All @@ -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]

0 comments on commit 2f60683

Please sign in to comment.