fix notifications integration tests

test/integration/helpers.py

@@ -496,13 +496,13 @@ def execute(
         print_output(p.returncode, stdout_decoded, stderr_decoded)
         return p.returncode, stdout_decoded, stderr_decoded
 
-    def should_succeed_json(self, args, additional_env: dict | None = None):
+    def should_succeed_json(self, args, additional_env: dict | None = None, **kwargs):
         """
         Runs the command-line with the given arguments.  Raises an exception
         if there was an error; otherwise, treats the stdout as JSON and returns
         the data in it.
         """
-        result = self.should_succeed(args, additional_env=additional_env)
+        result = self.should_succeed(args, additional_env=additional_env, **kwargs)
         try:
             loaded_result = json.loads(result)
         except json.JSONDecodeError:
@@ -531,9 +531,13 @@ def reauthorize(self, check_key_capabilities=False):
         )
         if check_key_capabilities:
             auth_dict = self.should_succeed_json(['get-account-info'])
+            private_preview_caps = {
+                'readBucketNotifications',
+                'writeBucketNotifications',
+            }
             missing_capabilities = set(ALL_CAPABILITIES) - {
                 'readBuckets', 'listAllBucketNames'
-            } - set(auth_dict['allowed']['capabilities'])
+            } - private_preview_caps - set(auth_dict['allowed']['capabilities'])
             assert not missing_capabilities, f'it appears that the raw_api integration test is being run with a non-full key. Missing capabilities: {missing_capabilities}'
 
     def list_file_versions(self, bucket_name):

test/integration/test_b2_command_line.py

@@ -3070,8 +3070,14 @@ def assert_expected(file_info, expected=expected_file_info):
 
 
 def test_notification_rules(b2_tool, bucket_name):
+    auth_dict = b2_tool.should_succeed_json(['get-account-info'])
+    if 'writeBucketNotifications' not in auth_dict['allowed']['capabilities']:
+        pytest.skip('Test account does not have writeBucketNotifications capability')
+
+    private_preview_pattern = re.compile(r'FeaturePreviewWarning')
     assert b2_tool.should_succeed_json(
-        ["notification-rules", "list", f"b2://{bucket_name}", "--json"]
+        ["notification-rules", "list", f"b2://{bucket_name}", "--json"],
+        expected_stderr_pattern=private_preview_pattern
     ) == []
 
     notification_rule = {
@@ -3098,7 +3104,8 @@ def test_notification_rules(b2_tool, bucket_name):
             "https://example.com/webhook",
             "--event-type",
             "b2:ObjectCreated:*",
-        ]
+        ],
+        expected_stderr_pattern=private_preview_pattern
     )
     expected_rules = [{**notification_rule, "isSuspended": False, "suspensionReason": ""}]
     assert created_rule == expected_rules[0]
@@ -3112,20 +3119,24 @@ def test_notification_rules(b2_tool, bucket_name):
             f"b2://{bucket_name}/prefix",
             "test-rule",
             "--disable",
-        ]
+        ],
+        expected_stderr_pattern=private_preview_pattern
     )
     expected_rules[0].update({"objectNamePrefix": "prefix", "isEnabled": False})
     assert modified_rule == expected_rules[0]
 
     # read updated rules
     assert b2_tool.should_succeed_json(
-        ["notification-rules", "list", f"b2://{bucket_name}", "--json"]
+        ["notification-rules", "list", f"b2://{bucket_name}", "--json"],
+        expected_stderr_pattern=private_preview_pattern
     ) == expected_rules
 
     # delete rule by name
     assert b2_tool.should_succeed(
         ["notification-rules", "delete", f"b2://{bucket_name}", "test-rule"],
+        expected_stderr_pattern=private_preview_pattern
     ) == f"Rule 'test-rule' has been deleted from b2://{bucket_name}/\n"
     assert b2_tool.should_succeed_json(
-        ["notification-rules", "list", f"b2://{bucket_name}", "--json"]
+        ["notification-rules", "list", f"b2://{bucket_name}", "--json"],
+        expected_stderr_pattern=private_preview_pattern
     ) == []