Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add common utilities #6

Merged
merged 1 commit into from
Sep 14, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/sghi/utils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
ensure_not_none_nor_empty,
ensure_predicate,
)
from .others import future_succeeded, type_fqn

__all__ = [
"ensure_greater_or_equal",
Expand All @@ -20,4 +21,6 @@
"ensure_not_none",
"ensure_not_none_nor_empty",
"ensure_predicate",
"future_succeeded",
"type_fqn",
]
44 changes: 44 additions & 0 deletions src/sghi/utils/others.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
"""Other useful utilities."""
from collections.abc import Callable
from concurrent.futures import Future
from typing import Any

from .checkers import ensure_not_none


def future_succeeded(future: Future[Any]) -> bool:
"""
Check if a :external+python:py:class:`~concurrent.futures.Future` completed
successfully and return ``True`` if so, or ``False`` otherwise.

In this context, a ``Future`` is considered to have completed successfully
if it wasn't canceled and no uncaught exceptions were raised by its callee.

:param future: A ``Future`` instance to check for successful completion.
This MUST not be ``None``.

:return: ``True`` if the future completed successfully, ``False``
otherwise.

:raises ValueError: If ``future`` is ``None``.
"""
ensure_not_none(future, "'future' MUST not be None.")
return bool(
future.done()
and not future.cancelled()
and future.exception() is None,
)


def type_fqn(klass: type[Any] | Callable[..., Any]) -> str:
"""Return the fully qualified name of a type or callable.

:param klass: A type or callable whose fully qualified name is to be
determined. This MUST not be ``None``.

:return: The fully qualified name of the given type/callable.

:raises ValueError: If ``klass`` is ``None``.
"""
ensure_not_none(klass, "'klass' MUST not be None.")
return ".".join((klass.__module__, klass.__qualname__))
99 changes: 99 additions & 0 deletions test/sghi/utils_tests/others_tests.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
from concurrent.futures import Future

import pytest

from sghi.disposable import Disposable, ResourceDisposedError, not_disposed
from sghi.utils import future_succeeded, type_fqn


def test_future_succeeded_fails_on_none_input() -> None:
"""
:func:`future_succeeded` should raise a ``ValueError`` when given a
``None`` as it's input.
"""

with pytest.raises(ValueError, match="MUST not be None") as exc_info:
future_succeeded(None) # type: ignore

assert exc_info.value.args[0] == "'future' MUST not be None."


def test_future_succeeded_return_value_when_given_cancelled_futures() -> None:
"""
:func:`future_succeeded` should return ``False`` when given a canceled
``Future`` as it's input.
"""
future: Future[int] = Future()

assert future.cancel()
assert not future_succeeded(future)


def test_future_succeeded_return_value_when_given_failed_futures() -> None:
"""
:func:`future_succeeded` should return ``False`` when given a ``Future``
whose callee raised an exception as it's input.
"""
future: Future[int] = Future()
future.set_exception(ValueError(":("))

assert future.exception() is not None
assert not future_succeeded(future)


def test_future_succeeded_return_value_when_given_successful_futures() -> None:
"""
:func:`future_succeeded` should return ``True`` when given a ``Future``
that completed without any errors as it's input.
"""
future: Future[int] = Future()
future.set_result(10)

assert future.result() == 10
assert future_succeeded(future)


def test_type_fqn_return_value_on_first_party_types() -> None:
"""
:func:`type_fqn` should return the correct full qualified name when
given a first party (part of the current project) type or function.
"""
assert type_fqn(Disposable) == "sghi.disposable.Disposable"
assert type_fqn(ResourceDisposedError) == "sghi.disposable.ResourceDisposedError" # noqa: E501
assert type_fqn(not_disposed) == "sghi.disposable.not_disposed"
assert type_fqn(type_fqn) == "sghi.utils.others.type_fqn"


def test_type_fqn_return_value_on_standard_lib_types() -> None:
"""
:func:`type_fqn` should return the correct full qualified name when
given a standard library type or function.
"""
assert type_fqn(str) == "builtins.str"
assert type_fqn(dict) == "builtins.dict"
assert type_fqn(repr) == "builtins.repr"
assert type_fqn(round) == "builtins.round"


def test_type_fqn_return_value_on_third_party_types() -> None:
"""
:func:`type_fqn` should return the correct full qualified name when
given a third party (third party library) type or function.
"""

# FIXME: This might break on upgrade of pytest
assert type_fqn(pytest.approx) == "_pytest.python_api.approx"
assert type_fqn(pytest.importorskip) == "_pytest.outcomes.importorskip"
assert type_fqn(pytest.raises) == "_pytest.python_api.raises"


def test_type_fqn_fails_on_none_input() -> None:
"""
:func:`type_fqn` should raise a ``ValueError`` when given a ``None`` as
it's input.
"""

with pytest.raises(ValueError, match="MUST not be None") as exc_info:
type_fqn(None) # type: ignore

assert exc_info.value.args[0] == "'klass' MUST not be None."