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
158
159
160
161
162
163
164
165
166
167
168
169
170
|
"""Test for the Feed Manager."""
import datetime
from unittest import mock
import pytest
from georss_client.feed_manager import FeedManagerBase
from tests import MockGeoRssFeed
from tests.utils import load_fixture
HOME_COORDINATES_1 = (-31.0, 151.0)
HOME_COORDINATES_2 = (-37.0, 150.0)
@mock.patch("requests.Request")
@mock.patch("requests.Session")
def test_feed_manager(mock_session, mock_request):
"""Test the feed manager."""
mock_session.return_value.__enter__.return_value.send.return_value.ok = True
mock_session.return_value.__enter__.return_value.send.return_value.text = (
load_fixture("generic_feed_1.xml")
)
feed = MockGeoRssFeed(HOME_COORDINATES_1, None)
# This will just record calls and keep track of external ids.
generated_entity_external_ids = []
updated_entity_external_ids = []
removed_entity_external_ids = []
def _generate_entity(external_id):
"""Generate new entity."""
generated_entity_external_ids.append(external_id)
def _update_entity(external_id):
"""Update entity."""
updated_entity_external_ids.append(external_id)
def _remove_entity(external_id):
"""Remove entity."""
removed_entity_external_ids.append(external_id)
feed_manager = FeedManagerBase(
feed, _generate_entity, _update_entity, _remove_entity
)
assert (
repr(feed_manager) == "<FeedManagerBase("
"feed=<MockGeoRssFeed(home="
"(-31.0, 151.0), url=None, "
"radius=None, categories=None)>)>"
)
feed_manager.update()
entries = feed_manager.feed_entries
assert entries is not None
assert len(entries) == 5
assert feed_manager.last_update is not None
assert feed_manager.last_timestamp == datetime.datetime(2018, 9, 23, 9, 10)
assert len(generated_entity_external_ids) == 5
assert len(updated_entity_external_ids) == 0
assert len(removed_entity_external_ids) == 0
feed_entry = entries.get("1234")
assert feed_entry.title == "Title 1"
assert feed_entry.external_id == "1234"
assert feed_entry.coordinates == (-37.2345, 149.1234)
assert feed_entry.distance_to_home == pytest.approx(714.4, 0.1)
assert repr(feed_entry) == "<FeedEntry(id=1234)>"
feed_entry = entries.get("2345")
assert feed_entry.title == "Title 2"
assert feed_entry.external_id == "2345"
feed_entry = entries.get("Title 3")
assert feed_entry.title == "Title 3"
assert feed_entry.external_id == "Title 3"
external_id = hash((-37.8901, 149.7890))
feed_entry = entries.get(external_id)
assert feed_entry.title is None
assert feed_entry.external_id == external_id
feed_entry = entries.get("5678")
assert feed_entry.title == "Title 5"
assert feed_entry.external_id == "5678"
# Simulate an update with several changes.
generated_entity_external_ids.clear()
updated_entity_external_ids.clear()
removed_entity_external_ids.clear()
mock_session.return_value.__enter__.return_value.send.return_value.text = (
load_fixture("generic_feed_4.xml")
)
feed_manager.update()
entries = feed_manager.feed_entries
assert entries is not None
assert len(entries) == 3
assert len(generated_entity_external_ids) == 1
assert len(updated_entity_external_ids) == 2
assert len(removed_entity_external_ids) == 3
feed_entry = entries.get("1234")
assert feed_entry.title == "Title 1 UPDATED"
feed_entry = entries.get("2345")
assert feed_entry.title == "Title 2"
feed_entry = entries.get("6789")
assert feed_entry.title == "Title 6"
# Simulate an update with no data.
generated_entity_external_ids.clear()
updated_entity_external_ids.clear()
removed_entity_external_ids.clear()
mock_session.return_value.__enter__.return_value.send.return_value.ok = False
feed_manager.update()
entries = feed_manager.feed_entries
assert len(entries) == 0
assert len(generated_entity_external_ids) == 0
assert len(updated_entity_external_ids) == 0
assert len(removed_entity_external_ids) == 3
@mock.patch("requests.Request")
@mock.patch("requests.Session")
def test_feed_manager_no_timestamp(mock_session, mock_request):
"""Test updating feed is ok."""
mock_session.return_value.__enter__.return_value.send.return_value.ok = True
mock_session.return_value.__enter__.return_value.send.return_value.text = (
load_fixture("generic_feed_5.xml")
)
feed = MockGeoRssFeed(HOME_COORDINATES_1, None)
# This will just record calls and keep track of external ids.
generated_entity_external_ids = []
updated_entity_external_ids = []
removed_entity_external_ids = []
def _generate_entity(external_id):
"""Generate new entity."""
generated_entity_external_ids.append(external_id)
def _update_entity(external_id):
"""Update entity."""
updated_entity_external_ids.append(external_id)
def _remove_entity(external_id):
"""Remove entity."""
removed_entity_external_ids.append(external_id)
feed_manager = FeedManagerBase(
feed, _generate_entity, _update_entity, _remove_entity
)
assert (
repr(feed_manager) == "<FeedManagerBase("
"feed=<MockGeoRssFeed(home="
"(-31.0, 151.0), url=None, "
"radius=None, categories=None)>)>"
)
feed_manager.update()
entries = feed_manager.feed_entries
assert entries is not None
assert len(entries) == 1
assert feed_manager.last_timestamp is None
|
