--bypassGovernance added to delete-file-version

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 * Add ability to upload from an unbound source such as standard input or a named pipe
+* --bypassGovernance option to delete_file_version
 * Declare official support of Python 3.12
 * Cache-Control option when uploading files
 * Add `--lifecycleRule` to `create-bucket` and `update-bucket` and deprecate `--lifecycleRules` argument

b2/console_tool.py

@@ -1236,16 +1236,28 @@ class DeleteFileVersion(FileIdAndOptionalFileNameMixin, Command):
 
     {FILEIDANDOPTIONALFILENAMEMIXIN}
 
+    If a file is in governance retention mode, and the retention period has not expired, adding ``--bypassGovernance``
+    is required.
+
     Requires capability:
 
     - **deleteFiles**
     - **readFiles** (if file name not provided)
+
+    and optionally:
+
+    - **bypassGovernance**
     """
 
+    @classmethod
+    def _setup_parser(cls, parser):
+        super()._setup_parser(parser)
+        parser.add_argument('--bypassGovernance', action='store_true', default=False)
+
     def run(self, args):
         file_name = self._get_file_name_from_args(args)
 
-        file_info = self.api.delete_file_version(args.fileId, file_name)
+        file_info = self.api.delete_file_version(args.fileId, file_name, args.bypassGovernance)
         self._print_json(file_info)
         return 0
 

requirements.txt

@@ -1,6 +1,6 @@
 argcomplete>=2,<4
 arrow>=1.0.2,<2.0.0
-b2sdk>=1.23.0,<2
+b2sdk>=1.24.0,<2
 docutils==0.19
 idna~=2.2.0; platform_system == 'Java'
 importlib-metadata~=3.3; python_version < '3.8'

test/integration/test_b2_command_line.py

@@ -17,6 +17,7 @@
 import os.path
 import re
 import sys
+import time
 from pathlib import Path
 from typing import Optional, Tuple
 
@@ -1823,17 +1824,32 @@ def test_file_lock(b2_tool, application_key_id, application_key, b2_api):
         retain_until=now_millis + 1.25 * ONE_HOUR_MILLIS,
         legal_hold=LegalHold.OFF
     )
+    lock_disabled_key_id, lock_disabled_key = make_lock_disabled_key(b2_tool)
+
+    b2_tool.should_succeed(
+        [
+            'authorize-account', '--environment', b2_tool.realm, lock_disabled_key_id,
+            lock_disabled_key
+        ],
+    )
 
     file_lock_without_perms_test(
         b2_tool, lock_enabled_bucket_name, lock_disabled_bucket_name, lockable_file['fileId'],
         not_lockable_file['fileId']
     )
 
+    b2_tool.should_succeed(
+        ['authorize-account', '--environment', b2_tool.realm, application_key_id, application_key],
+    )
+
+    deleting_locked_files(
+        b2_tool, lock_enabled_bucket_name, lock_disabled_key_id, lock_disabled_key
+    )
+
     # ---- perform test cleanup ----
     b2_tool.should_succeed(
         ['authorize-account', '--environment', b2_tool.realm, application_key_id, application_key],
     )
-    # b2_tool.reauthorize(check_key_capabilities=False)
     buckets = [
         bucket for bucket in b2_api.api.list_buckets()
         if bucket.name in {lock_enabled_bucket_name, lock_disabled_bucket_name}
@@ -1842,23 +1858,23 @@ def test_file_lock(b2_tool, application_key_id, application_key, b2_api):
         b2_api.clean_bucket(bucket)
 
 
-def file_lock_without_perms_test(
-    b2_tool, lock_enabled_bucket_name, lock_disabled_bucket_name, lockable_file_id,
-    not_lockable_file_id
-):
+def make_lock_disabled_key(b2_tool):
     key_name = 'no-perms-for-file-lock' + random_hex(6)
     created_key_stdout = b2_tool.should_succeed(
         [
             'create-key',
             key_name,
-            'listFiles,listBuckets,readFiles,writeKeys',
+            'listFiles,listBuckets,readFiles,writeKeys,deleteFiles',
         ]
     )
-    key_one_id, key_one = created_key_stdout.split()
+    key_id, key = created_key_stdout.split()
+    return key_id, key
 
-    b2_tool.should_succeed(
-        ['authorize-account', '--environment', b2_tool.realm, key_one_id, key_one],
-    )
+
+def file_lock_without_perms_test(
+    b2_tool, lock_enabled_bucket_name, lock_disabled_bucket_name, lockable_file_id,
+    not_lockable_file_id
+):
 
     b2_tool.should_fail(
         [
@@ -1974,6 +1990,58 @@ def file_lock_without_perms_test(
     )
 
 
+def upload_locked_file(b2_tool, bucket_name):
+    return b2_tool.should_succeed_json(
+        [
+            'upload-file',
+            '--noProgress',
+            '--quiet',
+            '--fileRetentionMode',
+            'governance',
+            '--retainUntil',
+            str(int(time.time()) + 1000),
+            bucket_name,
+            'README.md',
+            'a-locked',
+        ]
+    )
+
+
+def deleting_locked_files(
+    b2_tool, lock_enabled_bucket_name, lock_disabled_key_id, lock_disabled_key
+):
+    locked_file = upload_locked_file(b2_tool, lock_enabled_bucket_name)
+    b2_tool.should_fail(
+        [  # master key
+            'delete-file-version',
+            locked_file['fileName'],
+            locked_file['fileId'],
+        ],
+        "ERROR: Access Denied for application key "
+    )
+    b2_tool.should_succeed([  # master key
+        'delete-file-version',
+        locked_file['fileName'],
+        locked_file['fileId'],
+        '--bypassGovernance'
+    ])
+
+    locked_file = upload_locked_file(b2_tool, lock_enabled_bucket_name)
+
+    b2_tool.should_succeed(
+        [
+            'authorize-account', '--environment', b2_tool.realm, lock_disabled_key_id,
+            lock_disabled_key
+        ],
+    )
+    b2_tool.should_fail([  # lock disabled key
+        'delete-file-version',
+        locked_file['fileName'],
+        locked_file['fileId'],
+        '--bypassGovernance',
+    ], "ERROR: unauthorized for application key with capabilities '")
+
+
 def test_profile_switch(b2_tool):
     # this test could be unit, but it adds a lot of complexity because of
     # necessity to pass mocked B2Api to ConsoleTool; it's much easier to

test/unit/test_console_tool.py

@@ -2687,12 +2687,12 @@ def _run_problematic_removal(
         original_delete_file_version = self.b2_api.raw_api.delete_file_version
 
         def mocked_delete_file_version(
-            this, account_auth_token, file_id, file_name, *args, **kwargs
+            this, account_auth_token, file_id, file_name, bypass_governance=False, *args, **kwargs
         ):
             if file_name == 'b/b1/test.csv':
                 raise Conflict()
             return original_delete_file_version(
-                this, account_auth_token, file_id, file_name, *args, **kwargs
+                this, account_auth_token, file_id, file_name, bypass_governance, *args, **kwargs
             )
 
         with mock.patch.object(