|
1
|
|
|
import pytest |
|
2
|
|
|
|
|
3
|
|
|
@pytest.fixture |
|
4
|
|
|
def subject(): |
|
5
|
|
|
from so_magic.utils import Subject |
|
6
|
|
|
return Subject |
|
7
|
|
|
|
|
8
|
|
|
|
|
9
|
|
|
@pytest.fixture |
|
10
|
|
|
def observer(): |
|
11
|
|
|
from so_magic.utils import Observer |
|
12
|
|
|
return Observer |
|
13
|
|
|
|
|
14
|
|
|
|
|
15
|
|
|
def test_attrs_sanity(): |
|
16
|
|
|
import attr |
|
17
|
|
|
|
|
18
|
|
|
@attr.s |
|
19
|
|
|
class A: |
|
20
|
|
|
_observers = attr.ib(init=True, default=[]) |
|
21
|
|
|
|
|
22
|
|
|
i1 = A() |
|
23
|
|
|
i2 = A() |
|
24
|
|
|
assert id(i1._observers) == id(i2._observers) # SOS |
|
25
|
|
|
|
|
26
|
|
|
|
|
27
|
|
|
def test_attrs_sanity1(): |
|
28
|
|
|
import attr |
|
29
|
|
|
|
|
30
|
|
|
@attr.s |
|
31
|
|
|
class A: |
|
32
|
|
|
_observers = attr.ib(init=True) |
|
33
|
|
|
def add(self, *objects): |
|
34
|
|
|
self._observers.extend([_ for _ in objects]) |
|
35
|
|
|
i1 = A([]) |
|
36
|
|
|
i2 = A([]) |
|
37
|
|
|
assert id(i1._observers) != id(i2._observers) |
|
38
|
|
|
i1.add('obs1') |
|
39
|
|
|
assert i1._observers == ['obs1'] |
|
40
|
|
|
assert i2._observers == [] |
|
41
|
|
|
assert id(i1._observers) != id(i2._observers) |
|
42
|
|
|
|
|
43
|
|
|
|
|
44
|
|
|
def test_observers_sanity_test1(subject): |
|
45
|
|
|
subject1 = subject([]) |
|
46
|
|
|
subject2 = subject([]) |
|
47
|
|
|
assert id(subject1._observers) != id(subject2._observers) |
|
48
|
|
|
|
|
49
|
|
|
|
|
50
|
|
|
def test_observer_as_constructor(observer): |
|
51
|
|
|
with pytest.raises(TypeError) as instantiation_from_interface_error: |
|
52
|
|
|
observer_instance = observer() |
|
53
|
|
|
assert "Can't instantiate abstract class Observer with abstract methods update" in str(instantiation_from_interface_error.value) |
|
54
|
|
|
|
|
55
|
|
|
|
|
56
|
|
|
def test_scenario(subject, observer): |
|
57
|
|
|
# The client code. |
|
58
|
|
|
|
|
59
|
|
|
print("------ Scenario 1 ------\n") |
|
60
|
|
|
class ObserverA(observer): |
|
61
|
|
|
def update(self, a_subject) -> None: |
|
62
|
|
|
if a_subject.state == 0: |
|
63
|
|
|
print("ObserverA: Reacted to the event") |
|
64
|
|
|
|
|
65
|
|
|
s1 = subject([]) |
|
66
|
|
|
# o1 = observer() |
|
67
|
|
|
# s1.attach(o1) |
|
68
|
|
|
|
|
69
|
|
|
# business logic |
|
70
|
|
|
s1.state = 0 |
|
71
|
|
|
s1.notify() |
|
72
|
|
|
|
|
73
|
|
|
print("------ Scenario 2 ------\n") |
|
74
|
|
|
# example 2 |
|
75
|
|
|
class Businessubject(subject): |
|
76
|
|
|
|
|
77
|
|
|
def some_business_logic(self) -> None: |
|
78
|
|
|
""" |
|
79
|
|
|
Usually, the subscription logic is only a fraction of what a Subject can |
|
80
|
|
|
really do. Subjects commonly hold some important business logic, that |
|
81
|
|
|
triggers a notification method whenever something important is about to |
|
82
|
|
|
happen (or after it). |
|
83
|
|
|
""" |
|
84
|
|
|
print("\nSubject: I'm doing something important.") |
|
85
|
|
|
from random import randrange |
|
86
|
|
|
self._state = randrange(0, 10) |
|
87
|
|
|
print(f"Subject: My state has just changed to: {self._state}") |
|
88
|
|
|
self.notify() |
|
89
|
|
|
|
|
90
|
|
|
class ObserverB(observer): |
|
91
|
|
|
def update(self, a_subject) -> None: |
|
92
|
|
|
if a_subject.state == 0 or a_subject.state >= 2: |
|
93
|
|
|
print("ObserverB: Reacted to the event") |
|
94
|
|
|
|
|
95
|
|
|
s2 = Businessubject([]) |
|
96
|
|
|
assert id(s1) != id(s2) |
|
97
|
|
|
assert id(s1._observers) != id(s2._observers) |
|
98
|
|
|
o1, o2 = ObserverA(), ObserverB() |
|
99
|
|
|
# s2.attach(o1) |
|
100
|
|
|
# s2.attach(o2) |
|
101
|
|
|
s2.add(o1, o2) |
|
102
|
|
|
# business logic |
|
103
|
|
|
print(s2._observers) |
|
104
|
|
|
s2.some_business_logic() |
|
105
|
|
|
s2.some_business_logic() |
|
106
|
|
|
|
|
107
|
|
|
s2.detach(o1) |
|
108
|
|
|
s2.some_business_logic() |
|
109
|
|
|
|
boromir674 /
so-magic
