tests.checker: Migrate to pytest

swh/deposit/loader/checker.py

@@ -5,8 +5,12 @@
 
 import logging
 
+from typing import Mapping
 
-from ..client import PrivateApiDepositClient
+from swh.deposit.client import PrivateApiDepositClient
+
+
+logger = logging.getLogger(__name__)
 
 
 class DepositChecker():
@@ -18,16 +22,14 @@ class DepositChecker():
     def __init__(self, client=None):
         super().__init__()
         self.client = client if client else PrivateApiDepositClient()
-        logging_class = '%s.%s' % (self.__class__.__module__,
-                                   self.__class__.__name__)
-        self.log = logging.getLogger(logging_class)
 
-    def check(self, deposit_check_url):
+    def check(self, deposit_check_url: str) -> Mapping[str, str]:
+        status = None
         try:
-            self.client.check(deposit_check_url)
+            r = self.client.check(deposit_check_url)
+            status = 'eventful' if r == 'verified' else 'failed'
         except Exception:
-            self.log.exception("Failure during check on '%s'" % (
+            logger.exception("Failure during check on '%s'" % (
                 deposit_check_url, ))
-            return {'status': 'failed'}
-        else:
-            return {'status': 'eventful'}
+            status = 'failed'
+        return {'status': status}

swh/deposit/tests/conftest.py

@@ -3,6 +3,7 @@
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
+import os
 import base64
 import pytest
 import psycopg2
@@ -35,6 +36,16 @@ TEST_USER = {
 }
 
 
+@pytest.fixture(autouse=True, scope='session')
+def swh_proxy():
+    """Automatically inject this fixture in all tests to ensure no outside
+       connection takes place.
+
+    """
+    os.environ['http_proxy'] = 'http://localhost:999'
+    os.environ['https_proxy'] = 'http://localhost:999'
+
+
 def execute_sql(sql):
     """Execute sql to postgres db"""
     with psycopg2.connect(database='postgres') as conn:

swh/deposit/tests/loader/conftest.py

@@ -3,8 +3,12 @@
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
+import os
 import pytest
+import yaml
+
 from swh.scheduler.tests.conftest import *  # noqa
+from swh.deposit.loader.checker import DepositChecker
 
 
 @pytest.fixture(scope='session')
@@ -12,3 +16,21 @@ def celery_includes():
     return [
         'swh.deposit.loader.tasks',
     ]
+
+
+@pytest.fixture
+def swh_config(tmp_path, monkeypatch):
+    storage_config = {
+        'url': 'https://deposit.softwareheritage.org/',
+    }
+
+    conffile = os.path.join(tmp_path, 'deposit.yml')
+    with open(conffile, 'w') as f:
+        f.write(yaml.dump(storage_config))
+    monkeypatch.setenv('SWH_CONFIG_FILENAME', conffile)
+    return conffile
+
+
+@pytest.fixture
+def deposit_checker(swh_config):
+    return DepositChecker()

swh/deposit/tests/loader/data/https_deposit.softwareheritage.org/1_private_test_1_check

+{"status": "verified"}

swh/deposit/tests/loader/data/https_deposit.softwareheritage.org/1_private_test_2_check

+{
+  "status": "rejected"
+}

swh/deposit/tests/loader/test_checker.py

@@ -3,66 +3,43 @@
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
-from rest_framework.test import APITestCase
-
-from swh.deposit.models import Deposit
-from swh.deposit.config import PRIVATE_CHECK_DEPOSIT, DEPOSIT_STATUS_VERIFIED
-from swh.deposit.config import DEPOSIT_STATUS_REJECTED
-from swh.deposit.loader.checker import DepositChecker
 from django.urls import reverse
+from unittest.mock import patch
 
+from swh.deposit.config import PRIVATE_CHECK_DEPOSIT
 
-from .common import SWHDepositTestClient, CLIENT_TEST_CONFIG
-from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
-from ..common import FileSystemCreationRoutine
-
-
-class DepositCheckerScenarioTest(APITestCase, WithAuthTestCase,
-                                 BasicTestCase, CommonCreationRoutine,
-                                 FileSystemCreationRoutine):
-
-    def setUp(self):
-        super().setUp()
-
-        # 2. Sets a basic client which accesses the test data
-        checker_client = SWHDepositTestClient(client=self.client,
-                                              config=CLIENT_TEST_CONFIG)
-        # 3. setup loader with no persistence and that client
-        self.checker = DepositChecker(client=checker_client)
-
-    def test_check_deposit_ready(self):
-        """Check on a valid 'deposited' deposit should result in 'verified'
 
-        """
-        # 1. create a deposit with archive and metadata
-        deposit_id = self.create_simple_binary_deposit()
-        deposit_id = self.update_binary_deposit(deposit_id,
-                                                status_partial=False)
+def test_check_deposit_ready(
+        swh_config, requests_mock_datadir, deposit_checker):
+    """Check on a valid 'deposited' deposit should result in 'verified'
 
-        args = [self.collection.name, deposit_id]
-        deposit_check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=args)
+    """
+    deposit_check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=['test', 1])
+    actual_result = deposit_checker.check(deposit_check_url=deposit_check_url)
+    assert actual_result == {'status': 'eventful'}
 
-        # when
-        actual_result = self.checker.check(deposit_check_url=deposit_check_url)
-        # then
-        deposit = Deposit.objects.get(pk=deposit_id)
-        self.assertEqual(deposit.status, DEPOSIT_STATUS_VERIFIED)
-        self.assertEqual(actual_result, {'status': 'eventful'})
 
-    def test_check_deposit_rejected(self):
-        """Check on invalid 'deposited' deposit should result in 'rejected'
+def test_check_deposit_rejected(
+        swh_config, requests_mock_datadir, deposit_checker):
+    """Check on invalid 'deposited' deposit should result in 'rejected'
 
-        """
-        # 1. create a deposit with archive and metadata
-        deposit_id = self.create_deposit_with_invalid_archive()
+    """
+    deposit_check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=[
+        'test', 2
+    ])
+    actual_result = deposit_checker.check(deposit_check_url=deposit_check_url)
+    assert actual_result == {'status': 'failed'}
 
-        args = [self.collection.name, deposit_id]
-        deposit_check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=args)
 
-        # when
-        actual_result = self.checker.check(deposit_check_url=deposit_check_url)
+@patch('swh.deposit.client.requests.get')
+def test_check_deposit_rejected_exception(
+        mock_requests, swh_config, deposit_checker):
+    """Check on invalid 'deposited' deposit should result in 'rejected'
 
-        # then
-        deposit = Deposit.objects.get(pk=deposit_id)
-        self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED)
-        self.assertEqual(actual_result, {'status': 'eventful'})
+    """
+    mock_requests.side_effect = ValueError('simulated problem when checking')
+    deposit_check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=[
+        'test', 3
+    ])
+    actual_result = deposit_checker.check(deposit_check_url=deposit_check_url)
+    assert actual_result == {'status': 'failed'}