27458bf13a
This commit applies a subset of the fixes that were applied via `hatch check code --unsafe-fixes --fix`.
234 lines
7.8 KiB
Python
234 lines
7.8 KiB
Python
import hashlib
|
|
from pathlib import PurePosixPath
|
|
|
|
import pytest
|
|
|
|
from dump_things_service import (
|
|
HTTP_200_OK,
|
|
HTTP_201_CREATED,
|
|
HTTP_406_NOT_ACCEPTABLE,
|
|
HTTP_409_CONFLICT,
|
|
)
|
|
from dump_things_service.abstract_config import (
|
|
TokenCollectionConfig,
|
|
TokenModes,
|
|
dump_things_config_iri,
|
|
dump_things_private_collection_name,
|
|
get_config_backends,
|
|
read_config,
|
|
)
|
|
from dump_things_service.collection_endpoints import CollectionRequest
|
|
from dump_things_service.exceptions import ConfigError
|
|
from dump_things_service.tests import schema_file
|
|
from dump_things_service.token_endpoints import TokenRequest
|
|
|
|
collection_request_pattern = CollectionRequest(
|
|
name='',
|
|
schema=str(schema_file),
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath('curate_dir'),
|
|
incoming=PurePosixPath('incoming_dir'),
|
|
)
|
|
|
|
|
|
def test_illegal_collection_name_detection(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
for name in (
|
|
'collections',
|
|
'tokens',
|
|
'admin_tokens',
|
|
dump_things_private_collection_name,
|
|
):
|
|
response = test_client.post(
|
|
'/collections',
|
|
json={
|
|
**collection_request_pattern.model_dump(mode='json', by_alias=True),
|
|
'name': name,
|
|
},
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_409_CONFLICT
|
|
|
|
|
|
@pytest.mark.skip(
|
|
reason='Reuse detection is disabled to support existing old configurations'
|
|
)
|
|
def test_collection_dir_reuse_detection(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
for curated_path, incoming_path in (
|
|
('curated/collection_1', 'incoming/XXXX'),
|
|
('curated/XXXX', 'incoming/collection_1'),
|
|
('curated/collection_1', 'incoming/collection_2'),
|
|
):
|
|
response = test_client.post(
|
|
'/collections',
|
|
json={
|
|
**collection_request_pattern.model_dump(mode='json', by_alias=True),
|
|
'curated': curated_path,
|
|
'incoming': incoming_path,
|
|
},
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_409_CONFLICT
|
|
|
|
|
|
def test_scanner_error_detection(tmp_path_factory):
|
|
tmp_path = tmp_path_factory.mktemp('config_scanner_test')
|
|
|
|
config_backend, _audit_backend = get_config_backends(tmp_path)
|
|
config_backend.add_record(
|
|
iri=dump_things_config_iri,
|
|
class_name='DumpThingsConfig',
|
|
json_object={'pid': dump_things_config_iri},
|
|
)
|
|
|
|
md5_hexdigest = hashlib.md5(dump_things_config_iri.encode()).hexdigest()
|
|
config_file_path = (
|
|
config_backend.root / 'DumpThingsConfig' / f'{md5_hexdigest}.yaml'
|
|
)
|
|
config_file_path.write_text('collections: ::: -\n sdsdfsdf: xxx')
|
|
with pytest.raises(ConfigError):
|
|
read_config(tmp_path, force_reload=True)
|
|
|
|
|
|
def test_structure_error_detection(tmp_path_factory):
|
|
tmp_path = tmp_path_factory.mktemp('config_scanner_test')
|
|
|
|
config_backend, _audit_backend = get_config_backends(tmp_path)
|
|
config_backend.add_record(
|
|
iri=dump_things_config_iri,
|
|
class_name='DumpThingsConfig',
|
|
json_object={'pid': dump_things_config_iri},
|
|
)
|
|
|
|
md5_hexdigest = hashlib.md5(dump_things_config_iri.encode()).hexdigest()
|
|
config_file_path = (
|
|
config_backend.root / 'DumpThingsConfig' / f'{md5_hexdigest}.yaml'
|
|
)
|
|
config_file_path.write_text('type: 1\n')
|
|
with pytest.raises(ConfigError):
|
|
read_config(tmp_path, force_reload=True)
|
|
|
|
|
|
def test_missing_incoming_detection(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
# Add a collection without incoming
|
|
collection_request = CollectionRequest(
|
|
name='missing_incoming_detection_test',
|
|
default_token='Test XXXXX (CURATOR)',
|
|
curated=PurePosixPath('missing_incoming_detection'),
|
|
schema=str(schema_file),
|
|
)
|
|
|
|
response = test_client.post(
|
|
'/collections',
|
|
json=collection_request.model_dump(mode='json', by_alias=True),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
|
|
# Add a write token that references the collection, expect this to
|
|
# fail because the collection does not contain an incoming path
|
|
token_request = TokenRequest(
|
|
name='missing-incoming-token',
|
|
user_id='missing_incoming_user',
|
|
collections={
|
|
'missing_incoming_detection_test': TokenCollectionConfig(
|
|
mode=TokenModes.CURATOR,
|
|
incoming_label='',
|
|
)
|
|
},
|
|
)
|
|
|
|
# Check that a write token for a collection without incoming path cannot
|
|
# be created.
|
|
response = test_client.post(
|
|
'/tokens',
|
|
json=token_request.model_dump(mode='json', by_alias=True),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_406_NOT_ACCEPTABLE
|
|
|
|
# Remove the collection without incoming path
|
|
response = test_client.delete(
|
|
'/collections/missing_incoming_detection_test',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
# Add a collection with incoming path
|
|
collection_request.incoming = PurePosixPath(
|
|
'missing_incoming_detection_test_incoming'
|
|
)
|
|
response = test_client.post(
|
|
'/collections',
|
|
json=collection_request.model_dump(mode='json', by_alias=True),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
|
|
# Check that a write token for a collection with an incoming path but a
|
|
# missing label cannot be created.
|
|
response = test_client.post(
|
|
'/tokens',
|
|
json=token_request.model_dump(mode='json', by_alias=True),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_406_NOT_ACCEPTABLE
|
|
|
|
# Check that a write token for a collection with an incoming path can be created
|
|
token_request.collections['missing_incoming_detection_test'] = (
|
|
TokenCollectionConfig(
|
|
mode=TokenModes.CURATOR,
|
|
incoming_label='test_incoming_label',
|
|
)
|
|
)
|
|
response = test_client.post(
|
|
'/tokens',
|
|
json=token_request.model_dump(mode='json', by_alias=True),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
|
|
# Remove the collection with the incoming path
|
|
response = test_client.delete(
|
|
'/collections/missing_incoming_detection_test',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
# Check that a creation attempt for the collection without incoming path fails
|
|
collection_request.incoming = None
|
|
response = test_client.post(
|
|
'/collections',
|
|
json=collection_request.model_dump(mode='json', by_alias=True),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_406_NOT_ACCEPTABLE
|
|
|
|
|
|
def test_submission_tags_handling(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
response = test_client.get(
|
|
'/collections/collection_8',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
json_obj = response.json()
|
|
assert json_obj['submission_tags'] == {
|
|
'submitter_id_tag': 'no_default_id_tag',
|
|
'submission_time_tag': 'no_default_time_tag',
|
|
}
|
|
|
|
response = test_client.get(
|
|
'/collections/collection_1',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
json_obj = response.json()
|
|
assert json_obj['submission_tags'] == {
|
|
'submitter_id_tag': 'http://purl.obolibrary.org/obo/NCIT_C54269',
|
|
'submission_time_tag': 'http://semanticscience.org/resource/SIO_001083',
|
|
}
|