diff --git a/swh/deposit/tests/api/test_service_document.py b/swh/deposit/tests/api/test_service_document.py index 61d8e074..d13f8044 100644 --- a/swh/deposit/tests/api/test_service_document.py +++ b/swh/deposit/tests/api/test_service_document.py @@ -1,102 +1,89 @@ # 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 -class ServiceDocumentNoAuthCase(APITestCase, BasicTestCase): - """Service document endpoints are protected with basic authentication. +def test_service_document_no_auth_fails(client): + """Without authentication, service document endpoint should return 401 """ - def test_service_document_no_authentication_fails(self): - """Without authentication, service document endpoint should return 401 + url = reverse(SD_IRI) + response = client.get(url) + assert response.status_code == status.HTTP_401_UNAUTHORIZED - """ - url = reverse(SD_IRI) - response = self.client.get(url) +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 - self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) - def test_service_document_with_http_accept_should_not_break(self): - """Without auth, sd endpoint through browser should return 401 +def test_service_document(authenticated_client, deposit_user): + """With authentication, service document list user's collection - """ - url = reverse(SD_IRI) + """ + url = reverse(SD_IRI) + response = authenticated_client.get(url) + check_response(response, deposit_user.username) - # when - response = self.client.get( - url, - HTTP_ACCEPT='text/html,application/xml;q=9,*/*,q=8') - self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) +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) -class ServiceDocumentCase(APITestCase, WithAuthTestCase, BasicTestCase): - def assertResponseOk(self, response): # noqa: N802 - self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.content.decode('utf-8'), + +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'], - self.username, - self.username, - self.username, - self.username)) # noqa - - def test_service_document(self): - """With authentication, service document list user's collection - - """ - url = reverse(SD_IRI) - - # when - response = self.client.get(url) - - # then - self.assertResponseOk(response) - - def test_service_document_with_http_accept_header(self): - """With authentication, with browser, sd list user's collection - - """ - url = reverse(SD_IRI) - - # when - response = self.client.get( - url, - HTTP_ACCEPT='text/html,application/xml;q=9,*/*,q=8') - - self.assertResponseOk(response) + username, + username, + username, + username) # noqa diff --git a/swh/deposit/tests/conftest.py b/swh/deposit/tests/conftest.py index 1d65a983..20f6fe22 100644 --- a/swh/deposit/tests/conftest.py +++ b/swh/deposit/tests/conftest.py @@ -1,54 +1,126 @@ # 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), ]: 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()