commands: filter on module name in cron command (#39090)

(instead of dotted name, e.g. lingo vs combo.apps.lingo)

combo/data/management/commands/cron.py

@@ -16,6 +16,7 @@
 
 from __future__ import print_function
 
+import sys
 import traceback
 
 from django.apps import apps
@@ -31,11 +32,12 @@ class Command(BaseCommand):
                 help='limit updates to given application')
 
     def handle(self, **options):
+        stdout = options.get('stdout', sys.stdout)  # for testing
         errors = []
         for appconfig in apps.get_app_configs():
             if not hasattr(appconfig, 'hourly'):
                 continue
-            if options.get('application') and appconfig.name != options.get('application'):
+            if options.get('application') and appconfig.name.rsplit('.')[-1] != options.get('application'):
                 continue
             if hasattr(appconfig, 'is_enabled') and not appconfig.is_enabled():
                 continue
@@ -46,8 +48,8 @@ class Command(BaseCommand):
         if errors:
             for error in errors:
                 if options['verbosity'] >= 1:
-                    print('%s: error: %s' % (error['application'], error['exception']))
+                    print('%s: error: %s' % (error['application'], error['exception']), file=stdout)
                 if options['verbosity'] >= 2:
-                    print(error['traceback'])
-                    print()
+                    print(error['traceback'], file=stdout)
+                    print(file=stdout)
             raise CommandError('error running jobs')

tests/test_cron.py

@@ -0,0 +1,40 @@
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.utils.six import StringIO
+
+import mock
+import pytest
+
+pytestmark = pytest.mark.django_db
+
+
+def test_cron_run(app):
+    with mock.patch('combo.apps.lingo.AppConfig.hourly') as hourly_job:
+        call_command('cron')
+        assert hourly_job.call_count == 1
+
+        call_command('cron', application='dashboard')
+        assert hourly_job.call_count == 1
+
+        call_command('cron', application='lingo')
+        assert hourly_job.call_count == 2
+
+
+def test_cron_error(app):
+    with mock.patch('combo.apps.lingo.AppConfig.hourly') as hourly_job:
+        hourly_job.side_effect = Exception('test error')
+        out = StringIO()
+        with pytest.raises(CommandError):
+            call_command('cron', application='lingo', verbosity=0, stdout=out)
+        assert out.getvalue() == ''
+
+        out = StringIO()
+        with pytest.raises(CommandError):
+            call_command('cron', application='lingo', verbosity=1, stdout=out)
+        assert out.getvalue() == 'combo.apps.lingo: error: test error\n'
+
+        out = StringIO()
+        with pytest.raises(CommandError):
+            call_command('cron', application='lingo', verbosity=2, stdout=out)
+        assert out.getvalue().startswith('combo.apps.lingo: error: test error\n')
+        assert 'Traceback' in out.getvalue()