Skip to content
Snippets Groups Projects
Verified Commit d491b648 authored by Antoine R. Dumont's avatar Antoine R. Dumont
Browse files

test.api.service_document: Migrate to pytest tests

parent 9b9e9c71
No related branches found
No related tags found
1 merge request!40Migrate most deposit tests to pytest
......@@ -12,38 +12,50 @@ 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') == \
'''<?xml version="1.0" ?>
<service xmlns:dcterms="http://purl.org/dc/terms/"
xmlns:sword="http://purl.org/net/sword/terms/"
......@@ -71,32 +83,7 @@ class ServiceDocumentCase(APITestCase, WithAuthTestCase, BasicTestCase):
</workspace>
</service>
''' % (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
......@@ -3,15 +3,29 @@
# 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:
......@@ -47,8 +61,66 @@ def pytest_load_initial_conftests(early_config, parser, args):
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()
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment