Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_datacite_register_fail(mocker, app, db, es, minimal_record):
# Make the datacite API unavailable
dc_mock = mocker.patch(
'invenio_pidstore.providers.datacite.DataCiteMDSClient')
dc_mock().metadata_post.side_effect = datacite.errors.HttpError()
# Create a reserved recid
record = Record.create(minimal_record)
record_uuid = record.id
recid = record['recid']
recid_pid = PersistentIdentifier.create(
'recid', recid, status=PIDStatus.RESERVED)
# Mint the record
zenodo_record_minter(record_uuid, record)
record.commit()
db.session.commit()
with pytest.raises(datacite.errors.HttpError):
datacite_register.apply((recid_pid.pid_value, str(record_uuid)))
'invenio_pidstore.providers.datacite.DataCiteMDSClient')
dc_mock().metadata_post.side_effect = datacite.errors.HttpError()
# Create a reserved recid
record = Record.create(minimal_record)
record_uuid = record.id
recid = record['recid']
recid_pid = PersistentIdentifier.create(
'recid', recid, status=PIDStatus.RESERVED)
# Mint the record
zenodo_record_minter(record_uuid, record)
record.commit()
db.session.commit()
with pytest.raises(datacite.errors.HttpError):
datacite_register.apply((recid_pid.pid_value, str(record_uuid)))
# Check that the task was retried ("max_retries" + 1) times
dc_calls = len(dc_mock().metadata_post.mock_calls)
assert dc_calls == datacite_register.max_retries + 1
Note that retries are done on another thread so only the initial call to
retry can be tested.
"""
patched_client = mocker.patch(
'cd2h_repo_project.modules.doi.tasks.DataCiteMDSClient')()
patched_retry = mocker.patch(
'cd2h_repo_project.modules.doi.tasks.register_doi.retry')
# Because publish() triggers the task, we need to perform some of the steps
# of publish() without calling publish()
record = create_record(published=False)
mint_pids_for_record(record.id, record)
doi_pid = PersistentIdentifier.get(pid_type='doi', pid_value=record['id'])
# Test HttpError
patched_client.metadata_post.side_effect = datacite.errors.HttpError()
register_doi(record['id'])
number_retries = len(patched_retry.mock_calls)
assert number_retries == 1
# Test DataCiteError
patched_client.metadata_post.side_effect = datacite.errors.DataCiteError()
register_doi(record['id'])
number_retries = len(patched_retry.mock_calls)
assert number_retries == 2
# Because publish() triggers the task, we need to perform some of the steps
# of publish() without calling publish()
record = create_record(published=False)
mint_pids_for_record(record.id, record)
doi_pid = PersistentIdentifier.get(pid_type='doi', pid_value=record['id'])
# Test HttpError
patched_client.metadata_post.side_effect = datacite.errors.HttpError()
register_doi(record['id'])
number_retries = len(patched_retry.mock_calls)
assert number_retries == 1
# Test DataCiteError
patched_client.metadata_post.side_effect = datacite.errors.DataCiteError()
register_doi(record['id'])
number_retries = len(patched_retry.mock_calls)
assert number_retries == 2