diff --git a/Makefile.local b/Makefile.local index 462b203d..8af9ffde 100644 --- a/Makefile.local +++ b/Makefile.local @@ -1,30 +1,27 @@ FLAKEFLAGS='--exclude=swh/deposit/manage.py,swh/deposit/settings.py,swh/deposit/migrations/' MANAGE=python3 -m swh.deposit.manage db-drop: dropdb swh-deposit-dev || return 0 db-create: db-drop createdb swh-deposit-dev db-prepare: $(MANAGE) makemigrations db-migrate: $(MANAGE) migrate db-load-data: $(MANAGE) loaddata deposit_data db-load-private-data: db-load-data $(MANAGE) loaddata ../private_data.yaml run-dev: $(MANAGE) runserver run: gunicorn3 -b 127.0.0.1:5006 swh.deposit.wsgi - -test: - ./swh/deposit/manage.py test diff --git a/swh/deposit/tests/api/test_converters.py b/swh/deposit/tests/api/test_converters.py index bfc5f039..6802ba1c 100644 --- a/swh/deposit/tests/api/test_converters.py +++ b/swh/deposit/tests/api/test_converters.py @@ -1,122 +1,123 @@ # Copyright (C) 2017-2019 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from swh.deposit.api.converters import convert_status_detail def test_convert_status_detail_empty(): for status_detail in [{}, {'dummy-keys': []}, None]: assert convert_status_detail(status_detail) is None def test_convert_status_detail(): status_detail = { 'url': { 'summary': "At least one url field must be compatible with the client\'s domain name. The following url fields failed the check", # noqa 'fields': ['blahurl', 'testurl'], }, 'metadata': [ { 'summary': 'Mandatory fields missing', 'fields': ['url', 'title'], }, { 'summary': 'Alternate fields missing', 'fields': ['name or title', 'url or badurl'] } ], 'archive': [{ 'summary': 'Unreadable archive', 'fields': ['1'], }], } expected_status_detail = '''- Mandatory fields missing (url, title) - Alternate fields missing (name or title, url or badurl) - Unreadable archive (1) - At least one url field must be compatible with the client's domain name. The following url fields failed the check (blahurl, testurl) ''' # noqa actual_status_detail = convert_status_detail(status_detail) assert actual_status_detail == expected_status_detail def test_convert_status_detail_2(): status_detail = { 'url': { 'summary': 'At least one compatible url field. Failed', 'fields': ['testurl'], }, 'metadata': [ { 'summary': 'Mandatory fields missing', 'fields': ['name'], }, ], 'archive': [ { 'summary': 'Invalid archive', 'fields': ['2'], }, { 'summary': 'Unsupported archive', 'fields': ['1'], } ], } expected_status_detail = '''- Mandatory fields missing (name) - Invalid archive (2) - Unsupported archive (1) - At least one compatible url field. Failed (testurl) ''' actual_status_detail = convert_status_detail(status_detail) assert actual_status_detail == expected_status_detail def test_convert_status_detail_3(): status_detail = { 'url': { 'summary': 'At least one compatible url field', }, } expected_status_detail = '- At least one compatible url field\n' actual_status_detail = convert_status_detail(status_detail) assert actual_status_detail == expected_status_detail + def test_convert_status_detail_edge_case(): status_detail = { 'url': { 'summary': 'At least one compatible url field. Failed', 'fields': ['testurl'], }, 'metadata': [ { 'summary': 'Mandatory fields missing', 'fields': ['9', 10, 1.212], }, ], 'archive': [ { 'summary': 'Invalid archive', 'fields': ['3'], }, { 'summary': 'Unsupported archive', 'fields': [2], } ], } expected_status_detail = '''- Mandatory fields missing (9, 10, 1.212) - Invalid archive (3) - Unsupported archive (2) - At least one compatible url field. Failed (testurl) ''' actual_status_detail = convert_status_detail(status_detail) assert actual_status_detail == expected_status_detail diff --git a/swh/deposit/tests/api/test_service_document.py b/swh/deposit/tests/api/test_service_document.py index d13f8044..558c7598 100644 --- a/swh/deposit/tests/api/test_service_document.py +++ b/swh/deposit/tests/api/test_service_document.py @@ -1,89 +1,87 @@ # Copyright (C) 2017-2019 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from django.urls import reverse from rest_framework import status -from rest_framework.test import APITestCase from swh.deposit.tests import TEST_CONFIG from swh.deposit.config import SD_IRI -from ..common import BasicTestCase, WithAuthTestCase def test_service_document_no_auth_fails(client): """Without authentication, service document endpoint should return 401 """ url = reverse(SD_IRI) response = client.get(url) assert response.status_code == status.HTTP_401_UNAUTHORIZED def test_service_document_no_auth_with_http_auth_should_not_break(client): """Without auth, sd endpoint through browser should return 401 """ url = reverse(SD_IRI) response = client.get( url, HTTP_ACCEPT='text/html,application/xml;q=9,*/*,q=8') assert response.status_code == status.HTTP_401_UNAUTHORIZED def test_service_document(authenticated_client, deposit_user): """With authentication, service document list user's collection """ url = reverse(SD_IRI) response = authenticated_client.get(url) check_response(response, deposit_user.username) def test_service_document_with_http_accept_header( authenticated_client, deposit_user): """With authentication, with browser, sd list user's collection """ url = reverse(SD_IRI) response = authenticated_client.get( url, HTTP_ACCEPT='text/html,application/xml;q=9,*/*,q=8') check_response(response, deposit_user.username) def check_response(response, username): assert response.status_code == status.HTTP_200_OK assert response.content.decode('utf-8') == \ ''' 2.0 %s The Software Heritage (SWH) Archive %s Software Collection application/zip application/x-tar Collection Policy Software Heritage Archive Collect, Preserve, Share false false http://purl.org/net/sword/package/SimpleZip http://testserver/1/%s/ %s ''' % (TEST_CONFIG['max_upload_size'], username, username, username, username) # noqa diff --git a/swh/deposit/tests/conftest.py b/swh/deposit/tests/conftest.py index 20f6fe22..49ba4c4f 100644 --- a/swh/deposit/tests/conftest.py +++ b/swh/deposit/tests/conftest.py @@ -1,126 +1,127 @@ # Copyright (C) 2019 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import base64 import pytest import psycopg2 -from django.db import connections from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from rest_framework.test import APIClient from swh.scheduler.tests.conftest import * # noqa TEST_USER = { 'username': 'test', 'password': 'password', 'email': 'test@example.org', 'provider_url': 'https://hal-test.archives-ouvertes.fr/', 'domain': 'archives-ouvertes.fr/', 'collection': { 'name': 'test' }, } def execute_sql(sql): """Execute sql to postgres db""" with psycopg2.connect(database='postgres') as conn: conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = conn.cursor() cur.execute(sql) @pytest.hookimpl(tryfirst=True) def pytest_load_initial_conftests(early_config, parser, args): """This hook is done prior to django loading. Used to initialize the deposit's server db. """ import project.app.signals def prepare_db(*args, **kwargs): from django.conf import settings db_name = 'tests' print('before: %s' % settings.DATABASES) # work around db settings for django - for k, v in [('ENGINE', 'django.db.backends.postgresql'), - ('NAME', 'tests'), - ('USER', postgresql_proc.user), - ('HOST', postgresql_proc.host), - ('PORT', postgresql_proc.port), + for k, v in [ + ('ENGINE', 'django.db.backends.postgresql'), + ('NAME', 'tests'), + ('USER', postgresql_proc.user), # noqa + ('HOST', postgresql_proc.host), # noqa + ('PORT', postgresql_proc.port), # noqa ]: settings.DATABASES['default'][k] = v + print('after: %s' % settings.DATABASES) execute_sql('DROP DATABASE IF EXISTS %s' % db_name) execute_sql('CREATE DATABASE %s TEMPLATE template0' % db_name) project.app.signals.something = prepare_db @pytest.fixture def deposit_user(db): """Create/Return the test_user "test" """ from swh.deposit.models import DepositCollection, DepositClient # UserModel = django_user_model collection_name = TEST_USER['collection']['name'] try: collection = DepositCollection._default_manager.get( name=collection_name) except DepositCollection.DoesNotExist: collection = DepositCollection(name=collection_name) collection.save() # Create a user try: user = DepositClient._default_manager.get( username=TEST_USER['username']) except DepositClient.DoesNotExist: user = DepositClient._default_manager.create_user( username=TEST_USER['username'], email=TEST_USER['email'], password=TEST_USER['password'], provider_url=TEST_USER['provider_url'], domain=TEST_USER['domain'], ) user.collections = [collection.id] user.save() return user # @pytest.fixture # def headers(deposit_user): import base64 _token = '%s:%s' % (deposit_user.username, TEST_USER['password']) token = base64.b64encode(_token.encode('utf-8')) authorization = 'Basic %s' % token.decode('utf-8') return { 'AUTHENTICATION': authorization, } @pytest.fixture def client(): """Override pytest-django one which does not work for djangorestframework. """ return APIClient() # <- drf's client @pytest.yield_fixture def authenticated_client(client, deposit_user): """Returned a logged client """ _token = '%s:%s' % (deposit_user.username, TEST_USER['password']) token = base64.b64encode(_token.encode('utf-8')) authorization = 'Basic %s' % token.decode('utf-8') client.credentials(HTTP_AUTHORIZATION=authorization) yield client client.logout() diff --git a/swh/deposit/tests/test_utils.py b/swh/deposit/tests/test_utils.py index a0c41350..f44710a6 100644 --- a/swh/deposit/tests/test_utils.py +++ b/swh/deposit/tests/test_utils.py @@ -1,216 +1,215 @@ # Copyright (C) 2018-2019 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information -import unittest import pytest from unittest.mock import patch from swh.deposit import utils from swh.deposit.models import Deposit, DepositClient def test_origin_url_from(): """With correctly setup-ed deposit, all is fine """ for provider_url, external_id in ( ('http://somewhere.org', 'uuid'), ('http://overthejungle.org', 'diuu'), ): deposit = Deposit( client=DepositClient(provider_url=provider_url), external_id=external_id ) actual_origin_url = utils.origin_url_from(deposit) assert actual_origin_url == '%s/%s' % ( provider_url.rstrip('/'), external_id) def test_origin_url_from_ko(): """Badly configured deposit should raise """ for provider_url, external_id in ( (None, 'uuid'), ('http://overthejungle.org', None), ): deposit = Deposit( client=DepositClient(provider_url=provider_url), external_id=None ) with pytest.raises(AssertionError): utils.origin_url_from(deposit) def test_merge(): """Calling utils.merge on dicts should merge without losing information """ d0 = { 'author': 'someone', 'license': [['gpl2']], 'a': 1 } d1 = { 'author': ['author0', {'name': 'author1'}], 'license': [['gpl3']], 'b': { '1': '2' } } d2 = { 'author': map(lambda x: x, ['else']), 'license': 'mit', 'b': { '2': '3', } } d3 = { 'author': (v for v in ['no one']), } actual_merge = utils.merge(d0, d1, d2, d3) expected_merge = { 'a': 1, 'license': [['gpl2'], ['gpl3'], 'mit'], 'author': [ 'someone', 'author0', {'name': 'author1'}, 'else', 'no one'], 'b': { '1': '2', '2': '3', } } assert actual_merge == expected_merge def test_merge_2(): d0 = { 'license': 'gpl2', 'runtime': { 'os': 'unix derivative' } } d1 = { 'license': 'gpl3', 'runtime': 'GNU/Linux' } expected = { 'license': ['gpl2', 'gpl3'], 'runtime': [ { 'os': 'unix derivative' }, 'GNU/Linux' ], } actual = utils.merge(d0, d1) assert actual == expected def test_merge_edge_cases(): input_dict = { 'license': ['gpl2', 'gpl3'], 'runtime': [ { 'os': 'unix derivative' }, 'GNU/Linux' ], } # against empty dict actual = utils.merge(input_dict, {}) assert actual == input_dict # against oneself actual = utils.merge(input_dict, input_dict, input_dict) assert actual == input_dict def test_merge_one_dict(): """Merge one dict should result in the same dict value """ input_and_expected = {'anything': 'really'} actual = utils.merge(input_and_expected) assert actual == input_and_expected def test_merge_raise(): """Calling utils.merge with any no dict argument should raise """ d0 = { 'author': 'someone', 'a': 1 } d1 = ['not a dict'] with pytest.raises(ValueError): utils.merge(d0, d1) with pytest.raises(ValueError): utils.merge(d1, d0) with pytest.raises(ValueError): utils.merge(d1) assert utils.merge(d0) == d0 @patch('swh.deposit.utils.normalize_timestamp', side_effect=lambda x: x) def test_normalize_date_0(mock_normalize): """When date is a list, choose the first date and normalize it Note: We do not test swh.model.identifiers which is already tested in swh.model """ actual_date = utils.normalize_date(['2017-10-12', 'date1']) expected_date = '2017-10-12 00:00:00+00:00' assert str(actual_date) == expected_date @patch('swh.deposit.utils.normalize_timestamp', side_effect=lambda x: x) def test_normalize_date_1(mock_normalize): """Providing a date in a reasonable format, everything is fine Note: We do not test swh.model.identifiers which is already tested in swh.model """ actual_date = utils.normalize_date('2018-06-11 17:02:02') expected_date = '2018-06-11 17:02:02+00:00' assert str(actual_date) == expected_date @patch('swh.deposit.utils.normalize_timestamp', side_effect=lambda x: x) def test_normalize_date_doing_irrelevant_stuff(mock_normalize): """Providing a date with only the year results in a reasonable date Note: We do not test swh.model.identifiers which is already tested in swh.model """ actual_date = utils.normalize_date('2017') expected_date = '2017-01-01 00:00:00+00:00' assert str(actual_date) == expected_date