0001-commands-filter-on-module-name-in-cron-command-39090.patch
combo/data/management/commands/cron.py | ||
---|---|---|
16 | 16 | |
17 | 17 |
from __future__ import print_function |
18 | 18 | |
19 |
import sys |
|
19 | 20 |
import traceback |
20 | 21 | |
21 | 22 |
from django.apps import apps |
... | ... | |
31 | 32 |
help='limit updates to given application') |
32 | 33 | |
33 | 34 |
def handle(self, **options): |
35 |
stdout = options.get('stdout', sys.stdout) # for testing |
|
34 | 36 |
errors = [] |
35 | 37 |
for appconfig in apps.get_app_configs(): |
36 | 38 |
if not hasattr(appconfig, 'hourly'): |
37 | 39 |
continue |
38 |
if options.get('application') and appconfig.name != options.get('application'): |
|
40 |
if options.get('application') and appconfig.name.rsplit('.')[-1] != options.get('application'):
|
|
39 | 41 |
continue |
40 | 42 |
if hasattr(appconfig, 'is_enabled') and not appconfig.is_enabled(): |
41 | 43 |
continue |
... | ... | |
46 | 48 |
if errors: |
47 | 49 |
for error in errors: |
48 | 50 |
if options['verbosity'] >= 1: |
49 |
print('%s: error: %s' % (error['application'], error['exception'])) |
|
51 |
print('%s: error: %s' % (error['application'], error['exception']), file=stdout)
|
|
50 | 52 |
if options['verbosity'] >= 2: |
51 |
print(error['traceback']) |
|
52 |
print() |
|
53 |
print(error['traceback'], file=stdout)
|
|
54 |
print(file=stdout)
|
|
53 | 55 |
raise CommandError('error running jobs') |
tests/test_cron.py | ||
---|---|---|
1 |
from django.core.management import call_command |
|
2 |
from django.core.management.base import CommandError |
|
3 |
from django.utils.six import StringIO |
|
4 | ||
5 |
import mock |
|
6 |
import pytest |
|
7 | ||
8 |
pytestmark = pytest.mark.django_db |
|
9 | ||
10 | ||
11 |
def test_cron_run(app): |
|
12 |
with mock.patch('combo.apps.lingo.AppConfig.hourly') as hourly_job: |
|
13 |
call_command('cron') |
|
14 |
assert hourly_job.call_count == 1 |
|
15 | ||
16 |
call_command('cron', application='dashboard') |
|
17 |
assert hourly_job.call_count == 1 |
|
18 | ||
19 |
call_command('cron', application='lingo') |
|
20 |
assert hourly_job.call_count == 2 |
|
21 | ||
22 | ||
23 |
def test_cron_error(app): |
|
24 |
with mock.patch('combo.apps.lingo.AppConfig.hourly') as hourly_job: |
|
25 |
hourly_job.side_effect = Exception('test error') |
|
26 |
out = StringIO() |
|
27 |
with pytest.raises(CommandError): |
|
28 |
call_command('cron', application='lingo', verbosity=0, stdout=out) |
|
29 |
assert out.getvalue() == '' |
|
30 | ||
31 |
out = StringIO() |
|
32 |
with pytest.raises(CommandError): |
|
33 |
call_command('cron', application='lingo', verbosity=1, stdout=out) |
|
34 |
assert out.getvalue() == 'combo.apps.lingo: error: test error\n' |
|
35 | ||
36 |
out = StringIO() |
|
37 |
with pytest.raises(CommandError): |
|
38 |
call_command('cron', application='lingo', verbosity=2, stdout=out) |
|
39 |
assert out.getvalue().startswith('combo.apps.lingo: error: test error\n') |
|
40 |
assert 'Traceback' in out.getvalue() |
|
0 |
- |