sdk/python/tests/test_basedirs.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 import os
   6 import stat
   7
   8 import pytest
   9
  10 from pathlib import Path
  11
  12 from arvados._internal import basedirs
  13
  14 class TestBaseDirectories:
  15     SELF_PATH = Path(__file__)
  16
  17     @pytest.fixture
  18     def dir_spec(self, tmp_path):
  19         return basedirs.BaseDirectorySpec(
  20             'TEST_DIRECTORY',
  21             'XDG_TEST_HOME',
  22             Path('.test'),
  23             'XDG_TEST_DIRS',
  24             f"{tmp_path / '.test1'}:{tmp_path / '.test2'}",
  25         )
  26
  27     @pytest.fixture
  28     def env(self, tmp_path):
  29         return {'HOME': str(tmp_path)}
  30
  31     @pytest.fixture
  32     def umask(self):
  33         orig_umask = os.umask(0o002)
  34         try:
  35             yield
  36         finally:
  37             os.umask(orig_umask)
  38
  39     def test_search_systemd_dirs(self, dir_spec, env, tmp_path):
  40         env['TEST_DIRECTORY'] = f'{tmp_path}:{self.SELF_PATH.parent}'
  41         dirs = basedirs.BaseDirectories(dir_spec, env, 'tests')
  42         actual = list(dirs.search(self.SELF_PATH.name))
  43         assert actual == [self.SELF_PATH]
  44
  45     def test_search_xdg_home(self, dir_spec, env, tmp_path):
  46         env['XDG_TEST_HOME'] = str(self.SELF_PATH.parent.parent)
  47         dirs = basedirs.BaseDirectories(dir_spec, env, 'tests')
  48         actual = list(dirs.search(self.SELF_PATH.name))
  49         assert actual == [self.SELF_PATH]
  50
  51     def test_search_xdg_dirs(self, dir_spec, env, tmp_path):
  52         env['XDG_TEST_DIRS'] = f'{tmp_path}:{self.SELF_PATH.parent.parent}'
  53         dirs = basedirs.BaseDirectories(dir_spec, env, 'tests')
  54         actual = list(dirs.search(self.SELF_PATH.name))
  55         assert actual == [self.SELF_PATH]
  56
  57     def test_search_all_dirs(self, dir_spec, env, tmp_path):
  58         env['TEST_DIRECTORY'] = f'{tmp_path}:{self.SELF_PATH.parent}'
  59         env['XDG_TEST_HOME'] = str(self.SELF_PATH.parent.parent)
  60         env['XDG_TEST_DIRS'] = f'{tmp_path}:{self.SELF_PATH.parent.parent}'
  61         dirs = basedirs.BaseDirectories(dir_spec, env, 'tests')
  62         actual = list(dirs.search(self.SELF_PATH.name))
  63         assert actual == [self.SELF_PATH, self.SELF_PATH, self.SELF_PATH]
  64
  65     def test_search_default_home(self, dir_spec, env, tmp_path):
  66         expected = tmp_path / dir_spec.xdg_home_default / 'default_home'
  67         expected.parent.mkdir()
  68         expected.touch()
  69         dirs = basedirs.BaseDirectories(dir_spec, env, '.')
  70         actual = list(dirs.search(expected.name))
  71         assert actual == [expected]
  72
  73     def test_search_default_dirs(self, dir_spec, env, tmp_path):
  74         _, _, default_dir = dir_spec.xdg_dirs_default.rpartition(':')
  75         expected = Path(default_dir, 'default_dirs')
  76         expected.parent.mkdir()
  77         expected.touch()
  78         dirs = basedirs.BaseDirectories(dir_spec, env, '.')
  79         actual = list(dirs.search(expected.name))
  80         assert actual == [expected]
  81
  82     def test_search_no_default_dirs(self, dir_spec, env, tmp_path):
  83         dir_spec.xdg_dirs_key = None
  84         dir_spec.xdg_dirs_default = None
  85         for subdir in ['.test1', '.test2', dir_spec.xdg_home_default]:
  86             expected = tmp_path / subdir / 'no_dirs'
  87             expected.parent.mkdir()
  88             expected.touch()
  89         dirs = basedirs.BaseDirectories(dir_spec, env, '.')
  90         actual = list(dirs.search(expected.name))
  91         assert actual == [expected]
  92
  93     def test_ignore_relative_directories(self, dir_spec, env, tmp_path):
  94         test_path = Path(*self.SELF_PATH.parts[-2:])
  95         assert test_path.exists(), "test setup problem: need an existing file in a subdirectory of ."
  96         parent_path = str(test_path.parent)
  97         env['TEST_DIRECTORY'] = '.'
  98         env['XDG_TEST_HOME'] = parent_path
  99         env['XDG_TEST_DIRS'] = parent_path
 100         dirs = basedirs.BaseDirectories(dir_spec, env, parent_path)
 101         assert not list(dirs.search(test_path.name))
 102
 103     def test_search_warns_nondefault_home(self, dir_spec, env, tmp_path, caplog):
 104         search_path = tmp_path / dir_spec.xdg_home_default / 'Search' / 'SearchConfig'
 105         search_path.parent.mkdir(parents=True)
 106         search_path.touch()
 107         env[dir_spec.xdg_home_key] = str(tmp_path / '.nonexistent')
 108         dirs = basedirs.BaseDirectories(dir_spec, env, search_path.parent.name)
 109         results = list(dirs.search(search_path.name))
 110         expect_msg = "{} was not found under your configured ${} ({}), but does exist at the default location ({})".format(
 111             Path(*search_path.parts[-2:]),
 112             dir_spec.xdg_home_key,
 113             env[dir_spec.xdg_home_key],
 114             Path(*search_path.parts[:-2]),
 115         )
 116         assert caplog.messages
 117         assert any(msg.startswith(expect_msg) for msg in caplog.messages)
 118         assert not results
 119
 120     def test_storage_path_systemd(self, dir_spec, env, tmp_path):
 121         expected = tmp_path / 'rwsystemd'
 122         expected.mkdir(0o700)
 123         env['TEST_DIRECTORY'] = str(expected)
 124         dirs = basedirs.BaseDirectories(dir_spec, env)
 125         assert dirs.storage_path() == expected
 126
 127     def test_storage_path_systemd_mixed_modes(self, dir_spec, env, tmp_path):
 128         rodir = tmp_path / 'rodir'
 129         rodir.mkdir(0o500)
 130         expected = tmp_path / 'rwdir'
 131         expected.mkdir(0o700)
 132         env['TEST_DIRECTORY'] = f'{rodir}:{expected}'
 133         dirs = basedirs.BaseDirectories(dir_spec, env)
 134         assert dirs.storage_path() == expected
 135
 136     def test_storage_path_xdg_home(self, dir_spec, env, tmp_path):
 137         expected = tmp_path / '.xdghome' / 'arvados'
 138         env['XDG_TEST_HOME'] = str(expected.parent)
 139         dirs = basedirs.BaseDirectories(dir_spec, env)
 140         assert dirs.storage_path() == expected
 141         exp_mode = stat.S_IFDIR | stat.S_IWUSR
 142         assert (expected.stat().st_mode & exp_mode) == exp_mode
 143
 144     def test_storage_path_default(self, dir_spec, env, tmp_path):
 145         expected = tmp_path / dir_spec.xdg_home_default / 'arvados'
 146         dirs = basedirs.BaseDirectories(dir_spec, env)
 147         assert dirs.storage_path() == expected
 148         exp_mode = stat.S_IFDIR | stat.S_IWUSR
 149         assert (expected.stat().st_mode & exp_mode) == exp_mode
 150
 151     @pytest.mark.parametrize('subdir,mode', [
 152         ('str/dir', 0o750),
 153         (Path('sub', 'path'), 0o770),
 154     ])
 155     def test_storage_path_subdir(self, dir_spec, env, umask, tmp_path, subdir, mode):
 156         expected = tmp_path / dir_spec.xdg_home_default / 'arvados' / subdir
 157         dirs = basedirs.BaseDirectories(dir_spec, env)
 158         actual = dirs.storage_path(subdir, mode)
 159         assert actual == expected
 160         expect_mode = mode | stat.S_IFDIR
 161         actual_mode = actual.stat().st_mode
 162         assert (actual_mode & expect_mode) == expect_mode
 163         assert not (actual_mode & stat.S_IRWXO)
 164
 165     def test_empty_xdg_home(self, dir_spec, env, tmp_path):
 166         env['XDG_TEST_HOME'] = ''
 167         expected = tmp_path / dir_spec.xdg_home_default / 'emptyhome'
 168         dirs = basedirs.BaseDirectories(dir_spec, env, expected.name)
 169         assert dirs.storage_path() == expected
 170
 171     def test_empty_xdg_dirs(self, dir_spec, env, tmp_path):
 172         env['XDG_TEST_DIRS'] = ''
 173         _, _, default_dir = dir_spec.xdg_dirs_default.rpartition(':')
 174         expected = Path(default_dir, 'empty_dirs')
 175         expected.parent.mkdir()
 176         expected.touch()
 177         dirs = basedirs.BaseDirectories(dir_spec, env, '.')
 178         actual = list(dirs.search(expected.name))
 179         assert actual == [expected]
 180
 181     def test_spec_key_lookup(self):
 182         dirs = basedirs.BaseDirectories('CACHE')
 183         assert dirs._spec.systemd_key == 'CACHE_DIRECTORY'
 184         assert dirs._spec.xdg_dirs_key is None
 185
 186     def test_spec_enum_lookup(self):
 187         dirs = basedirs.BaseDirectories(basedirs.BaseDirectorySpecs.CONFIG)
 188         assert dirs._spec.systemd_key == 'CONFIGURATION_DIRECTORY'