1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
# pylint: disable=missing-module-docstring
# pylint: disable=missing-class-docstring
# pylint: disable=too-few-public-methods
import unittest.mock
import pytest
import smartcard.Observer
def test_state_changes():
"""Verify that the "changed" state can be set and cleared."""
# Base state
observable = smartcard.Observer.Observable()
assert not observable.hasChanged(), "Observable must not start in a changed state"
# Set
observable.setChanged()
assert observable.hasChanged(), ".setChanged() must *set* the state"
observable.setChanged()
assert observable.hasChanged(), ".setChanged() must not *toggle* the state"
# Clear
observable.clearChanged()
assert not observable.hasChanged(), ".clearChanged() must *clear* the state"
observable.clearChanged()
assert not observable.hasChanged(), ".clearChanged() must not *toggle* the state"
@pytest.mark.parametrize(
"unbound_method",
(
# Observer management methods
smartcard.Observer.Observable.addObserver,
smartcard.Observer.Observable.countObservers,
smartcard.Observer.Observable.deleteObserver,
smartcard.Observer.Observable.deleteObservers,
smartcard.Observer.Observable.notifyObservers,
# State management methods
smartcard.Observer.Observable.clearChanged,
smartcard.Observer.Observable.hasChanged,
smartcard.Observer.Observable.setChanged,
),
)
def test_synchronization(unbound_method):
"""Verify that all Observable methods are synchronized.
The `.mutex` attribute is mocked and is confirmed to be acquired and released.
The test implementation here doesn't necessarily make valid method calls;
it relies on the knowledge that most Observable methods are decorated
and that the lock is acquired before the underlying method is called
(and thus, before Python realizes that required method parameters are unfilled
and raises a TypeError).
"""
mutex = unittest.mock.MagicMock()
# Ensure the context manager protocol is functional.
mutex.__enter__.return_value = mutex
observable = smartcard.Observer.Observable()
observable.mutex = mutex
try:
# Call the unbound Observable method, passing *observable* as `self`.
unbound_method(observable) # noqa
except TypeError:
# Ignore TypeErrors caused by missing mandatory method arguments.
pass
mutex.__enter__.assert_called_once()
mutex.__exit__.assert_called_once()
def test_registered_observers_are_always_notified():
"""Verify all observers are notified, even if the observer list changes."""
class JealousObserver(smartcard.Observer.Observer):
update_count = 0
def update(self, observable, _):
# Try to prevent all other observers from receiving an update.
for other_observer in observers - {self}:
try:
observable.deleteObserver(other_observer)
except ValueError:
pass
# Track the number of times an update was received.
JealousObserver.update_count += 1
jealous_observer_count = 10
observers = {JealousObserver() for _ in range(jealous_observer_count)}
# Create a fair observable and notify the observers of a change.
fair_observable = smartcard.Observer.Observable()
for observer in observers:
fair_observable.addObserver(observer)
fair_observable.setChanged()
fair_observable.notifyObservers()
# Confirm that the observers have all erased each other from the list of observers.
assert fair_observable.countObservers() == 0
# Confirm that the observers were all updated, despite their jealous efforts.
assert JealousObserver.update_count == jealous_observer_count
def test_double_observer_additions():
"""Verify that an observer cannot be added twice."""
observer = smartcard.Observer.Observer()
observable = smartcard.Observer.Observable()
observable.addObserver(observer)
observable.addObserver(observer)
assert observable.countObservers() == 1
def test_double_observer_removals():
"""Verify that an observer cannot be removed twice."""
observer = smartcard.Observer.Observer()
observable = smartcard.Observer.Observable()
observable.addObserver(observer)
observable.deleteObserver(observer)
with pytest.raises(ValueError, match="x not in list"):
observable.deleteObserver(observer)
def test_no_notifications_when_no_changes():
"""Verify that no notifications are sent when no changes are detected."""
class Observer(smartcard.Observer.Observer):
def update(self, *_, **__):
raise RuntimeError("no updates were expected")
observable = smartcard.Observer.Observable()
observable.addObserver(Observer())
# Attempt to send a notification when no changes have been made.
# Nothing needs to be asserted here;
# the lack of a RuntimeError during test execution demonstrates the test passed.
observable.notifyObservers()
def test_default_observer_updates():
"""Call the default Observer.update() method.
This test exists solely for the purposes of reaching 100% test coverage.
It can be removed if Observer becomes an abstract base class.
"""
observable = smartcard.Observer.Observable()
assert smartcard.Observer.Observer().update(observable, "arg") is None
|
