test_sqlite_account_info.py

######################################################################
#
# File: test/unit/account_info/test_sqlite_account_info.py
#
# Copyright 2021 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
from __future__ import annotations

import json
import os

import pytest
from apiver_deps import (
    B2_ACCOUNT_INFO_DEFAULT_FILE,
    B2_ACCOUNT_INFO_ENV_VAR,
    XDG_CONFIG_HOME_ENV_VAR,
    AbstractAccountInfo,
    SqliteAccountInfo,
)
from apiver_deps_exception import MissingAccountData

from b2sdk.v2 import SqliteAccountInfo as V2SqliteAccountInfo
from b2sdk.v3 import SqliteAccountInfo as V3SqliteAccountInfo

from .fixtures import *


class TestDatabseMigrations:
    @pytest.fixture(autouse=True)
    def setup(self, sqlite_account_info_factory, account_info_default_data_schema_0):
        self.sqlite_account_info_factory = sqlite_account_info_factory
        self.account_info_default_data = account_info_default_data_schema_0

    def test_upgrade_1_default_allowed(self):
        """The 'allowed' field should be the default for upgraded databases."""
        old_account_info = self.sqlite_account_info_factory(last_upgrade_to_run=0)
        old_account_info.set_auth_data_with_schema_0_for_test(**self.account_info_default_data)
        new_account_info = self.sqlite_account_info_factory(file_name=old_account_info.filename)

        assert AbstractAccountInfo.DEFAULT_ALLOWED == new_account_info.get_allowed()

    def test_upgrade_2_default_app_key(self):
        """The 'application_key_id' field should default to the account ID."""
        old_account_info = self.sqlite_account_info_factory(last_upgrade_to_run=0)
        old_account_info.set_auth_data_with_schema_0_for_test(**self.account_info_default_data)
        new_account_info = self.sqlite_account_info_factory(file_name=old_account_info.filename)

        assert 'account_id' == new_account_info.get_application_key_id()

    def test_upgrade_3_default_s3_api_url(self):
        """The 's3_api_url' field should be set."""
        old_account_info = self.sqlite_account_info_factory(last_upgrade_to_run=0)
        old_account_info.set_auth_data_with_schema_0_for_test(**self.account_info_default_data)
        new_account_info = self.sqlite_account_info_factory(file_name=old_account_info.filename)

        assert '' == new_account_info.get_s3_api_url()

    def test_migrate_to_4(self):
        old_account_info = self.sqlite_account_info_factory(last_upgrade_to_run=0)
        old_account_info.set_auth_data_with_schema_0_for_test(**self.account_info_default_data)
        new_account_info = self.sqlite_account_info_factory(file_name=old_account_info.filename)

        with new_account_info._get_connection() as conn:
            sizes = conn.execute(
                'SELECT recommended_part_size, absolute_minimum_part_size from account'
            ).fetchone()
        assert (100, 5000000) == sizes

    @pytest.mark.parametrize(
        'allowed',
        [
            dict(
                bucketId='123',
                bucketName='bucket1',
                capabilities=['listBuckets', 'readBuckets'],
                namePrefix=None,
            ),
            dict(
                bucketId=None,
                bucketName=None,
                capabilities=['listBuckets', 'readBuckets'],
                namePrefix=None,
            ),
        ],
    )
    @pytest.mark.apiver(to_ver=2)
    @pytest.mark.parametrize(
        ('start_version', 'end_version'),
        [
            (4, 5),
            (5, 6),
        ],
    )
    def test_migrate_to_5_prior_to_v3(
        self,
        sqlite_account_info_factory,
        account_info_default_data: dict,
        allowed: dict,
        start_version: int,
        end_version: int,
    ):
        old_account_info = sqlite_account_info_factory(last_upgrade_to_run=start_version)
        old_account_info.set_auth_data(**account_info_default_data)

        allowed_text = json.dumps(allowed)
        with old_account_info._get_connection() as conn:
            conn.execute(f"UPDATE account SET allowed = ('{allowed_text}');")

        assert old_account_info._get_update_count(start_version) == 1
        assert old_account_info._get_update_count(end_version) == 0

        assert old_account_info.get_allowed() == allowed

        new_account_info = self.sqlite_account_info_factory(
            file_name=old_account_info.filename, last_upgrade_to_run=end_version
        )

        assert new_account_info._get_update_count(end_version) == 1
        assert new_account_info.get_allowed() == allowed

        with new_account_info._get_connection() as conn:
            new_allowed = json.loads(conn.execute('SELECT allowed FROM account;').fetchone()[0])

        assert 'buckets' in new_allowed
        assert 'bucketId' not in new_allowed
        assert 'bucketName' not in new_allowed

    @pytest.mark.parametrize(
        ('old_allowed', 'exp_allowed'),
        [
            (
                dict(
                    bucketId='123',
                    bucketName='bucket1',
                    capabilities=['listBuckets', 'readBuckets'],
                    namePrefix=None,
                ),
                dict(
                    buckets=[{'id': '123', 'name': 'bucket1'}],
                    capabilities=['listBuckets', 'readBuckets'],
                    namePrefix=None,
                ),
            ),
            (
                dict(
                    bucketId=None,
                    bucketName=None,
                    capabilities=['listBuckets', 'readBuckets'],
                    namePrefix=None,
                ),
                dict(
                    buckets=None,
                    capabilities=['listBuckets', 'readBuckets'],
                    namePrefix=None,
                ),
            ),
        ],
    )
    @pytest.mark.parametrize(
        ('start_version', 'end_version'),
        [
            (4, 5),
            (5, 6),
        ],
    )
    @pytest.mark.apiver(from_ver=3)
    def test_migrate_to_5_v3(
        self,
        sqlite_account_info_factory,
        account_info_default_data: dict,
        old_allowed: dict,
        exp_allowed: dict,
        start_version: int,
        end_version: int,
    ):
        old_account_info = sqlite_account_info_factory(last_upgrade_to_run=start_version)
        old_account_info.set_auth_data(**account_info_default_data)

        allowed_text = json.dumps(old_allowed)
        with old_account_info._get_connection() as conn:
            conn.execute(f"UPDATE account SET allowed = ('{allowed_text}');")

        assert old_account_info.get_allowed() == old_allowed
        assert old_account_info._get_update_count(start_version) == 1
        assert old_account_info._get_update_count(end_version) == 0

        new_account_info = self.sqlite_account_info_factory(
            file_name=old_account_info.filename, last_upgrade_to_run=end_version
        )

        assert new_account_info.get_allowed() == exp_allowed
        assert new_account_info._get_update_count(end_version) == 1

    @pytest.mark.apiver(2)
    def test_multi_bucket_key_error_apiver_v2(
        self,
        sqlite_account_info_factory,
        account_info_default_data,
    ):
        allowed = dict(
            buckets=[{'id': 1, 'name': 'bucket1'}, {'id': 2, 'name': 'bucket2'}],
            capabilities=['listBuckets', 'readBuckets'],
            namePrefix=None,
        )

        v3_account_info = sqlite_account_info_factory(account_info_class=V3SqliteAccountInfo)
        account_info_default_data.update({'allowed': allowed})
        v3_account_info.set_auth_data(**account_info_default_data)

        assert v3_account_info.get_allowed() == allowed

        v2_account_info = sqlite_account_info_factory(
            file_name=v3_account_info.filename,
            account_info_class=V2SqliteAccountInfo,
            last_upgrade_to_run=4,
        )
        with pytest.raises(MissingAccountData):
            v2_account_info.get_allowed()


class TestSqliteAccountProfileFileLocation:
    @pytest.fixture(autouse=True)
    def setup(self, monkeypatch, tmpdir):
        monkeypatch.setenv(
            'HOME', str(tmpdir)
        )  # this affects .expanduser() and protects the real HOME folder
        monkeypatch.setenv('USERPROFILE', str(tmpdir))  # same as HOME, but for Windows
        monkeypatch.delenv(B2_ACCOUNT_INFO_ENV_VAR, raising=False)
        monkeypatch.delenv(XDG_CONFIG_HOME_ENV_VAR, raising=False)

    def test_invalid_profile_name(self):
        with pytest.raises(ValueError):
            SqliteAccountInfo.get_user_account_info_path(profile='&@(*$')

    def test_profile_and_file_name_conflict(self):
        with pytest.raises(ValueError):
            SqliteAccountInfo.get_user_account_info_path(file_name='foo', profile='bar')

    def test_profile_and_env_var_conflict(self, monkeypatch):
        monkeypatch.setenv(B2_ACCOUNT_INFO_ENV_VAR, 'foo')
        with pytest.raises(ValueError):
            SqliteAccountInfo.get_user_account_info_path(profile='bar')

    def test_profile_and_xdg_config_env_var(self, monkeypatch):
        monkeypatch.setenv(XDG_CONFIG_HOME_ENV_VAR, os.path.join('~', 'custom'))
        account_info_path = SqliteAccountInfo.get_user_account_info_path(profile='secondary')
        assert account_info_path == os.path.expanduser(
            os.path.join('~', 'custom', 'b2', 'db-secondary.sqlite')
        )

    def test_profile(self, monkeypatch):
        xdg_config_path = SqliteAccountInfo._get_xdg_config_path()
        if xdg_config_path:
            expected_path = (xdg_config_path, 'b2', 'db-foo.sqlite')
        else:
            expected_path = ('~', '.b2db-foo.sqlite')

        account_info_path = SqliteAccountInfo.get_user_account_info_path(profile='foo')
        assert account_info_path == os.path.expanduser(os.path.join(*expected_path))

    def test_file_name(self):
        account_info_path = SqliteAccountInfo.get_user_account_info_path(
            file_name=os.path.join('~', 'foo')
        )
        assert account_info_path == os.path.expanduser(os.path.join('~', 'foo'))

    def test_env_var(self, monkeypatch):
        monkeypatch.setenv(B2_ACCOUNT_INFO_ENV_VAR, os.path.join('~', 'foo'))
        account_info_path = SqliteAccountInfo.get_user_account_info_path()
        assert account_info_path == os.path.expanduser(os.path.join('~', 'foo'))

    def test_default_file_if_exists(self, monkeypatch):
        # ensure that XDG_CONFIG_HOME_ENV_VAR doesn't matter if default file exists
        monkeypatch.setenv(XDG_CONFIG_HOME_ENV_VAR, 'some')
        account_file_path = os.path.expanduser(B2_ACCOUNT_INFO_DEFAULT_FILE)
        parent_dir = os.path.abspath(os.path.join(account_file_path, os.pardir))
        os.makedirs(parent_dir, exist_ok=True)
        with open(account_file_path, 'w') as account_file:
            account_file.write('')
        account_info_path = SqliteAccountInfo.get_user_account_info_path()
        assert account_info_path == os.path.expanduser(B2_ACCOUNT_INFO_DEFAULT_FILE)

    def test_xdg_config_env_var(self, monkeypatch):
        monkeypatch.setenv(XDG_CONFIG_HOME_ENV_VAR, os.path.join('~', 'custom'))
        account_info_path = SqliteAccountInfo.get_user_account_info_path()
        assert account_info_path == os.path.expanduser(
            os.path.join('~', 'custom', 'b2', 'account_info')
        )

    def test_default_file(self):
        xdg_config_path = SqliteAccountInfo._get_xdg_config_path()
        if xdg_config_path:
            expected_path = os.path.join(xdg_config_path, 'b2', 'account_info')
        else:
            expected_path = B2_ACCOUNT_INFO_DEFAULT_FILE

        account_info_path = SqliteAccountInfo.get_user_account_info_path()
        assert account_info_path == os.path.expanduser(expected_path)