2019-09-21 22:04:46 +02:00
|
|
|
import os
|
2022-06-07 11:52:38 +02:00
|
|
|
from pathlib import Path
|
2021-08-30 00:43:28 +02:00
|
|
|
from typing import Iterator
|
2021-02-10 11:38:21 +01:00
|
|
|
from unittest.mock import Mock
|
2019-09-21 22:04:46 +02:00
|
|
|
|
|
|
|
import pytest
|
|
|
|
from pip._vendor.cachecontrol.caches import FileCache
|
|
|
|
|
|
|
|
from pip._internal.network.cache import SafeFileCache
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
2021-08-30 00:43:28 +02:00
|
|
|
def cache_tmpdir(tmpdir: Path) -> Iterator[Path]:
|
2019-09-21 22:04:46 +02:00
|
|
|
cache_dir = tmpdir.joinpath("cache")
|
|
|
|
cache_dir.mkdir(parents=True)
|
|
|
|
yield cache_dir
|
|
|
|
|
|
|
|
|
|
|
|
class TestSafeFileCache:
|
|
|
|
"""
|
|
|
|
The no_perms test are useless on Windows since SafeFileCache uses
|
|
|
|
pip._internal.utils.filesystem.check_path_owner which is based on
|
|
|
|
os.geteuid which is absent on Windows.
|
|
|
|
"""
|
|
|
|
|
2021-08-30 00:43:28 +02:00
|
|
|
def test_cache_roundtrip(self, cache_tmpdir: Path) -> None:
|
2022-06-07 11:52:38 +02:00
|
|
|
cache = SafeFileCache(os.fspath(cache_tmpdir))
|
2019-09-21 22:04:46 +02:00
|
|
|
assert cache.get("test key") is None
|
|
|
|
cache.set("test key", b"a test string")
|
2023-10-19 00:14:22 +02:00
|
|
|
# Body hasn't been stored yet, so the entry isn't valid yet
|
|
|
|
assert cache.get("test key") is None
|
|
|
|
|
|
|
|
# With a body, the cache entry is valid:
|
|
|
|
cache.set_body("test key", b"body")
|
2019-09-21 22:04:46 +02:00
|
|
|
assert cache.get("test key") == b"a test string"
|
|
|
|
cache.delete("test key")
|
|
|
|
assert cache.get("test key") is None
|
|
|
|
|
2022-05-23 19:06:38 +02:00
|
|
|
def test_cache_roundtrip_body(self, cache_tmpdir: Path) -> None:
|
|
|
|
cache = SafeFileCache(os.fspath(cache_tmpdir))
|
|
|
|
assert cache.get_body("test key") is None
|
|
|
|
cache.set_body("test key", b"a test string")
|
2023-10-19 00:14:22 +02:00
|
|
|
# Metadata isn't available, so the entry isn't valid yet (this
|
|
|
|
# shouldn't happen, but just in case)
|
|
|
|
assert cache.get_body("test key") is None
|
|
|
|
|
|
|
|
# With metadata, the cache entry is valid:
|
|
|
|
cache.set("test key", b"metadata")
|
2022-05-24 18:11:34 +02:00
|
|
|
body = cache.get_body("test key")
|
|
|
|
assert body is not None
|
2022-05-24 21:11:16 +02:00
|
|
|
with body:
|
|
|
|
assert body.read() == b"a test string"
|
2022-05-23 19:06:38 +02:00
|
|
|
cache.delete("test key")
|
|
|
|
assert cache.get_body("test key") is None
|
|
|
|
|
2019-09-21 22:04:46 +02:00
|
|
|
@pytest.mark.skipif("sys.platform == 'win32'")
|
2021-08-30 00:43:28 +02:00
|
|
|
def test_safe_get_no_perms(
|
|
|
|
self, cache_tmpdir: Path, monkeypatch: pytest.MonkeyPatch
|
|
|
|
) -> None:
|
2019-09-21 22:04:46 +02:00
|
|
|
os.chmod(cache_tmpdir, 000)
|
|
|
|
|
|
|
|
monkeypatch.setattr(os.path, "exists", lambda x: True)
|
|
|
|
|
2022-06-07 11:52:38 +02:00
|
|
|
cache = SafeFileCache(os.fspath(cache_tmpdir))
|
2019-09-21 22:04:46 +02:00
|
|
|
cache.get("foo")
|
|
|
|
|
|
|
|
@pytest.mark.skipif("sys.platform == 'win32'")
|
2021-08-30 00:43:28 +02:00
|
|
|
def test_safe_set_no_perms(self, cache_tmpdir: Path) -> None:
|
2019-09-21 22:04:46 +02:00
|
|
|
os.chmod(cache_tmpdir, 000)
|
|
|
|
|
2022-06-07 11:52:38 +02:00
|
|
|
cache = SafeFileCache(os.fspath(cache_tmpdir))
|
2019-09-21 22:04:46 +02:00
|
|
|
cache.set("foo", b"bar")
|
|
|
|
|
|
|
|
@pytest.mark.skipif("sys.platform == 'win32'")
|
2021-08-30 00:43:28 +02:00
|
|
|
def test_safe_delete_no_perms(self, cache_tmpdir: Path) -> None:
|
2019-09-21 22:04:46 +02:00
|
|
|
os.chmod(cache_tmpdir, 000)
|
|
|
|
|
2022-06-07 11:52:38 +02:00
|
|
|
cache = SafeFileCache(os.fspath(cache_tmpdir))
|
2019-09-21 22:04:46 +02:00
|
|
|
cache.delete("foo")
|
|
|
|
|
2021-08-30 00:43:28 +02:00
|
|
|
def test_cache_hashes_are_same(self, cache_tmpdir: Path) -> None:
|
2022-06-07 11:52:38 +02:00
|
|
|
cache = SafeFileCache(os.fspath(cache_tmpdir))
|
2019-09-21 22:04:46 +02:00
|
|
|
key = "test key"
|
|
|
|
fake_cache = Mock(FileCache, directory=cache.directory, encode=FileCache.encode)
|
|
|
|
assert cache._get_cache_path(key) == FileCache._fn(fake_cache, key)
|