backblaze__b2_command_line_tool-488

{
  "status": "ok",
  "stdout": "./b2/api.py:206:    def get_bucket_by_name(self, bucket_name):\n./b2/api.py:263:                bucket_id = self.get_bucket_by_name(bucket_name).id_\n./b2/raw_simulator.py:709:            bucket = self._get_bucket_by_name(bucket_name)\n./b2/raw_simulator.py:924:    def _get_bucket_by_name(self, bucket_name):\n./b2/sync/folder.py:230:        self.bucket = api.get_bucket_by_name(bucket_name)\n./b2/console_tool.py:299:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:405:            bucket_id_or_none = self.api.get_bucket_by_name(args.bucket).id_\n./b2/console_tool.py:433:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:529:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:656:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:691:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:733:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:779:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:802:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:932:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:983:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:1236:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./b2/console_tool.py:1303:        bucket = self.api.get_bucket_by_name(args.bucketName)\n./test/test_console_tool.py:388:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:395:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:603:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:610:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:794:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:829:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:873:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:913:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_console_tool.py:1070:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n./test/test_policy.py:69:        api.get_bucket_by_name.return_value = bucket\n./test/test_sync.py:248:        self.api.get_bucket_by_name.return_value = self.bucket\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        )\n        return self.transferer.download_file_from_url(url, download_dest, progress_listener, range_)\n\n    def get_bucket_by_id(self, bucket_id):\n        return Bucket(self, bucket_id)\n\n    def get_bucket_by_name(self, bucket_name):\n        \"\"\"\n        Returns the bucket_id for the given bucket_name.\n\n        If we don't already know it from the cache, try fetching it from\n        the B2 service.\n        \"\"\"\n        # If we can get it from the stored info, do that.\n        self.check_bucket_restrictions(bucket_name)\n        id_ = self.cache.get_bucket_id_or_none_from_bucket_name(bucket_name)\n        if id_ is not None:\n            return Bucket(self, id_, name=bucket_name)\n\n        for bucket in self.list_buckets():\n            if bucket.name == bucket_name:\n                return bucket\n        raise NonExistentBucket(bucket_name)\n\n    def delete_bucket(self, bucket):\n        \"\"\"\n        Deletes the bucket remotely.\n        For legacy reasons it returns whatever server sends in response,\n        but API user should not rely on the response: if it doesn't raise\n        an exception, it means that the operation was a success\n        \"\"\"\n        account_id = self.account_info.get_account_id()\n        return self.session.delete_bucket(account_id, bucket.id_)\n\n    def list_buckets(self, bucket_name=None):\n        \"\"\"\n        Calls b2_list_buckets and returns a list of buckets.\n\n        When no bucket name is specified, returns *all* of the buckets\n        in the account.  When a bucket name is given, returns just that\n        bucket.  When authorized with an application key restricted to\n        one bucket, you must specify the bucket name, or the request\n        will be unauthorized.\n\n        :param bucket_name: Optional: the name of the one bucket to return.\n        :return: A list of Bucket objects.\n        \"\"\"\n        account_id = self.account_info.get_account_id()\n        self.check_bucket_restrictions(bucket_name)\n\n        # TEMPORARY work around until we fix the API endpoint bug that things requests\n        # with a bucket name are not authorized.  When it's fixed, well just pass the\n        # bucket name (or None) to the raw API.\n        if bucket_name is None:\n            bucket_id = None\n        else:\n            allowed = self.account_info.get_allowed()\n            if allowed['bucketId'] is not None:\n                # We just checked that if there is a bucket restriction we have a bucket name\n                # and it matches.  So if there's a restriction we know that's the bucket we're\n                # looking for.\n                bucket_id = allowed['bucketId']\n            else:\n                bucket_id = self.get_bucket_by_name(bucket_name).id_\n\n        response = self.session.list_buckets(account_id, bucket_id=bucket_id)\n        buckets = BucketFactory.from_api_response(self, response)\n\n        if bucket_name is not None:\n            # If a bucket_name is specified we don't clear the cache because the other buckets could still\n            # be valid. So we save the one bucket returned from the list_buckets call.\n            for bucket in buckets:\n                self.cache.save_bucket(bucket)\n        else:\n            # Otherwise we want to clear the cache and save the buckets returned from list_buckets\n            # since we just got a new list of all the buckets for this account.\n            self.cache.set_bucket_name_cache(buckets)\n        return buckets\n\n    def list_parts(self, file_id, start_part_number=None, batch_size=None):\n        \"\"\"\n        Generator that yields a Part for each of the parts that have been uploaded.\n\n        :param file_id: the ID of the large file that is not finished\n        :param start_part_number: the first part number to return.  defaults to the first part.\n        :param batch_size: the number of parts to fetch at a time from the server\n        \"\"\"\n        batch_size = batch_size or 100\n        while True:\n            response = self.session.list_parts(file_id, start_part_number, batch_size)\n            for part_dict in response['parts']:\n                yield PartFactory.from_list_parts_dict(part_dict)\n            start_part_number = response.get('nextPartNumber')\n            if start_part_number is None:\n                break\n\n    # delete/cancel\n    def cancel_large_file(self, file_id):\n        response = self.session.cancel_large_file(file_id)\n        return FileVersionInfoFactory.from_cancel_large_file_response(response)\n\n    def delete_file_version(self, file_id, file_name):\n        # filename argument is not first, because one day it may become optional\n        response = self.session.delete_file_version(file_id, file_name)\n        assert response['fileId'] == file_id\n        assert response['fileName'] == file_name\n        return FileIdAndName(file_id, file_name)\n\n    # download\n    def get_download_url_for_fileid(self, file_id):\n        url = url_for_api(self.account_info, 'b2_download_file_by_id')\n        return '%s?fileId=%s' % (url, file_id)\n\n    def get_download_url_for_file_name(self, bucket_name, file_name):\n        \"\"\"\n        Returns a URL to download the given file by name.\n        \"\"\"\n        self.check_bucket_restrictions(bucket_name)\n        return '%s/file/%s/%s' % (\n            self.account_info.get_download_url(), bucket_name, b2_url_encode(file_name)\n        )\n\n    # keys\n    def create_key(\n        self, capabilities, key_name, valid_duration_seconds=None, bucket_id=None, name_prefix=None\n    ):\n        account_id = self.account_info.get_account_id()\n\n        response = self.session.create_key(\n            account_id,\n            capabilities=capabilities,\n            key_name=key_name,\n            valid_duration_seconds=valid_duration_seconds,\n            bucket_id=bucket_id,\n            name_prefix=name_prefix\n        )\n\n        assert set(response['capabilities']) == set(capabilities)\n        assert response['keyName'] == key_name\n\n        return response\n\n    def delete_key(self, application_key_id):\n\n        response = self.session.delete_key(application_key_id=application_key_id)\n        return response\n\n    def list_keys(self, start_application_key_id=None):\n        account_id = self.account_info.get_account_id()\n",
  "start_line": 200,
  "end_line": 349,
  "total_lines": 372,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "######################################################################\n#\n# File: b2/api.py\n#\n# Copyright 2018 Backblaze Inc. All Rights Reserved.\n#\n# License https://www.backblaze.com/using_b2_code.html\n#\n######################################################################\n\nimport six\n\nfrom .account_info.sqlite_account_info import SqliteAccountInfo\nfrom .account_info.exception import MissingAccountData\nfrom .b2http import B2Http\nfrom .bucket import Bucket, BucketFactory\nfrom .cache import AuthInfoCache, DummyCache\nfrom .transferer import Transferer\nfrom .exception import NonExistentBucket, RestrictedBucket\nfrom .file_version import FileVersionInfoFactory, FileIdAndName\nfrom .part import PartFactory\nfrom .raw_api import B2RawApi\nfrom .session import B2Session\nfrom .utils import B2TraceMeta, b2_url_encode, limit_trace_arguments\n\ntry:\n    import concurrent.futures as futures\nexcept ImportError:\n    import futures\n\n\ndef url_for_api(info, api_name):\n    if api_name in ['b2_download_file_by_id']:\n        base = info.get_download_url()\n    else:\n        base = info.get_api_url()\n    return base + '/b2api/v1/' + api_name\n\n\n@six.add_metaclass(B2TraceMeta)\nclass B2Api(object):\n    \"\"\"\n    Provides file-level access to B2 services.\n\n    While B2RawApi provides direct access to the B2 web APIs, this\n    class handles several things that simplify the task of uploading\n    and downloading files:\n      - re-acquires authorization tokens when they expire\n      - retrying uploads when an upload URL is busy\n      - breaking large files into parts\n      - emulating a directory structure (B2 buckets are flat)\n\n    Adds an object-oriented layer on top of the raw API, so that\n    buckets and files returned are Python objects with accessor\n    methods.\n\n    Also,  keeps a cache of information needed to access the service,\n    such as auth tokens and upload URLs.\n    \"\"\"\n\n    def __init__(self, account_info=None, cache=None, raw_api=None, max_upload_workers=10):\n        \"\"\"\n        Initializes the API using the given account info.\n        :param account_info:\n        :param cache:\n        :param raw_api:\n        :return:\n        \"\"\"\n        self.raw_api = raw_api or B2RawApi(B2Http())\n        if account_info is None:\n            account_info = SqliteAccountInfo()\n            if cache is None:\n                cache = AuthInfoCache(account_info)\n        self.session = B2Session(self, self.raw_api)\n        self.transferer = Transferer(self.session, account_info)\n        self.account_info = account_info\n        if cache is None:\n            cache = DummyCache()\n        self.cache = cache\n        self.upload_executor = None\n        self.max_workers = max_upload_workers\n\n    def set_thread_pool_size(self, max_workers):\n        \"\"\"\n        Sets the size of the thread pool to use for uploads and downloads.\n\n        Must be called before any work starts, or the thread pool will get\n        the default size of 1.\n        \"\"\"\n        if self.upload_executor is not None:\n            raise Exception('thread pool already created')\n        self.max_workers = max_workers\n\n    def get_thread_pool(self):\n        \"\"\"\n        Returns the thread pool executor to use for uploads and downloads.\n        \"\"\"\n        if self.upload_executor is None:\n            self.upload_executor = futures.ThreadPoolExecutor(max_workers=self.max_workers)\n        return self.upload_executor\n\n    def authorize_automatically(self):\n        try:\n            self.authorize_account(\n                self.account_info.get_realm(),\n                self.account_info.get_account_id(),\n                self.account_info.get_application_key(),\n            )\n        except MissingAccountData:\n            return False\n        return True\n\n    @limit_trace_arguments(only=('self', 'realm'))\n    def authorize_account(self, realm, account_id_or_key_id, application_key):\n        # Clean up any previous account info if it was for a different account.\n        try:\n            old_account_id = self.account_info.get_account_id()\n            old_realm = self.account_info.get_realm()\n            if account_id_or_key_id != old_account_id or realm != old_realm:\n                self.cache.clear()\n        except MissingAccountData:\n            self.cache.clear()\n\n        # Authorize\n        realm_url = self.account_info.REALM_URLS[realm]\n        response = self.raw_api.authorize_account(realm_url, account_id_or_key_id, application_key)\n        allowed = response['allowed']\n\n        # If there is a bucket restriction, get the name of the bucket.\n        # And, if we have a list of the one allowed bucket, go ahead and\n        # save it.\n        if 'listBuckets' in allowed['capabilities']:\n            if allowed['bucketId'] is not None:\n                allowed_bucket_response = self.raw_api.list_buckets(\n                    api_url=response['apiUrl'],\n                    account_auth_token=response['authorizationToken'],\n                    account_id=response['accountId'],\n                    bucket_id=allowed['bucketId'],\n                )\n                allowed_bucket_list = allowed_bucket_response['buckets']\n                allowed_bucket_count = len(allowed_bucket_list)\n                assert allowed_bucket_count in [0, 1]\n                if allowed_bucket_count == 0:\n                    # bucket has been deleted since the key was made\n                    allowed['bucketName'] = None\n                else:\n                    allowed['bucketName'] = allowed_bucket_list[0]['bucketName']\n                    self.cache.set_bucket_name_cache(\n                        BucketFactory.from_api_response(self, allowed_bucket_response)\n                    )",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 372,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "359:    def check_bucket_restrictions(self, bucket_name):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        return self.session.list_keys(\n            account_id, max_key_count=1000, start_application_key_id=start_application_key_id\n        )\n\n    # other\n    def get_file_info(self, file_id):\n        \"\"\" legacy interface which just returns whatever remote API returns \"\"\"\n        return self.session.get_file_info(file_id)\n\n    def check_bucket_restrictions(self, bucket_name):\n        \"\"\"\n        Checks to see if the allowed field from authorize-account\n        has a bucket restriction.\n\n        If it does, does the bucket_name for a given api call match that.\n        If not it raises a RestrictedBucket error.\n        \"\"\"\n        allowed = self.account_info.get_allowed()\n        allowed_bucket_name = allowed['bucketName']\n\n        if allowed_bucket_name is not None:\n            if allowed_bucket_name != bucket_name:\n                raise RestrictedBucket(allowed_bucket_name)",
  "start_line": 350,
  "end_line": 372,
  "total_lines": 372,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "######################################################################\n#\n# File: b2/api.py\n#\n# Copyright 2018 Backblaze Inc. All Rights Reserved.\n#\n# License https://www.backblaze.com/using_b2_code.html\n#\n######################################################################\n\nimport six\n\nfrom .account_info.sqlite_account_info import SqliteAccountInfo\nfrom .account_info.exception import MissingAccountData\nfrom .b2http import B2Http\nfrom .bucket import Bucket, BucketFactory\nfrom .cache import AuthInfoCache, DummyCache\nfrom .transferer import Transferer\nfrom .exception import NonExistentBucket, RestrictedBucket\nfrom .file_version import FileVersionInfoFactory, FileIdAndName\nfrom .part import PartFactory\nfrom .raw_api import B2RawApi\nfrom .session import B2Session\nfrom .utils import B2TraceMeta, b2_url_encode, limit_trace_arguments\n\ntry:\n    import concurrent.futures as futures\nexcept ImportError:\n    import futures\n\n\ndef url_for_api(info, api_name):\n    if api_name in ['b2_download_file_by_id']:\n        base = info.get_download_url()\n    else:\n        base = info.get_api_url()\n    return base + '/b2api/v1/' + api_name\n\n\n@six.add_metaclass(B2TraceMeta)\nclass B2Api(object):\n    \"\"\"\n    Provides file-level access to B2 services.\n\n    While B2RawApi provides direct access to the B2 web APIs, this\n    class handles several things that simplify the task of uploading\n    and downloading files:\n      - re-acquires authorization tokens when they expire\n      - retrying uploads when an upload URL is busy\n      - breaking large files into parts\n      - emulating a directory structure (B2 buckets are flat)\n\n    Adds an object-oriented layer on top of the raw API, so that\n    buckets and files returned are Python objects with accessor\n    methods.\n\n    Also,  keeps a cache of information needed to access the service,\n    such as auth tokens and upload URLs.\n    \"\"\"\n\n    def __init__(self, account_info=None, cache=None, raw_api=None, max_upload_workers=10):\n        \"\"\"\n        Initializes the API using the given account info.\n        :param account_info:\n        :param cache:\n        :param raw_api:\n        :return:\n        \"\"\"\n        self.raw_api = raw_api or B2RawApi(B2Http())\n        if account_info is None:\n            account_info = SqliteAccountInfo()\n            if cache is None:\n                cache = AuthInfoCache(account_info)\n        self.session = B2Session(self, self.raw_api)\n        self.transferer = Transferer(self.session, account_info)\n        self.account_info = account_info\n        if cache is None:\n            cache = DummyCache()\n        self.cache = cache\n        self.upload_executor = None\n        self.max_workers = max_upload_workers\n\n    def set_thread_pool_size(self, max_workers):\n        \"\"\"\n        Sets the size of the thread pool to use for uploads and downloads.\n\n        Must be called before any work starts, or the thread pool will get\n        the default size of 1.\n        \"\"\"\n        if self.upload_executor is not None:\n            raise Exception('thread pool already created')\n        self.max_workers = max_workers\n\n    def get_thread_pool(self):\n        \"\"\"\n        Returns the thread pool executor to use for uploads and downloads.\n        \"\"\"\n        if self.upload_executor is None:\n            self.upload_executor = futures.ThreadPoolExecutor(max_workers=self.max_workers)\n        return self.upload_executor\n\n    def authorize_automatically(self):\n        try:\n            self.authorize_account(\n                self.account_info.get_realm(),\n                self.account_info.get_account_id(),\n                self.account_info.get_application_key(),\n            )\n        except MissingAccountData:\n            return False\n        return True\n\n    @limit_trace_arguments(only=('self', 'realm'))\n    def authorize_account(self, realm, account_id_or_key_id, application_key):\n        # Clean up any previous account info if it was for a different account.\n        try:\n            old_account_id = self.account_info.get_account_id()\n            old_realm = self.account_info.get_realm()\n            if account_id_or_key_id != old_account_id or realm != old_realm:\n                self.cache.clear()\n        except MissingAccountData:\n            self.cache.clear()\n\n        # Authorize\n        realm_url = self.account_info.REALM_URLS[realm]\n        response = self.raw_api.authorize_account(realm_url, account_id_or_key_id, application_key)\n        allowed = response['allowed']\n\n        # If there is a bucket restriction, get the name of the bucket.\n        # And, if we have a list of the one allowed bucket, go ahead and\n        # save it.\n        if 'listBuckets' in allowed['capabilities']:\n            if allowed['bucketId'] is not None:\n                allowed_bucket_response = self.raw_api.list_buckets(\n                    api_url=response['apiUrl'],\n                    account_auth_token=response['authorizationToken'],\n                    account_id=response['accountId'],\n                    bucket_id=allowed['bucketId'],\n                )\n                allowed_bucket_list = allowed_bucket_response['buckets']\n                allowed_bucket_count = len(allowed_bucket_list)\n                assert allowed_bucket_count in [0, 1]\n                if allowed_bucket_count == 0:\n                    # bucket has been deleted since the key was made\n                    allowed['bucketName'] = None\n                else:\n                    allowed['bucketName'] = allowed_bucket_list[0]['bucketName']\n                    self.cache.set_bucket_name_cache(\n                        BucketFactory.from_api_response(self, allowed_bucket_response)\n                    )",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 372,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "test/test_api.py:15:from b2.exception import RestrictedBucket\ntest/test_api.py:62:        with self.assertRaises(RestrictedBucket):\ntest/test_api.py:71:        with self.assertRaises(RestrictedBucket):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "test/test_api.py",
  "content": "        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])\n        self.assertEqual(\n            ['bucket1'],\n            [b.name for b in self.api.list_buckets(bucket_name=bucket1.name)],\n        )\n\n    def test_list_buckets_with_restriction_and_wrong_name(self):\n        self._authorize_account()\n        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')\n        bucket2 = self.api.create_bucket('bucket2', 'allPrivate')\n        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)\n        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])\n        with self.assertRaises(RestrictedBucket):\n            self.api.list_buckets(bucket_name=bucket2.name)\n\n    def test_list_buckets_with_restriction_and_no_name(self):\n        self._authorize_account()\n        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')\n        self.api.create_bucket('bucket2', 'allPrivate')\n        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)\n        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])\n        with self.assertRaises(RestrictedBucket):\n            self.api.list_buckets()\n\n    def _authorize_account(self):\n        self.api.authorize_account('production', self.account_id, self.master_key)",
  "start_line": 50,
  "end_line": 75,
  "total_lines": 75,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "test/test_console_tool.py:388:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:395:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:603:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:610:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:794:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:829:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:873:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:913:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:1070:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_policy.py:69:        api.get_bucket_by_name.return_value = bucket\ntest/test_sync.py:248:        self.api.get_bucket_by_name.return_value = self.bucket\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "E\n======================================================================\nERROR: test_get_bucket_by_name_restricted (__main__.TestB2Api)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"/workspace/b2/account_info/sqlite_account_info.py\", line 269, in _get_account_info_or_raise\n    value = cursor.fetchone()[0]\nTypeError: 'NoneType' object is not subscriptable\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"<stdin>\", line 27, in test_get_bucket_by_name_restricted\n  File \"<stdin>\", line 21, in test_get_bucket_by_name_restricted\n  File \"/workspace/b2/api.py\", line 219, in get_bucket_by_name\n    for bucket in self.list_buckets():\n  File \"/usr/local/lib/python3.7/site-packages/logfury/_logfury/trace_call.py\", line 86, in wrapper\n    return function(*wrapee_args, **wrapee_kwargs)\n  File \"/workspace/b2/api.py\", line 247, in list_buckets\n    account_id = self.account_info.get_account_id()\n  File \"/workspace/b2/account_info/sqlite_account_info.py\", line 238, in get_account_id\n    return self._get_account_info_or_raise('account_id')\n  File \"/workspace/b2/account_info/sqlite_account_info.py\", line 276, in _get_account_info_or_raise\n    raise MissingAccountData(str(e))\nb2.account_info.exception.MissingAccountData: Missing account data: 'NoneType' object is not subscriptable\n\n----------------------------------------------------------------------\nRan 1 test in 0.016s\n\nFAILED (errors=1)\n",
  "details": {
    "stdout": "Caught unexpected exception: MissingAccountData: Missing account data: 'NoneType' object is not subscriptable\n",
    "stderr": "E\n======================================================================\nERROR: test_get_bucket_by_name_restricted (__main__.TestB2Api)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"/workspace/b2/account_info/sqlite_account_info.py\", line 269, in _get_account_info_or_raise\n    value = cursor.fetchone()[0]\nTypeError: 'NoneType' object is not subscriptable\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"<stdin>\", line 27, in test_get_bucket_by_name_restricted\n  File \"<stdin>\", line 21, in test_get_bucket_by_name_restricted\n  File \"/workspace/b2/api.py\", line 219, in get_bucket_by_name\n    for bucket in self.list_buckets():\n  File \"/usr/local/lib/python3.7/site-packages/logfury/_logfury/trace_call.py\", line 86, in wrapper\n    return function(*wrapee_args, **wrapee_kwargs)\n  File \"/workspace/b2/api.py\", line 247, in list_buckets\n    account_id = self.account_info.get_account_id()\n  File \"/workspace/b2/account_info/sqlite_account_info.py\", line 238, in get_account_id\n    return self._get_account_info_or_raise('account_id')\n  File \"/workspace/b2/account_info/sqlite_account_info.py\", line 276, in _get_account_info_or_raise\n    raise MissingAccountData(str(e))\nb2.account_info.exception.MissingAccountData: Missing account data: 'NoneType' object is not subscriptable\n\n----------------------------------------------------------------------\nRan 1 test in 0.016s\n\nFAILED (errors=1)\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "Caught expected exception: Application key is restricted to bucket: my-restricted-bucket\n",
  "stderr": ".\n----------------------------------------------------------------------\nRan 1 test in 0.001s\n\nOK\n",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        If we don't already know it from the cache, try fetching it from\n        the B2 service.\n        \"\"\"\n        # If we can get it from the stored info, do that.\n        self.check_bucket_restrictions(bucket_name)\n        id_ = self.cache.get_bucket_id_or_none_from_bucket_name(bucket_name)\n        if id_ is not None:\n            return Bucket(self, id_, name=bucket_name)\n\n        for bucket in self.list_buckets():\n            if bucket.name == bucket_name:\n                return bucket\n        raise NonExistentBucket(bucket_name)\n\n    def delete_bucket(self, bucket):\n        \"\"\"",
  "start_line": 210,
  "end_line": 225,
  "total_lines": 372,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "b2/api.py:41:class B2Api(object):\nb2/raw_simulator.py:508:class RawSimulator(AbstractRawApi):\nb2/console_tool.py:1189:class TestRawApi(Command):\nb2/raw_api.py:51:class AbstractRawApi(object):\nb2/raw_api.py:124:class B2RawApi(AbstractRawApi):\n",
  "stderr": "",
  "exit_code": 0
}