10798992192

Committed 10 Sep 2024 07:03PM CUT coverage: 98.807% (-0.09%) from 98.892%

Build # 10798992192

Build Type

Pull #1044

github

Committed by

web-flow

Commit Message

Merge bc24d2bbb into ae617c7b3

Pull Request Pull Request #1044: test runner for basilisp.test #980

Run Details

1286 of 1294 branches covered (99.38%)

Branch coverage included in aggregate %.

8570 of 8681 relevant lines covered (98.72%)

0.99 hits per line

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

96.92

/src/basilisp/lang/reference.py

import threading
from typing import Any, Callable, Optional, TypeVar

from typing_extensions import Concatenate, ParamSpec

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,
)

P = ParamSpec("P")
AlterMeta = Callable[Concatenate[Optional[IPersistentMap], P], 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 `_lock` and `_meta` properties defined."""

    _lock: threading.RLock
    _meta: Optional[IPersistentMap]

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

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

    def reset_meta(self, meta: Optional[IPersistentMap]) -> Optional[IPersistentMap]:
        with self._lock:
            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[T]]
    _watches: IPersistentMap[RefWatchKey, RefWatcher[T]]

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

    def _notify_watches(self, old: T, new: T) -> None:
        for k, wf in self._watches.items():
            wf(k, self, old, new)

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

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

    def set_validator(self, vf: Optional[RefValidator[T]] = None) -> None:
        with self._lock:
            if vf is not None:
                self._validate(self.deref(), vf=vf)
            self._validator = vf

    def _validate(self, val: Any, vf: Optional[RefValidator[T]] = None) -> None:
        vf = vf or self._validator
        if vf is not None:
            try:
                res = vf(val)
            except Exception:  # pylint: disable=broad-exception-caught
                res = False

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

1	import threading	1✔
2	from typing import Any, Callable, Optional, TypeVar	1✔
3
4	from typing_extensions import Concatenate, ParamSpec	1✔
5
6	from basilisp.lang import keyword as kw	1✔
7	from basilisp.lang import map as lmap	1✔
8	from basilisp.lang.exception import ExceptionInfo	1✔
9	from basilisp.lang.interfaces import (	1✔
10	IPersistentMap,
11	IRef,
12	IReference,
13	RefValidator,
14	RefWatcher,
15	RefWatchKey,
16	)
17
18	P = ParamSpec("P")	1✔
19	AlterMeta = Callable[Concatenate[Optional[IPersistentMap], P], Optional[IPersistentMap]]	1✔
20
21
22	class ReferenceBase(IReference):	1✔
23	"""Mixin for IReference classes to define the full IReference interface.
24
25	`basilisp.lang.runtime.Namespace` objects are the only objects which are
26	`IReference` objects without also being `IRef` objects.
27
28	Implementers must have the `_lock` and `_meta` properties defined."""
29
30	_lock: threading.RLock	1✔
31	_meta: Optional[IPersistentMap]	1✔
32
33	@property	1✔
34	def meta(self) -> Optional[IPersistentMap]:	1✔
35	with self._lock:	1✔
36	return self._meta	1✔
37
38	def alter_meta(self, f: AlterMeta, *args) -> Optional[IPersistentMap]:	1✔
39	with self._lock:	1✔
40	self._meta = f(self._meta, *args)	1✔
41	return self._meta	1✔
42
43	def reset_meta(self, meta: Optional[IPersistentMap]) -> Optional[IPersistentMap]:	1✔
44	with self._lock:	1✔
45	self._meta = meta	1✔
46	return meta	1✔
47
48
49	T = TypeVar("T")	1✔
50
51
52	class RefBase(IRef[T], ReferenceBase):	1✔
53	"""
54	Mixin for IRef classes to define the full IRef interface.
55
56	`IRef` objects are generally shared, mutable state objects such as Atoms and
57	Vars.
58
59	Implementers must have the `_validators` and `_watches` properties defined.
60	"""
61
62	_validator: Optional[RefValidator[T]]	1✔
63	_watches: IPersistentMap[RefWatchKey, RefWatcher[T]]	1✔
64
65	def add_watch(self, k: RefWatchKey, wf: RefWatcher[T]) -> "RefBase[T]":	1✔
66	with self._lock:	1✔
67	self._watches = self._watches.assoc(k, wf)	1✔
68	return self	1✔
69
70	def _notify_watches(self, old: T, new: T) -> None:	1✔
71	for k, wf in self._watches.items():	1✔
72	wf(k, self, old, new)	1✔
73
74	def remove_watch(self, k: RefWatchKey) -> "RefBase[T]":	1✔
75	with self._lock:	1✔
76	self._watches = self._watches.dissoc(k)	1✔
77	return self	1✔
78
79	def get_validator(self) -> Optional[RefValidator[T]]:	1✔
80	return self._validator	1✔
81
82	def set_validator(self, vf: Optional[RefValidator[T]] = None) -> None:	1✔
83	with self._lock:	1✔
84	if vf is not None:	1✔
85	self._validate(self.deref(), vf=vf)	1✔
86	self._validator = vf	1✔
87
88	def _validate(self, val: Any, vf: Optional[RefValidator[T]] = None) -> None:	1✔
89	vf = vf or self._validator	1✔
90	if vf is not None:	1✔
91	try:	1✔
92	res = vf(val)	1✔
93	except Exception: # pylint: disable=broad-exception-caught	×
94	res = False	×
95
96	if not res:	1✔
97	raise ExceptionInfo(	1✔
98	"Invalid reference state",
99	lmap.map({kw.keyword("data"): val, kw.keyword("validator"): vf}),
100	)

basilisp-lang / basilisp / 10798992192

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