09be6912a0
The test-fixture `fast_api_simple` now returns a tuple containing: - test_client instance - store path - admin token
332 lines
11 KiB
Python
332 lines
11 KiB
Python
import hashlib
|
|
from pathlib import PurePosixPath
|
|
|
|
import pytest
|
|
import yaml
|
|
from json_flattener import GlobalConfig
|
|
from starlette.status import HTTP_406_NOT_ACCEPTABLE
|
|
|
|
from dump_things_service import (
|
|
HTTP_200_OK,
|
|
HTTP_201_CREATED,
|
|
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(f'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(
|
|
f'/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
|
|
|
|
|
|
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(
|
|
f'/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 xxx_test_submission_tags_handling(dump_stores_simple):
|
|
config_object = GlobalConfig(
|
|
**yaml.load(
|
|
"""
|
|
type: collections
|
|
version: 1
|
|
collections:
|
|
collection_1:
|
|
default_token: basic_access
|
|
curated: curated/in_token_1
|
|
incoming: contributions
|
|
submission_tags:
|
|
submitter_id_tag: no_default_id_tag
|
|
submission_time_tag: no_default_time_tag
|
|
collection_2:
|
|
default_token: basic_access
|
|
curated: curated/collection_2
|
|
incoming: contributions
|
|
tokens:
|
|
basic_access:
|
|
user_id: anonymous
|
|
collections:
|
|
collection_1:
|
|
mode: WRITE_COLLECTION
|
|
incoming_label: incoming_anonymous
|
|
collection_2:
|
|
mode: WRITE_COLLECTION
|
|
incoming_label: incoming_anonymous
|
|
""",
|
|
Loader=yaml.SafeLoader,
|
|
)
|
|
)
|
|
|
|
global_dict = {}
|
|
config = process_config_object(dump_stores_simple, config_object, [], global_dict)
|
|
# Check for specified tags in collection `collection_1`
|
|
assert config.collections['collection_1'].submission_tags.submission_time_tag == 'no_default_time_tag'
|
|
assert config.collections['collection_1'].submission_tags.submitter_id_tag == 'no_default_id_tag'
|
|
# Check for default tags in collection `collection_2`
|
|
assert config.collections['collection_2'].submission_tags.submission_time_tag == 'http://semanticscience.org/resource/SIO_001083'
|
|
assert config.collections['collection_2'].submission_tags.submitter_id_tag == 'http://purl.obolibrary.org/obo/NCIT_C54269'
|
|
|
|
|
|
def xxx_test_submission_tags_resolving(dump_stores_simple):
|
|
config_object = GlobalConfig(
|
|
**yaml.load(
|
|
"""
|
|
type: collections
|
|
version: 1
|
|
collections:
|
|
collection_1:
|
|
default_token: basic_access
|
|
curated: curated/in_token_1
|
|
incoming: contributions
|
|
submission_tags:
|
|
submitter_id_tag: abc:id
|
|
submission_time_tag: abc:time
|
|
tokens:
|
|
basic_access:
|
|
user_id: anonymous
|
|
collections:
|
|
collection_1:
|
|
mode: WRITE_COLLECTION
|
|
incoming_label: incoming_anonymous
|
|
""",
|
|
Loader=yaml.SafeLoader,
|
|
)
|
|
)
|
|
|
|
global_dict = {}
|
|
process_config_object(dump_stores_simple, config_object, [], global_dict)
|
|
|
|
|
|
def xxx_test_submission_tags_resolving_error(dump_stores_simple):
|
|
config_object = GlobalConfig(
|
|
**yaml.load(
|
|
"""
|
|
type: collections
|
|
version: 1
|
|
collections:
|
|
collection_1:
|
|
default_token: basic_access
|
|
curated: curated/in_token_1
|
|
incoming: contributions
|
|
submission_tags:
|
|
submitter_id_tag: non-existing:id
|
|
collection_2:
|
|
default_token: basic_access
|
|
curated: curated/in_token_1
|
|
incoming: contributions
|
|
submission_tags:
|
|
submission_time_tag: non-existing:time
|
|
collection_3:
|
|
default_token: basic_access
|
|
curated: curated/in_token_1
|
|
incoming: contributions
|
|
submission_tags:
|
|
submitter_id_tag: http://something/non-existing
|
|
collection_4:
|
|
default_token: basic_access
|
|
curated: curated/in_token_1
|
|
incoming: contributions
|
|
submission_tags:
|
|
submission_time_tag: http://something/non-existing
|
|
tokens:
|
|
basic_access:
|
|
user_id: anonymous
|
|
collections:
|
|
collection_1:
|
|
mode: WRITE_COLLECTION
|
|
incoming_label: incoming_anonymous
|
|
collection_2:
|
|
mode: WRITE_COLLECTION
|
|
incoming_label: incoming_anonymous
|
|
collection_3:
|
|
mode: WRITE_COLLECTION
|
|
incoming_label: incoming_anonymous
|
|
collection_4:
|
|
mode: WRITE_COLLECTION
|
|
incoming_label: incoming_anonymous
|
|
""",
|
|
Loader=yaml.SafeLoader,
|
|
)
|
|
)
|
|
|
|
global_dict = {}
|
|
with pytest.raises(ConfigError) as e:
|
|
process_config_object(dump_stores_simple, config_object, [], global_dict)
|