Extend test coverage

datahub/core/management/commands/rq_health_check.py

@@ -33,10 +33,10 @@ def handle(self, *args, **options):
             missing_queues = set([queue]) - set(queue_names)
 
             if missing_queues:
-                print(f'RQ queue not running: {missing_queues}')
+                logger.error(f'RQ queue not running: {missing_queues}')
                 sys.exit(1)
-            print('OK')
+            logger.info('OK')
             sys.exit(0)
 
-        print('Nothing checked! Please provide --queue parameter')
+        logger.error('Nothing checked! Please provide --queue parameter')
         sys.exit(1)

datahub/core/test/management/commands/test_rq_health_check.py

@@ -0,0 +1,63 @@
+import logging
+from unittest import mock
+from unittest.mock import patch
+
+import pytest
+
+from django.core.management import call_command
+
+
+class MockWorker:
+    """
+    Mock queue names object returned by worker
+    """
+
+    queue_name = ''
+
+    def __init__(self, queue_name, *args, **kwargs):
+        self.queue_name = queue_name
+
+    def queue_names(self):
+        return self.queue_name
+
+
+def test_rq_health_check_ok():
+    logger = logging.getLogger('datahub.core.management.commands.rq_health_check')
+    with patch(
+        'datahub.core.management.commands.rq_health_check.Worker.all',
+        return_value=[MockWorker(['short-running']), MockWorker(['long-running'])],
+    ):
+        with mock.patch.object(logger, 'info') as mock_info:
+            with pytest.raises(SystemExit) as exception_info:
+                call_command('rq_health_check', '--queue=short-running')
+
+            assert exception_info.value.code == 0
+            assert 'OK' in str(mock_info.call_args_list)
+            assert mock_info.call_count == 1
+
+
+def test_rq_health_check_rq_not_running():
+    logger = logging.getLogger('datahub.core.management.commands.rq_health_check')
+    with patch(
+        'datahub.core.management.commands.rq_health_check.Worker.all',
+        return_value=[MockWorker(['long-running'])],
+    ):
+        with mock.patch.object(logger, 'error') as mock_error:
+            with pytest.raises(SystemExit) as exception_info:
+                call_command('rq_health_check', '--queue=short-running')
+
+            assert exception_info.value.code == 1
+            assert "RQ queue not running: {\'short-running\'}" in str(mock_error.call_args_list)
+            assert mock_error.call_count == 1
+
+
+def test_command_called_without_parameter():
+    logger = logging.getLogger('datahub.core.management.commands.rq_health_check')
+    with mock.patch.object(logger, 'error') as mock_error:
+        with pytest.raises(SystemExit) as exception_info:
+            call_command('rq_health_check')
+
+        assert exception_info.value.code == 1
+        assert 'Nothing checked! Please provide --queue parameter' \
+            in str(mock_error.call_args_list)
+        assert mock_error.call_count == 1

datahub/ping/test/test_ping_view.py

@@ -10,33 +10,14 @@
 pytestmark = pytest.mark.django_db
 
 
-class MockWorker:
-    """
-    Mock queue names for worker
-    """
-
-    queue_name = ''
-
-    def __init__(self, queue_name, *args, **kwargs):
-        self.queue_name = queue_name
-
-    def queue_names(self):
-        return self.queue_name
-
-
 def test_all_good(client):
     """Test all good."""
     url = reverse('ping')
-    with patch(
-        'datahub.core.queues.health_check.Worker.all',
-        return_value=[MockWorker(['short-running']), MockWorker(['long-running'])],
-    ):
-
-        response = client.get(url)
+    response = client.get(url)
 
-        assert response.status_code == status.HTTP_200_OK
-        assert '<status>OK</status>' in str(response.content)
-        assert response.headers['content-type'] == 'text/xml'
+    assert response.status_code == status.HTTP_200_OK
+    assert '<status>OK</status>' in str(response.content)
+    assert response.headers['content-type'] == 'text/xml'
 
 
 def test_check_database_fail(client):
@@ -50,14 +31,3 @@ def test_check_database_fail(client):
         assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
         assert '<status>FALSE</status>' in str(response.content)
         assert response.headers['content-type'] == 'text/xml'
-
-
-def test_check_rq_workers_fail(client):
-    url = reverse('ping')
-    response = client.get(url)
-    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
-    assert '<status>FALSE</status>' in str(response.content)
-    assert '<!--RQ queue(s) not running:' in str(response.content)
-    assert 'long-running' in str(response.content)
-    assert 'short-running' in str(response.content)
-    assert response.headers['content-type'] == 'text/xml'