Differential D7233 Diff 26226 swh/deposit/tests/conftest.py

Changeset View

Standalone View

swh/deposit/tests/conftest.py

# Copyright (C) 2019-2021 The Software Heritage developers		# Copyright (C) 2019-2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import base64		import base64
from copy import deepcopy		from copy import deepcopy
from functools import partial		from functools import partial
import os		import os
import re		import re
from typing import TYPE_CHECKING, Dict, Mapping		from typing import TYPE_CHECKING, Dict, Mapping
		from xml.etree import ElementTree

from django.test.utils import setup_databases # type: ignore		from django.test.utils import setup_databases # type: ignore
from django.urls import reverse_lazy as reverse		from django.urls import reverse_lazy as reverse
import psycopg2		import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT		from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
import pytest		import pytest
from rest_framework import status		from rest_framework import status
from rest_framework.test import APIClient		from rest_framework.test import APIClient
Show All 9 Lines	from swh.deposit.config import (
DEPOSIT_STATUS_LOAD_FAILURE,		DEPOSIT_STATUS_LOAD_FAILURE,
DEPOSIT_STATUS_LOAD_SUCCESS,		DEPOSIT_STATUS_LOAD_SUCCESS,
DEPOSIT_STATUS_PARTIAL,		DEPOSIT_STATUS_PARTIAL,
DEPOSIT_STATUS_REJECTED,		DEPOSIT_STATUS_REJECTED,
DEPOSIT_STATUS_VERIFIED,		DEPOSIT_STATUS_VERIFIED,
SE_IRI,		SE_IRI,
setup_django_for,		setup_django_for,
)		)
from swh.deposit.parsers import parse_xml
from swh.deposit.tests.common import (		from swh.deposit.tests.common import (
create_arborescence_archive,		create_arborescence_archive,
post_archive,		post_archive,
post_atom,		post_atom,
)		)
		from swh.deposit.utils import NAMESPACES
from swh.model.hashutil import hash_to_bytes		from swh.model.hashutil import hash_to_bytes
from swh.model.swhids import CoreSWHID, ObjectType, QualifiedSWHID		from swh.model.swhids import CoreSWHID, ObjectType, QualifiedSWHID
from swh.scheduler import get_scheduler		from swh.scheduler import get_scheduler

if TYPE_CHECKING:		if TYPE_CHECKING:
from swh.deposit.models import Deposit, DepositClient, DepositCollection		from swh.deposit.models import Deposit, DepositClient, DepositCollection


▲ Show 20 Lines • Show All 389 Lines • ▼ Show 20 Lines	response = post_archive(
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_IN_PROGRESS=str(in_progress).lower(),		HTTP_IN_PROGRESS=str(in_progress).lower(),
)		)

# then		# then
assert response.status_code == status.HTTP_201_CREATED, response.content.decode()		assert response.status_code == status.HTTP_201_CREATED, response.content.decode()
from swh.deposit.models import Deposit		from swh.deposit.models import Deposit

response_content = parse_xml(response.content)		response_content = ElementTree.fromstring(response.content)
deposit_id = response_content["swh:deposit_id"]		deposit_id = int(
		response_content.findtext("swh:deposit_id", "", namespaces=NAMESPACES)
		)
deposit = Deposit._default_manager.get(id=deposit_id)		deposit = Deposit._default_manager.get(id=deposit_id)

if deposit.status != deposit_status:		if deposit.status != deposit_status:
deposit.status = deposit_status		deposit.status = deposit_status
deposit.save()		deposit.save()
assert deposit.status == deposit_status		assert deposit.status == deposit_status
return deposit		return deposit

▲ Show 20 Lines • Show All 99 Lines • ▼ Show 20 Lines	response = post_atom(
reverse(COL_IRI, args=[deposit_collection.name]),		reverse(COL_IRI, args=[deposit_collection.name]),
data=atom_dataset["entry-data1"],		data=atom_dataset["entry-data1"],
HTTP_SLUG="external-id-partial",		HTTP_SLUG="external-id-partial",
HTTP_IN_PROGRESS=True,		HTTP_IN_PROGRESS=True,
)		)

assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED

response_content = parse_xml(response.content)		response_content = ElementTree.fromstring(response.content)
deposit_id = response_content["swh:deposit_id"]		deposit_id = int(response_content.findtext("swh:deposit_id", namespaces=NAMESPACES))
from swh.deposit.models import Deposit		from swh.deposit.models import Deposit

deposit = Deposit._default_manager.get(pk=deposit_id)		deposit = Deposit._default_manager.get(pk=deposit_id)
assert deposit.status == DEPOSIT_STATUS_PARTIAL		assert deposit.status == DEPOSIT_STATUS_PARTIAL
return deposit		return deposit


@pytest.fixture		@pytest.fixture
Show All 33 Lines