11323269743

Committed 14 Oct 2024 07:55AM UTC coverage: 64.919% (+0.3%) from 64.591%

Build # 11323269743

Build Type

push

github

Committed by

LudovicRousseau

Commit Message

Test and improve the synchronization code

Added
-----

* Add tests to bring `Synchronization.py` to 100% coverage
* Add type annotations to the synchronization primitives
* Add `typing_extensions` as a Python 3.9 requirement;
  this is needed to annotate that the `synchronized()` decorator
  takes a function with a certain set of parameter and return types
  and returns a function with the same parameter and return types

Changed
-------

* Make the synchronized function's `self` parameter explicit;
  this allows the `self` parameter to be type-annotated
  so the dependency on the `self.mutex` attribute is explicit
* Use the `self.mutex` lock as a context manager
* Support keyword arguments to synchronized functions

Fixed
-----

* Wrap synchronized functions correctly;
  previous behavior was to lose the function name and docstring

Removed
-------

* Remove `print()` lines that are commented out
* Remove a `bytes` instance check; method names can only be strings

Run Details

100 of 466 branches covered (21.46%)

21 of 22 new or added lines in 1 file covered. (95.45%)

34 existing lines in 22 files now uncovered.

4210 of 6485 relevant lines covered (64.92%)

4.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

100.0

/src/smartcard/Synchronization.py

"""
from Thinking in Python, Bruce Eckel
https://python-3-patterns-idioms-test.readthedocs.io/en/latest/Observer.html

(c) Copyright 2008, Creative Commons Attribution-Share Alike 3.0.

Simple emulation of Java's 'synchronized'
keyword, from Peter Norvig.
"""

from __future__ import annotations

import functools
import sys
import threading
from collections.abc import Iterable
from typing import Any, Callable, Protocol, TypeVar

if sys.version_info >= (3, 10):
    from typing import ParamSpec
else:
    from typing_extensions import ParamSpec


T = TypeVar("T")
P = ParamSpec("P")


def synchronized(method: Callable[P, T]) -> Callable[P, T]:
    @functools.wraps(method)
    def f(self: _SynchronizationProtocol, *args: Any, **kwargs: Any) -> Any:
        with self.mutex:
            return method(self, *args, **kwargs)

    return f


def synchronize(klass: type, names: str | Iterable[str] | None = None) -> None:
    """Synchronize methods in the given class.
    Only synchronize the methods whose names are
    given, or all methods if names=None."""

    if isinstance(names, str):
        names = names.split()
    for name, val in list(klass.__dict__.items()):
        if callable(val) and name != "__init__" and (names is None or name in names):
            setattr(klass, name, synchronized(val))


class _SynchronizationProtocol(Protocol):
    mutex: threading.Lock | threading.RLock


class Synchronization(_SynchronizationProtocol):
    # You can create your own self.mutex, or inherit from this class:

    def __init__(self):
        self.mutex = threading.RLock()

1	"""
2	from Thinking in Python, Bruce Eckel
3	https://python-3-patterns-idioms-test.readthedocs.io/en/latest/Observer.html
4
5	(c) Copyright 2008, Creative Commons Attribution-Share Alike 3.0.
6
7	Simple emulation of Java's 'synchronized'
8	keyword, from Peter Norvig.
9	"""
10
11	from __future__ import annotations	20✔
12
13	import functools	20✔
14	import sys	20✔
15	import threading	20✔
16	from collections.abc import Iterable	20✔
17	from typing import Any, Callable, Protocol, TypeVar	20✔
18
19	if sys.version_info >= (3, 10):	20✔
20	from typing import ParamSpec	16✔
21	else:
NEW 22	from typing_extensions import ParamSpec	4✔
23
24
25	T = TypeVar("T")	20✔
26	P = ParamSpec("P")	20✔
27
28
29	def synchronized(method: Callable[P, T]) -> Callable[P, T]:	20✔
30	@functools.wraps(method)	20✔
31	def f(self: _SynchronizationProtocol, args: Any, *kwargs: Any) -> Any:	20✔
32	with self.mutex:	17✔
33	return method(self, args, *kwargs)	17✔
34
35	return f	20✔
36
37
38	def synchronize(klass: type, names: str \| Iterable[str] \| None = None) -> None:	20✔
39	"""Synchronize methods in the given class.
40	Only synchronize the methods whose names are
41	given, or all methods if names=None."""
42
43	if isinstance(names, str):	20✔
44	names = names.split()	20✔
45	for name, val in list(klass.__dict__.items()):	20✔
46	if callable(val) and name != "__init__" and (names is None or name in names):	20✔
47	setattr(klass, name, synchronized(val))	20✔
48
49
50	class _SynchronizationProtocol(Protocol):	20✔
51	mutex: threading.Lock \| threading.RLock	20✔
52
53
54	class Synchronization(_SynchronizationProtocol):	20✔
55	# You can create your own self.mutex, or inherit from this class:
56
57	def __init__(self):	20✔
58	self.mutex = threading.RLock()	17✔

LudovicRousseau / pyscard / 11323269743

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous