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
|
# Copyright 2013 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
from contextlib import contextmanager
from mongoqueue import MongoQueue
from mock import (
create_autospec,
patch,
)
from pyelasticsearch import ElasticSearch
from charmworld.testing import logger_buffer
from charmworld.jobs.config import settings
from charmworld.jobs.worker import (
QueueWorker,
DEFAULT_JOBS,
main,
)
from charmworld.jobs.ingest import IngestJob
from charmworld.jobs.tests.test_ingest import charm_update_environment
from charmworld.search import ElasticSearchClient
from charmworld.testing import (
factory,
MongoTestBase,
)
class SuccessfulRun(IngestJob):
def run(self, charm_data):
charm_data['ingested'] = True
class FailureRun(IngestJob):
def run(self, charm_data):
if charm_data['charm'] == 'foo':
raise Exception('Exception forced by job.')
else:
charm_data['ingested'] = True
class GatheringRun(IngestJob):
def setup(self):
self.charms = []
def run(self, charm_data):
self.charms.append(charm_data)
@contextmanager
def main_environment(charm, index_client, queue):
"""An environment in which worker.main() can be run."""
with patch('charmworld.jobs.worker.configure_logging'):
with patch('sys.argv', ['foo', '--quit-on-empty']):
with patch('charmworld.jobs.worker.get_queue',
lambda x: queue):
with charm_update_environment(charm, index_client):
yield
class WorkerTest(MongoTestBase):
def setUp(self):
super(WorkerTest, self).setUp()
self.queue = MongoQueue(self.db.queue, 'queue')
self.queue.put({'charm': 'foo'})
self.queue.put({'charm': 'bar'})
def tearDown(self):
self.queue.clear()
super(WorkerTest, self).tearDown()
def test_basic_running(self):
# Barring errors, the worker will go through its entire set of
# ingest jobs for a charm.
ingest_jobs = (
SuccessfulRun(),
GatheringRun(),
)
queue_jobs = {
self.queue: ingest_jobs
}
worker = QueueWorker(queue_jobs)
worker.run(quit_on_empty=True)
self.assertEqual(0, self.queue.size())
expected = [
{'ingested': True, 'charm': 'foo'},
{'ingested': True, 'charm': 'bar'},
]
charms = ingest_jobs[1].charms
self.assertEqual(expected, charms)
def test_failure_while_running(self):
ingest_jobs = (FailureRun(), GatheringRun())
queue_jobs = {
self.queue: ingest_jobs
}
worker = QueueWorker(queue_jobs)
worker.run(quit_on_empty=True)
self.assertEqual(0, self.queue.size())
expected = [
{'ingested': True, 'charm': 'bar'},
]
charms = ingest_jobs[1].charms
self.assertEqual(expected, charms)
def test_stop_when_next_item_none(self):
class LockedJobQueue:
"""Queue representing the state when a locked jobs exists."""
@staticmethod
def next():
"""A locked job cannot be processed."""
return None
@staticmethod
def size():
"""If a queue has a locked job, its length > 0"""
return 1
ingest_jobs = (
SuccessfulRun(),
GatheringRun(),
)
worker = QueueWorker({LockedJobQueue: ingest_jobs})
worker.run(quit_on_empty=True)
def test_ingest_job_contract(self):
# All ingest jobs accept the right inputs.
ingest_jobs = [create_autospec(job)() for job in DEFAULT_JOBS]
worker = QueueWorker({self.queue: ingest_jobs})
worker.run(quit_on_empty=True)
self.assertEqual(0, self.queue.size())
def test_interval_is_int(self):
worker = QueueWorker({self.queue: []})
self.assertIs(int, type(worker.interval),
'QueueWorker.interval is a %s, not an int.' %
type(worker.interval))
def run_main_with_exit(self):
with self.assertRaises(SystemExit) as e:
main()
return e.exception.code
def test_elasticsearch_failure(self):
payload = factory.get_payload_json()
self.queue.clear()
self.queue.put(payload)
charm = factory.get_charm_json(payload=payload)
index_client = ElasticSearchClient(
ElasticSearch('http://localhost:70'), 'foo')
with patch.dict(settings,
values={'es_urls': 'http://localhost:70'}):
with main_environment(charm, index_client, self.queue):
with logger_buffer('charm.worker') as handler:
exit_code = self.run_main_with_exit()
self.assertEqual(40, handler.buffer[1].levelno)
self.assertIn('Could not connect', handler.buffer[1].msg)
self.assertEqual(1, exit_code)
self.assertEqual(0, self.queue.size())
|