Add support for XDG_CONFIG_HOME

Author: 0az

b2sdk/account_info/sqlite_account_info.py

@@ -37,17 +37,43 @@ class SqliteAccountInfo(UrlPoolAccountInfo):
 
     def __init__(self, file_name=None, last_upgrade_to_run=None):
         """
-        If ``file_name`` argument is empty or ``None``, path from ``B2_ACCOUNT_INFO`` environment variable is used. If that is not available, a default of ``~/.b2_account_info`` is used.
+        Initialize SqliteAccountInfo.
+
+        The exact algorithm used to determine the location of the database file is not API in any sense.
+        If the location of the database file is required (for cleanup, etc), do not assume a specific resolution:
+        instead, use ``self.filename`` to get the actual resolved location.
+
+        SqliteAccountInfo currently checks locations in the following order:
+        * ``file_name``, if truthy
+        * ``B2_ACCOUNT_INFO_ENV_VAR``'s value, if set
+        * ``~/.b2_account_info``, if it exists
+        * ``$XDG_CONFIG_HOME/b2/account_info``, if XDG_CONFIG_HOME is set
+        * ``~/.b2_account_info``, as default
+
+        If the directory ``$XDG_CONFIG_HOME/b2`` does not exist, it is created.
 
         :param str file_name: The sqlite file to use; overrides the default.
         :param int last_upgrade_to_run: For testing only, override the auto-update on the db.
         """
         self.thread_local = threading.local()
-        user_account_info_path = file_name or os.environ.get(
-            B2_ACCOUNT_INFO_ENV_VAR, B2_ACCOUNT_INFO_DEFAULT_FILE
-        )
-        self.filename = file_name or os.path.expanduser(user_account_info_path)
+
+        if file_name:
+            user_account_info_path = file_name
+        elif B2_ACCOUNT_INFO_ENV_VAR in os.environ:
+            user_account_info_path = os.environ[B2_ACCOUNT_INFO_ENV_VAR]
+        elif os.path.exists(os.path.expanduser(B2_ACCOUNT_INFO_DEFAULT_FILE)):
+            user_account_info_path = B2_ACCOUNT_INFO_DEFAULT_FILE
+        elif 'XDG_CONFIG_HOME' in os.environ:
+            config_home = os.environ['XDG_CONFIG_HOME']
+            user_account_info_path = os.path.join(config_home, 'b2', 'account_info')
+            if not os.path.exists(os.path.join(config_home, 'b2')):
+                os.makedirs(os.path.join(config_home, 'b2'), mode=0o755)
+        else:
+            user_account_info_path = B2_ACCOUNT_INFO_DEFAULT_FILE
+
+        self.filename = os.path.expanduser(user_account_info_path)
         logger.debug('%s file path to use: %s', self.__class__.__name__, self.filename)
+
         self._validate_database()
         with self._get_connection() as conn:
             self._create_tables(conn, last_upgrade_to_run)

b2sdk/v1/account_info.py

@@ -8,10 +8,14 @@
 #
 ######################################################################
 import inspect
+import logging
+import threading
+import os
 from typing import Optional
 
 from b2sdk import _v2 as v2
 
+logger = logging.getLogger(__name__)
 
 class OldAccountInfoMethods:
     def set_auth_data(
@@ -76,7 +80,24 @@ class UrlPoolAccountInfo(v2.UrlPoolAccountInfo, AbstractAccountInfo):
 
 
 class SqliteAccountInfo(v2.SqliteAccountInfo, AbstractAccountInfo):
-    pass
+    def __init__(self, file_name=None, last_upgrade_to_run=None):
+        """
+        If ``file_name`` argument is empty or ``None``, path from ``B2_ACCOUNT_INFO`` environment variable is used. If that is not available, a default of ``~/.b2_account_info`` is used.
+
+        :param str file_name: The sqlite file to use; overrides the default.
+        :param int last_upgrade_to_run: For testing only, override the auto-update on the db.
+        """
+        # use legacy env var resolution, XDG not supported
+        self.thread_local = threading.local()
+        user_account_info_path = file_name or os.environ.get(
+            v2.B2_ACCOUNT_INFO_ENV_VAR, v2.B2_ACCOUNT_INFO_DEFAULT_FILE
+        )
+        self.filename = file_name or os.path.expanduser(user_account_info_path)
+        logger.debug('%s file path to use: %s', self.__class__.__name__, self.filename)
+        self._validate_database()
+        with self._get_connection() as conn:
+            self._create_tables(conn, last_upgrade_to_run)
+        super(v2.SqliteAccountInfo, self).__init__()
 
 
 class StubAccountInfo(v2.StubAccountInfo, AbstractAccountInfo):

test/unit/account_info/test_account_info.py

@@ -13,11 +13,12 @@
 import unittest.mock as mock
 import os
 import platform
+import shutil
 import tempfile
 
 import pytest
 
-from apiver_deps import AbstractAccountInfo, InMemoryAccountInfo, UploadUrlPool, SqliteAccountInfo
+from apiver_deps import AbstractAccountInfo, InMemoryAccountInfo, UploadUrlPool, SqliteAccountInfo, TempDir, B2_ACCOUNT_INFO_ENV_VAR
 from apiver_deps_exception import CorruptAccountInfo, MissingAccountData
 
 from .fixtures import *
@@ -288,12 +289,14 @@ def setUp(self, request):
             os.unlink(self.db_path)
         except OSError:
             pass
-        print('using %s' % self.db_path)
+        self.home = tempfile.mkdtemp()
+
         yield
         try:
             os.unlink(self.db_path)
         except OSError:
             pass
+        shutil.rmtree(self.home)
 
     def test_corrupted(self):
         """
@@ -331,8 +334,67 @@ def test_convert_from_json(self):
     def _make_info(self):
         return self._make_sqlite_account_info()
 
-    def _make_sqlite_account_info(self, last_upgrade_to_run=None):
+    def _make_sqlite_account_info(self, env=None, last_upgrade_to_run=None):
         """
         Returns a new SqliteAccountInfo that has just read the data from the file.
+
+        :param dict env: Override Environment variables.
         """
-        return SqliteAccountInfo(file_name=self.db_path, last_upgrade_to_run=None)
+        # Override HOME to ensure hermetic tests
+        with mock.patch('os.environ', env or {'HOME': self.home}):
+            return SqliteAccountInfo(
+                file_name=self.db_path if not env else None,
+                last_upgrade_to_run=last_upgrade_to_run,
+            )
+
+    def test_uses_default(self):
+        account_info = self._make_sqlite_account_info(
+            env={
+                'HOME': self.home,
+            }
+        )
+        actual_path = os.path.abspath(account_info.filename)
+        assert os.path.join(self.home, '.b2_account_info') == actual_path
+
+    def test_uses_xdg_config_home(self, apiver):
+        with TempDir() as d:
+            account_info = self._make_sqlite_account_info(
+                env={
+                    'HOME': self.home,
+                    'XDG_CONFIG_HOME': d,
+                }
+            )
+            if apiver in ['v0', 'v1']:
+                expected_path = os.path.abspath(os.path.join(self.home, '.b2_account_info'))
+            else:
+                assert os.path.exists(os.path.join(d, 'b2'))
+                expected_path = os.path.abspath(os.path.join(d, 'b2', 'account_info'))
+            actual_path = os.path.abspath(account_info.filename)
+            assert expected_path == actual_path
+
+    def test_uses_existing_file_and_ignores_xdg(self):
+        with TempDir() as d:
+            default_db_file_location = os.path.join(self.home, '.b2_account_info')
+            open(default_db_file_location, 'a').close()
+            account_info = self._make_sqlite_account_info(
+                env={
+                    'HOME': self.home,
+                    'XDG_CONFIG_HOME': d,
+                }
+            )
+            assert not os.path.exists(os.path.join(d, 'b2'))
+            actual_path = os.path.abspath(account_info.filename)
+            assert default_db_file_location == actual_path
+
+    def test_account_info_env_var_overrides_xdg_config_home(self):
+        with TempDir() as d:
+            account_info = self._make_sqlite_account_info(
+                env={
+                    'HOME': self.home,
+                    'XDG_CONFIG_HOME': d,
+                    B2_ACCOUNT_INFO_ENV_VAR: os.path.join(d, 'b2_account_info'),
+                }
+            )
+            expected_path = os.path.abspath(os.path.join(d, 'b2_account_info'))
+            actual_path = os.path.abspath(account_info.filename)
+            assert expected_path == actual_path