Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/loader/test_loader.py
# Copyright (C) 2017-2018 The Software Heritage developers | # Copyright (C) 2017-2019 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 os | import os | ||||
import unittest | import unittest | ||||
import shutil | import shutil | ||||
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines | class DepositLoaderScenarioTest(APITestCase, WithAuthTestCase, | ||||
BaseLoaderStorageTest): | BaseLoaderStorageTest): | ||||
def setUp(self): | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
# create the extraction dir used by the loader | # create the extraction dir used by the loader | ||||
os.makedirs(TEST_LOADER_CONFIG['extraction_dir'], exist_ok=True) | os.makedirs(TEST_LOADER_CONFIG['extraction_dir'], exist_ok=True) | ||||
# 1. create a deposit with archive and metadata | # Sets a basic client which accesses the test data | ||||
self.deposit_id = self.create_simple_binary_deposit() | |||||
# 2. Sets a basic client which accesses the test data | |||||
loader_client = SWHDepositTestClient(self.client, | loader_client = SWHDepositTestClient(self.client, | ||||
config=CLIENT_TEST_CONFIG) | config=CLIENT_TEST_CONFIG) | ||||
# 3. setup loader with that client | # Setup loader with that client | ||||
self.loader = loader.DepositLoader(client=loader_client) | self.loader = loader.DepositLoader(client=loader_client) | ||||
self.storage = self.loader.storage | self.storage = self.loader.storage | ||||
def tearDown(self): | def tearDown(self): | ||||
super().tearDown() | super().tearDown() | ||||
shutil.rmtree(TEST_LOADER_CONFIG['extraction_dir']) | shutil.rmtree(TEST_LOADER_CONFIG['extraction_dir']) | ||||
def test_inject_deposit_ready(self): | def test_inject_deposit_ready(self): | ||||
"""Load a deposit which is ready | """Load a deposit which is ready | ||||
""" | """ | ||||
args = [self.collection.name, self.deposit_id] | # create a deposit with archive and metadata | ||||
deposit_id = self.create_simple_binary_deposit() | |||||
ardumont: This was actually wrong to trigger a load of a deposit with a wrong status (partial here which… | |||||
Done Inline ActionsThat's why i changed the test ;) ardumont: That's why i changed the test ;) | |||||
self.update_binary_deposit(deposit_id, status_partial=False) | |||||
args = [self.collection.name, deposit_id] | |||||
archive_url = reverse(PRIVATE_GET_RAW_CONTENT, args=args) | archive_url = reverse(PRIVATE_GET_RAW_CONTENT, args=args) | ||||
deposit_meta_url = reverse(PRIVATE_GET_DEPOSIT_METADATA, args=args) | deposit_meta_url = reverse(PRIVATE_GET_DEPOSIT_METADATA, args=args) | ||||
deposit_update_url = reverse(PRIVATE_PUT_DEPOSIT, args=args) | deposit_update_url = reverse(PRIVATE_PUT_DEPOSIT, args=args) | ||||
# when | # when | ||||
res = self.loader.load(archive_url=archive_url, | res = self.loader.load(archive_url=archive_url, | ||||
deposit_meta_url=deposit_meta_url, | deposit_meta_url=deposit_meta_url, | ||||
deposit_update_url=deposit_update_url) | deposit_update_url=deposit_update_url) | ||||
# then | # then | ||||
self.assertEqual(res['status'], 'eventful', res) | self.assertEqual(res['status'], 'eventful', res) | ||||
self.assertCountContents(1) | self.assertCountContents(1) | ||||
self.assertCountDirectories(1) | self.assertCountDirectories(1) | ||||
self.assertCountRevisions(1) | self.assertCountRevisions(1) | ||||
self.assertCountReleases(0) | self.assertCountReleases(0) | ||||
self.assertCountSnapshots(1) | self.assertCountSnapshots(1) | ||||
def test_inject_deposit_verify_metadata(self): | def test_inject_deposit_verify_metadata(self): | ||||
"""Load a deposit with metadata, test metadata integrity | """Load a deposit with metadata, test metadata integrity | ||||
""" | """ | ||||
self.deposit_metadata_id = self.add_metadata_to_deposit( | deposit_id = self.create_simple_binary_deposit() | ||||
self.deposit_id) | self.add_metadata_to_deposit(deposit_id, status_partial=False) | ||||
args = [self.collection.name, self.deposit_metadata_id] | args = [self.collection.name, deposit_id] | ||||
archive_url = reverse(PRIVATE_GET_RAW_CONTENT, args=args) | archive_url = reverse(PRIVATE_GET_RAW_CONTENT, args=args) | ||||
deposit_meta_url = reverse(PRIVATE_GET_DEPOSIT_METADATA, args=args) | deposit_meta_url = reverse(PRIVATE_GET_DEPOSIT_METADATA, args=args) | ||||
deposit_update_url = reverse(PRIVATE_PUT_DEPOSIT, args=args) | deposit_update_url = reverse(PRIVATE_PUT_DEPOSIT, args=args) | ||||
# when | # when | ||||
self.loader.load(archive_url=archive_url, | self.loader.load(archive_url=archive_url, | ||||
deposit_meta_url=deposit_meta_url, | deposit_meta_url=deposit_meta_url, | ||||
Show All 38 Lines | def test_inject_deposit_verify_metadata(self): | ||||
'id': 'hal-01243065', | 'id': 'hal-01243065', | ||||
'client': 'hal', | 'client': 'hal', | ||||
codemeta + 'keywords': 'DSP programming,Web', | codemeta + 'keywords': 'DSP programming,Web', | ||||
codemeta + 'developmentStatus': 'stable' | codemeta + 'developmentStatus': 'stable' | ||||
} | } | ||||
self.assertOriginMetadataContains('deposit', origin_url, | self.assertOriginMetadataContains('deposit', origin_url, | ||||
expected_origin_metadata) | expected_origin_metadata) | ||||
deposit = Deposit.objects.get(pk=self.deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
self.assertRegex(deposit.swh_id, r'^swh:1:dir:.*') | self.assertRegex(deposit.swh_id, r'^swh:1:dir:.*') | ||||
self.assertEqual(deposit.swh_id_context, '%s;origin=%s' % ( | self.assertEqual(deposit.swh_id_context, '%s;origin=%s' % ( | ||||
deposit.swh_id, origin_url | deposit.swh_id, origin_url | ||||
)) | )) | ||||
self.assertRegex(deposit.swh_anchor_id, r'^swh:1:rev:.*') | self.assertRegex(deposit.swh_anchor_id, r'^swh:1:rev:.*') | ||||
self.assertEqual(deposit.swh_anchor_id_context, '%s;origin=%s' % ( | self.assertEqual(deposit.swh_anchor_id_context, '%s;origin=%s' % ( | ||||
deposit.swh_anchor_id, origin_url | deposit.swh_anchor_id, origin_url | ||||
)) | )) |
This was actually wrong to trigger a load of a deposit with a wrong status (partial here which is only possible in tests).