summaryrefslogtreecommitdiffstats
path: root/contrib/python/propcache/tests
diff options
context:
space:
mode:
Diffstat (limited to 'contrib/python/propcache/tests')
-rw-r--r--contrib/python/propcache/tests/conftest.py115
-rw-r--r--contrib/python/propcache/tests/test_api.py11
-rw-r--r--contrib/python/propcache/tests/test_cached_property.py226
-rw-r--r--contrib/python/propcache/tests/test_init.py43
-rw-r--r--contrib/python/propcache/tests/test_under_cached_property.py252
-rw-r--r--contrib/python/propcache/tests/ya.make11
6 files changed, 658 insertions, 0 deletions
diff --git a/contrib/python/propcache/tests/conftest.py b/contrib/python/propcache/tests/conftest.py
new file mode 100644
index 00000000000..ee43f339231
--- /dev/null
+++ b/contrib/python/propcache/tests/conftest.py
@@ -0,0 +1,115 @@
+import argparse
+from dataclasses import dataclass
+from functools import cached_property
+from importlib import import_module
+from types import ModuleType
+
+import pytest
+
+C_EXT_MARK = pytest.mark.c_extension
+
+
+@dataclass(frozen=True)
+class PropcacheImplementation:
+ """A facade for accessing importable propcache module variants.
+
+ An instance essentially represents a c-extension or a pure-python module.
+ The actual underlying module is accessed dynamically through a property and
+ is cached.
+
+ It also has a text tag depending on what variant it is, and a string
+ representation suitable for use in Pytest's test IDs via parametrization.
+ """
+
+ is_pure_python: bool
+ """A flag showing whether this is a pure-python module or a C-extension."""
+
+ @cached_property
+ def tag(self) -> str:
+ """Return a text representation of the pure-python attribute."""
+ return "pure-python" if self.is_pure_python else "c-extension"
+
+ @cached_property
+ def imported_module(self) -> ModuleType:
+ """Return a loaded importable containing a propcache variant."""
+ importable_module = "_helpers_py" if self.is_pure_python else "_helpers_c"
+ return import_module(f"propcache.{importable_module}")
+
+ def __str__(self) -> str:
+ """Render the implementation facade instance as a string."""
+ return f"{self.tag}-module"
+
+
+ scope="session",
+ params=(
+ pytest.param(
+ PropcacheImplementation(is_pure_python=False),
+ marks=C_EXT_MARK,
+ ),
+ PropcacheImplementation(is_pure_python=True),
+ ),
+ ids=str,
+)
+def propcache_implementation(request: pytest.FixtureRequest) -> PropcacheImplementation:
+ """Return a propcache variant facade."""
+ return request.param # type: ignore[no-any-return]
+
+
[email protected](scope="session")
+def propcache_module(
+ propcache_implementation: PropcacheImplementation,
+) -> ModuleType:
+ """Return a pre-imported module containing a propcache variant."""
+ return propcache_implementation.imported_module
+
+
+def pytest_addoption(
+ parser: pytest.Parser,
+ pluginmanager: pytest.PytestPluginManager,
+) -> None:
+ """Define a new ``--c-extensions`` flag.
+
+ This lets the callers deselect tests executed against the C-extension
+ version of the ``propcache`` implementation.
+ """
+ del pluginmanager
+ parser.addoption(
+ "--c-extensions", # disabled with `--no-c-extensions`
+ action=argparse.BooleanOptionalAction,
+ default=True,
+ dest="c_extensions",
+ help="Test C-extensions (on by default)",
+ )
+
+
+def pytest_collection_modifyitems(
+ session: pytest.Session,
+ config: pytest.Config,
+ items: list[pytest.Item],
+) -> None:
+ """Deselect tests against C-extensions when requested via CLI."""
+ test_c_extensions = config.getoption("--c-extensions") is True
+
+ if test_c_extensions:
+ return
+
+ selected_tests: list[pytest.Item] = []
+ deselected_tests: list[pytest.Item] = []
+
+ for item in items:
+ c_ext = item.get_closest_marker(C_EXT_MARK.name) is not None
+
+ target_items_list = deselected_tests if c_ext else selected_tests
+ target_items_list.append(item)
+
+ config.hook.pytest_deselected(items=deselected_tests)
+ items[:] = selected_tests
+
+
+def pytest_configure(config: pytest.Config) -> None:
+ """Declare the C-extension marker in config."""
+ config.addinivalue_line(
+ "markers",
+ f"{C_EXT_MARK.name}: tests running against the C-extension implementation.",
+ )
diff --git a/contrib/python/propcache/tests/test_api.py b/contrib/python/propcache/tests/test_api.py
new file mode 100644
index 00000000000..85beec3606a
--- /dev/null
+++ b/contrib/python/propcache/tests/test_api.py
@@ -0,0 +1,11 @@
+"""Test we do not break the public API."""
+
+from propcache import _helpers, api
+
+
+def test_api() -> None:
+ """Verify the public API is accessible."""
+ assert api.cached_property is not None
+ assert api.under_cached_property is not None
+ assert api.cached_property is _helpers.cached_property
+ assert api.under_cached_property is _helpers.under_cached_property
diff --git a/contrib/python/propcache/tests/test_cached_property.py b/contrib/python/propcache/tests/test_cached_property.py
new file mode 100644
index 00000000000..a46c715c31b
--- /dev/null
+++ b/contrib/python/propcache/tests/test_cached_property.py
@@ -0,0 +1,226 @@
+import gc
+import sys
+from collections.abc import Callable
+from operator import not_
+from typing import TYPE_CHECKING, Any, Protocol, TypeVar
+
+import pytest
+
+from propcache.api import cached_property
+
+IS_PYPY = hasattr(sys, "pypy_version_info")
+
+if sys.version_info >= (3, 11):
+ from typing import assert_type
+
+_T_co = TypeVar("_T_co", covariant=True)
+
+
+class APIProtocol(Protocol):
+ def cached_property(
+ self, func: Callable[[Any], _T_co]
+ ) -> cached_property[_T_co]: ...
+
+
+def test_cached_property(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+
+ @propcache_module.cached_property
+ def prop(self) -> int:
+ return 1
+
+ a = A()
+ if sys.version_info >= (3, 11):
+ assert_type(a.prop, int)
+ assert a.prop == 1
+
+
+def test_cached_property_without_cache(propcache_module: APIProtocol) -> None:
+ class A:
+
+ __slots__ = ()
+
+ def __init__(self) -> None:
+ pass
+
+ @propcache_module.cached_property
+ def prop(self) -> None:
+ """Mock property."""
+
+ a = A()
+
+ with pytest.raises(AttributeError):
+ a.prop = 123 # type: ignore[assignment]
+
+
+def test_cached_property_check_without_cache(propcache_module: APIProtocol) -> None:
+ class A:
+
+ __slots__ = ()
+
+ def __init__(self) -> None:
+ """Init."""
+
+ @propcache_module.cached_property
+ def prop(self) -> None:
+ """Mock property."""
+
+ a = A()
+ with pytest.raises((TypeError, AttributeError)):
+ assert a.prop == 1
+
+
+def test_cached_property_caching(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+
+ @propcache_module.cached_property
+ def prop(self) -> int:
+ """Docstring."""
+ return 1
+
+ a = A()
+ assert a.prop == 1
+
+
+def test_cached_property_class_docstring(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+
+ @propcache_module.cached_property
+ def prop(self) -> None:
+ """Docstring."""
+
+ if TYPE_CHECKING:
+ assert isinstance(A.prop, cached_property)
+ else:
+ assert isinstance(A.prop, propcache_module.cached_property)
+ assert "Docstring." == A.prop.__doc__
+
+
+def test_set_name(propcache_module: APIProtocol) -> None:
+ """Test that the __set_name__ method is called and checked."""
+
+ class A:
+
+ @propcache_module.cached_property
+ def prop(self) -> None:
+ """Docstring."""
+
+ A.prop.__set_name__(A, "prop")
+
+ match = r"Cannot assign the same cached_property to two "
+ with pytest.raises(TypeError, match=match):
+ A.prop.__set_name__(A, "something_else")
+
+
+def test_get_without_set_name(propcache_module: APIProtocol) -> None:
+ """Test that get without __set_name__ fails."""
+ cp = propcache_module.cached_property(not_)
+
+ class A:
+ """A class."""
+
+ A.cp = cp # type: ignore[attr-defined]
+ match = r"Cannot use cached_property instance "
+ with pytest.raises(TypeError, match=match):
+ _ = A().cp # type: ignore[attr-defined]
+
+
+def test_ensured_wrapped_function_is_accessible(propcache_module: APIProtocol) -> None:
+ """Test that the wrapped function can be accessed from python."""
+
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+
+ @propcache_module.cached_property
+ def prop(self) -> int:
+ """Docstring."""
+ return 1
+
+ a = A()
+ assert A.prop.func(a) == 1
+
+
[email protected](IS_PYPY, reason="PyPy has no C extension")
+def test_cached_property_no_refcount_leak(propcache_module: APIProtocol) -> None:
+ """Test that cached_property does not leak references."""
+
+ class CachedPropertySentinel:
+ """A unique object we can track."""
+
+ def count_sentinels() -> int:
+ """Count the number of CachedPropertySentinel instances in gc."""
+ gc.collect()
+ return sum(
+ 1 for obj in gc.get_objects() if isinstance(obj, CachedPropertySentinel)
+ )
+
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+
+ @propcache_module.cached_property
+ def prop(self) -> CachedPropertySentinel:
+ """Return a sentinel object."""
+ return CachedPropertySentinel()
+
+ initial_sentinel_count = count_sentinels()
+
+ a = A()
+
+ # First access - creates and caches the object
+ result = a.prop
+ # sys.getrefcount returns 1 higher than actual (the temp ref from the call)
+ # After first access: result owns 1, __dict__ owns 1, getrefcount call owns 1 = 3
+ initial_refcount = sys.getrefcount(result)
+
+ # Should have exactly 1 Sentinel instance now
+ assert count_sentinels() == initial_sentinel_count + 1
+
+ # Second access - should return the cached object without creating new refs
+ result2 = a.prop
+ assert result is result2
+ # After second access: result owns 1, result2 owns 1, __dict__ owns 1, getrefcount call owns 1 = 4
+ # Only result2 should add 1
+ second_refcount = sys.getrefcount(result)
+ assert second_refcount == initial_refcount + 1
+
+ # Still should have exactly 1 Sentinel instance
+ assert count_sentinels() == initial_sentinel_count + 1
+
+ # Third access
+ result3 = a.prop
+ assert result is result3
+ # result2 and result3 each add 1
+ third_refcount = sys.getrefcount(result)
+ assert third_refcount == initial_refcount + 2
+
+ # Clean up local refs - should be back to just result and __dict__
+ del result2
+ del result3
+ after_cleanup_refcount = sys.getrefcount(result)
+ assert after_cleanup_refcount == initial_refcount
+
+ # Clear the cache and verify no leak when re-fetching
+ # After clearing: only result owns it
+ del a.__dict__["prop"]
+ cleared_refcount = sys.getrefcount(result)
+ assert cleared_refcount == initial_refcount - 1 # No longer in __dict__
+
+ # Re-fetch - this should create a new object, not affect old one
+ result4 = a.prop
+ assert result4 is not result # Should be a new object
+ refetch_refcount = sys.getrefcount(result)
+ assert refetch_refcount == cleared_refcount # Original object refcount unchanged
+
+ # Now we should have 2 Sentinel instances:
+ # - original in `result`
+ # - new one in `result4`
+ assert count_sentinels() == initial_sentinel_count + 2
diff --git a/contrib/python/propcache/tests/test_init.py b/contrib/python/propcache/tests/test_init.py
new file mode 100644
index 00000000000..6bcffe7faa6
--- /dev/null
+++ b/contrib/python/propcache/tests/test_init.py
@@ -0,0 +1,43 @@
+"""Test imports can happen from top-level."""
+
+import pytest
+
+import propcache
+from propcache import _helpers
+
+
+def test_api_at_top_level() -> None:
+ """Verify the public API is accessible at top-level."""
+ assert propcache.cached_property is not None
+ assert propcache.under_cached_property is not None
+ assert propcache.cached_property is _helpers.cached_property
+ assert propcache.under_cached_property is _helpers.under_cached_property
+
+
+ "prop_name",
+ ("cached_property", "under_cached_property"),
+)
+def test_public_api_is_discoverable_in_dir(prop_name: str) -> None:
+ """Verify the public API is discoverable programmatically."""
+ assert prop_name in dir(propcache)
+
+
+def test_importing_invalid_attr_raises() -> None:
+ """Verify importing an invalid attribute raises an AttributeError."""
+ match = r"^module 'propcache' has no attribute 'invalid_attr'$"
+ with pytest.raises(AttributeError, match=match):
+ propcache.invalid_attr
+
+
+def test_import_error_invalid_attr() -> None:
+ """Verify importing an invalid attribute raises an ImportError."""
+ # No match here because the error is raised by the import system
+ # and may vary between Python versions.
+ with pytest.raises(ImportError):
+ from propcache import invalid_attr # noqa: F401
+
+
+def test_no_wildcard_imports() -> None:
+ """Verify wildcard imports are prohibited."""
+ assert not propcache.__all__
diff --git a/contrib/python/propcache/tests/test_under_cached_property.py b/contrib/python/propcache/tests/test_under_cached_property.py
new file mode 100644
index 00000000000..9c0dc304f65
--- /dev/null
+++ b/contrib/python/propcache/tests/test_under_cached_property.py
@@ -0,0 +1,252 @@
+import gc
+import sys
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any, Protocol, TypedDict, TypeVar
+
+import pytest
+
+from propcache.api import under_cached_property
+
+IS_PYPY = hasattr(sys, "pypy_version_info")
+
+if sys.version_info >= (3, 11):
+ from typing import assert_type
+
+_T_co = TypeVar("_T_co", covariant=True)
+
+
+class APIProtocol(Protocol):
+ def under_cached_property(
+ self, func: Callable[[Any], _T_co]
+ ) -> under_cached_property[_T_co]: ...
+
+
+def test_under_cached_property(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ self._cache: dict[str, int] = {}
+
+ @propcache_module.under_cached_property
+ def prop(self) -> int:
+ return 1
+
+ @propcache_module.under_cached_property
+ def prop2(self) -> str:
+ return "foo"
+
+ a = A()
+ if sys.version_info >= (3, 11):
+ assert_type(a.prop, int)
+ assert a.prop == 1
+ if sys.version_info >= (3, 11):
+ assert_type(a.prop2, str)
+ assert a.prop2 == "foo"
+
+
+def test_under_cached_property_typeddict(propcache_module: APIProtocol) -> None:
+ """Test static typing passes with TypedDict."""
+
+ class _Cache(TypedDict, total=False):
+ prop: int
+ prop2: str
+
+ class A:
+ def __init__(self) -> None:
+ self._cache: _Cache = {}
+
+ @propcache_module.under_cached_property
+ def prop(self) -> int:
+ return 1
+
+ @propcache_module.under_cached_property
+ def prop2(self) -> str:
+ return "foo"
+
+ a = A()
+ if sys.version_info >= (3, 11):
+ assert_type(a.prop, int)
+ assert a.prop == 1
+ if sys.version_info >= (3, 11):
+ assert_type(a.prop2, str)
+ assert a.prop2 == "foo"
+
+
+def test_under_cached_property_assignment(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ self._cache: dict[str, Any] = {}
+
+ @propcache_module.under_cached_property
+ def prop(self) -> None:
+ """Mock property."""
+
+ a = A()
+
+ with pytest.raises(AttributeError):
+ a.prop = 123 # type: ignore[assignment]
+
+
+def test_under_cached_property_without_cache(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+ self._cache: dict[str, int] = {}
+
+ @propcache_module.under_cached_property
+ def prop(self) -> None:
+ """Mock property."""
+
+ a = A()
+
+ with pytest.raises(AttributeError):
+ a.prop = 123 # type: ignore[assignment]
+
+
+def test_under_cached_property_check_without_cache(
+ propcache_module: APIProtocol,
+) -> None:
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+ # Note that self._cache is intentionally missing
+ # here to verify AttributeError
+
+ @propcache_module.under_cached_property
+ def prop(self) -> None:
+ """Mock property."""
+
+ a = A()
+ with pytest.raises(AttributeError):
+ _ = a.prop # type: ignore[call-overload]
+
+
+def test_under_cached_property_caching(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ self._cache: dict[str, int] = {}
+
+ @propcache_module.under_cached_property
+ def prop(self) -> int:
+ """Docstring."""
+ return 1
+
+ a = A()
+ assert a.prop == 1
+
+
+def test_under_cached_property_class_docstring(propcache_module: APIProtocol) -> None:
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+
+ @propcache_module.under_cached_property
+ def prop(self) -> None:
+ """Docstring."""
+
+ if TYPE_CHECKING:
+ # At type checking, the fixture doesn't represent the real module, so
+ # we use the global-level imported module to verify the isinstance() check here
+ # matches the behaviour users would see in real code.
+ assert isinstance(A.prop, under_cached_property)
+ else:
+ assert isinstance(A.prop, propcache_module.under_cached_property)
+ assert "Docstring." == A.prop.__doc__
+
+
+def test_ensured_wrapped_function_is_accessible(propcache_module: APIProtocol) -> None:
+ """Test that the wrapped function can be accessed from python."""
+
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+ self._cache: dict[str, int] = {}
+
+ @propcache_module.under_cached_property
+ def prop(self) -> int:
+ """Docstring."""
+ return 1
+
+ a = A()
+ assert A.prop.wrapped(a) == 1
+
+
[email protected](IS_PYPY, reason="PyPy has no C extension")
+def test_under_cached_property_no_refcount_leak(propcache_module: APIProtocol) -> None:
+ """Test that under_cached_property does not leak references."""
+
+ class UnderCachedPropertySentinel:
+ """A unique object we can track."""
+
+ def count_sentinels() -> int:
+ """Count the number of UnderCachedPropertySentinel instances in gc."""
+ gc.collect()
+ return sum(
+ 1
+ for obj in gc.get_objects()
+ if isinstance(obj, UnderCachedPropertySentinel)
+ )
+
+ class A:
+ def __init__(self) -> None:
+ """Init."""
+ self._cache: dict[str, Any] = {}
+
+ @propcache_module.under_cached_property
+ def prop(self) -> UnderCachedPropertySentinel:
+ """Return a sentinel object."""
+ return UnderCachedPropertySentinel()
+
+ initial_sentinel_count = count_sentinels()
+
+ a = A()
+
+ # First access - creates and caches the object
+ result = a.prop
+ # sys.getrefcount returns 1 higher than actual (the temp ref from the call)
+ # After first access: result owns 1, _cache owns 1, getrefcount call owns 1 = 3
+ initial_refcount = sys.getrefcount(result)
+
+ # Should have exactly 1 Sentinel instance now
+ assert count_sentinels() == initial_sentinel_count + 1
+
+ # Second access - should return the cached object without creating new refs
+ result2 = a.prop
+ assert result is result2
+ # After second access: result owns 1, result2 owns 1, _cache owns 1, getrefcount call owns 1 = 4
+ # Only result2 should add 1
+ second_refcount = sys.getrefcount(result)
+ assert second_refcount == initial_refcount + 1
+
+ # Still should have exactly 1 Sentinel instance
+ assert count_sentinels() == initial_sentinel_count + 1
+
+ # Third access
+ result3 = a.prop
+ assert result is result3
+ # result2 and result3 each add 1
+ third_refcount = sys.getrefcount(result)
+ assert third_refcount == initial_refcount + 2
+
+ # Clean up local refs - should be back to just result and _cache
+ del result2
+ del result3
+ after_cleanup_refcount = sys.getrefcount(result)
+ assert after_cleanup_refcount == initial_refcount
+
+ # Clear the cache and verify no leak when re-fetching
+ # After clearing: only result owns it
+ del a._cache["prop"]
+ cleared_refcount = sys.getrefcount(result)
+ assert cleared_refcount == initial_refcount - 1 # No longer in _cache
+
+ # Re-fetch - this should create a new object, not affect old one
+ result4 = a.prop
+ assert result4 is not result # Should be a new object
+ refetch_refcount = sys.getrefcount(result)
+ assert refetch_refcount == cleared_refcount # Original object refcount unchanged
+
+ # Now we should have 2 Sentinel instances:
+ # - original in `result`
+ # - new one in `result4`
+ assert count_sentinels() == initial_sentinel_count + 2
diff --git a/contrib/python/propcache/tests/ya.make b/contrib/python/propcache/tests/ya.make
new file mode 100644
index 00000000000..53173b0a58f
--- /dev/null
+++ b/contrib/python/propcache/tests/ya.make
@@ -0,0 +1,11 @@
+PY3TEST()
+
+PEERDIR(
+ contrib/python/propcache
+)
+
+ALL_PYTEST_SRCS()
+
+NO_LINT()
+
+END()