5697667142

Committed 29 Jul 2023 01:09AM CUT coverage: 99.159%. First build

Build # 5697667142

Build Type

push

github

Committed by

chrisrink10

Commit Message

Use Tox 4.0 run command

Run Details

1634 of 1636 branches covered (99.88%)

Branch coverage included in aggregate %.

7686 of 7763 relevant lines covered (99.01%)

0.99 hits per line

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

92.75

/src/basilisp/lang/reference.py

from typing import Any, Callable, Optional, TypeVar

from readerwriterlock.rwlock import Lockable

from basilisp.lang import keyword as kw
from basilisp.lang import map as lmap
from basilisp.lang.exception import ExceptionInfo
from basilisp.lang.interfaces import (
    IPersistentMap,
    IRef,
    IReference,
    RefValidator,
    RefWatcher,
    RefWatchKey,
)

try:
    from typing import Protocol
except ImportError:
    AlterMeta = Callable[..., Optional[IPersistentMap]]
else:

    class AlterMeta(Protocol):  # type: ignore [no-redef]
        def __call__(
            self, meta: Optional[IPersistentMap], *args
        ) -> Optional[IPersistentMap]:
            ...


class ReferenceBase(IReference):
    """Mixin for IReference classes to define the full IReference interface.

    `basilisp.lang.runtime.Namespace` objects are the only objects which are
    `IReference` objects without also being `IRef` objects.

    Implementers must have the `_rlock`, `_wlock`, and `_meta` properties defined."""

    _rlock: Lockable
    _wlock: Lockable
    _meta: Optional[IPersistentMap]

    @property
    def meta(self) -> Optional[IPersistentMap]:
        with self._rlock:
            return self._meta

    def alter_meta(self, f: AlterMeta, *args) -> Optional[IPersistentMap]:
        with self._wlock:
            self._meta = f(self._meta, *args)
            return self._meta

    def reset_meta(self, meta: Optional[IPersistentMap]) -> Optional[IPersistentMap]:
        with self._wlock:
            self._meta = meta
            return meta


T = TypeVar("T")


class RefBase(IRef[T], ReferenceBase):
    """
    Mixin for IRef classes to define the full IRef interface.

    `IRef` objects are generally shared, mutable state objects such as Atoms and
    Vars.

    Implementers must have the `_validators` and `_watches` properties defined.
    """

    _validator: Optional[RefValidator]
    _watches: IPersistentMap

    def add_watch(self, k: RefWatchKey, wf: RefWatcher) -> "RefBase[T]":
        with self._wlock:
            self._watches = self._watches.assoc(k, wf)
            return self

    def _notify_watches(self, old: Any, new: Any):
        for k, wf in self._watches.items():
            wf(k, self, old, new)

    def remove_watch(self, k: RefWatchKey) -> "RefBase[T]":
        with self._wlock:
            self._watches = self._watches.dissoc(k)
            return self

    def get_validator(self) -> Optional[RefValidator]:
        return self._validator

    def set_validator(self, vf: Optional[RefValidator] = None) -> None:
        # We cannot use a write lock here since we're calling `self.deref()` which
        # attempts to acquire the read lock for the Ref and will deadlock if the
        # lock is not reentrant.
        #
        # There are no guarantees that the Ref lock is reentrant and the default
        # locks for Atoms and Vars are not).
        #
        # This is probably ok for most cases since we expect contention is low or
        # non-existent while setting a validator function.
        if vf is not None:
            self._validate(self.deref(), vf=vf)
        self._validator = vf

    def _validate(self, val: Any, vf: Optional[RefValidator] = None):
        vf = vf or self._validator
        if vf is not None:
            try:
                res = vf(val)
            except Exception:
                res = False

            if not res:
                raise ExceptionInfo(
                    "Invalid reference state",
                    lmap.map({kw.keyword("data"): val, kw.keyword("validator"): vf}),
                )

1	from typing import Any, Callable, Optional, TypeVar	1✔
2
3	from readerwriterlock.rwlock import Lockable	1✔
4
5	from basilisp.lang import keyword as kw	1✔
6	from basilisp.lang import map as lmap	1✔
7	from basilisp.lang.exception import ExceptionInfo	1✔
8	from basilisp.lang.interfaces import (	1✔
9	IPersistentMap,
10	IRef,
11	IReference,
12	RefValidator,
13	RefWatcher,
14	RefWatchKey,
15	)
16
17	try:	1✔
18	from typing import Protocol	1✔
19	except ImportError:	×
20	AlterMeta = Callable[..., Optional[IPersistentMap]]	×
21	else:
22
23	class AlterMeta(Protocol): # type: ignore [no-redef]	1✔
24	def __call__(	1✔
25	self, meta: Optional[IPersistentMap], *args
26	) -> Optional[IPersistentMap]:
27	...	×
28
29
30	class ReferenceBase(IReference):	1✔
31	"""Mixin for IReference classes to define the full IReference interface.
32
33	`basilisp.lang.runtime.Namespace` objects are the only objects which are
34	`IReference` objects without also being `IRef` objects.
35
36	Implementers must have the `_rlock`, `_wlock`, and `_meta` properties defined."""
37
38	_rlock: Lockable	1✔
39	_wlock: Lockable	1✔
40	_meta: Optional[IPersistentMap]	1✔
41
42	@property	1✔
43	def meta(self) -> Optional[IPersistentMap]:	1✔
44	with self._rlock:	1✔
45	return self._meta	1✔
46
47	def alter_meta(self, f: AlterMeta, *args) -> Optional[IPersistentMap]:	1✔
48	with self._wlock:	1✔
49	self._meta = f(self._meta, *args)	1✔
50	return self._meta	1✔
51
52	def reset_meta(self, meta: Optional[IPersistentMap]) -> Optional[IPersistentMap]:	1✔
53	with self._wlock:	1✔
54	self._meta = meta	1✔
55	return meta	1✔
56
57
58	T = TypeVar("T")	1✔
59
60
61	class RefBase(IRef[T], ReferenceBase):	1✔
62	"""
63	Mixin for IRef classes to define the full IRef interface.
64
65	`IRef` objects are generally shared, mutable state objects such as Atoms and
66	Vars.
67
68	Implementers must have the `_validators` and `_watches` properties defined.
69	"""
70
71	_validator: Optional[RefValidator]	1✔
72	_watches: IPersistentMap	1✔
73
74	def add_watch(self, k: RefWatchKey, wf: RefWatcher) -> "RefBase[T]":	1✔
75	with self._wlock:	1✔
76	self._watches = self._watches.assoc(k, wf)	1✔
77	return self	1✔
78
79	def _notify_watches(self, old: Any, new: Any):	1✔
80	for k, wf in self._watches.items():	1✔
81	wf(k, self, old, new)	1✔
82
83	def remove_watch(self, k: RefWatchKey) -> "RefBase[T]":	1✔
84	with self._wlock:	1✔
85	self._watches = self._watches.dissoc(k)	1✔
86	return self	1✔
87
88	def get_validator(self) -> Optional[RefValidator]:	1✔
89	return self._validator	1✔
90
91	def set_validator(self, vf: Optional[RefValidator] = None) -> None:	1✔
92	# We cannot use a write lock here since we're calling `self.deref()` which
93	# attempts to acquire the read lock for the Ref and will deadlock if the
94	# lock is not reentrant.
95	#
96	# There are no guarantees that the Ref lock is reentrant and the default
97	# locks for Atoms and Vars are not).
98	#
99	# This is probably ok for most cases since we expect contention is low or
100	# non-existent while setting a validator function.
101	if vf is not None:	1✔
102	self._validate(self.deref(), vf=vf)	1✔
103	self._validator = vf	1✔
104
105	def _validate(self, val: Any, vf: Optional[RefValidator] = None):	1✔
106	vf = vf or self._validator	1✔
107	if vf is not None:	1✔
108	try:	1✔
109	res = vf(val)	1✔
110	except Exception:	×
111	res = False	×
112
113	if not res:	1✔
114	raise ExceptionInfo(	1✔
115	"Invalid reference state",
116	lmap.map({kw.keyword("data"): val, kw.keyword("validator"): vf}),
117	)

basilisp-lang / basilisp / 5697667142

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