Michael Hanke
c332abbd1b
Some checks failed
Codespell / Check for spelling errors (push) Successful in 42s
Codespell / Check for spelling errors (pull_request) Successful in 43s
Ruff / Code linting (push) Failing after 52s
Ruff / Code linting (pull_request) Failing after 54s
Test execution / Test-all (push) Successful in 1m27s
Type annotation (PR) / check (pull_request) Failing after 1m30s
451 lines
14 KiB
Python
451 lines
14 KiB
Python
import sys
|
|
from pathlib import (
|
|
Path,
|
|
PurePosixPath,
|
|
)
|
|
from types import ModuleType
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from dump_things_service.abstract_config import (
|
|
GitAuditBackendConfig,
|
|
SQLiteBackendConfig,
|
|
TagSpec,
|
|
TokenCollectionConfig,
|
|
TokenModes,
|
|
hash_token_representation,
|
|
)
|
|
from dump_things_service.backends import StorageBackend
|
|
from dump_things_service.backends.record_dir import RecordDirStore
|
|
from dump_things_service.backends.sqlite import (
|
|
SQLiteBackend,
|
|
)
|
|
from dump_things_service.backends.sqlite import (
|
|
record_file_name as sqlite_db_filename,
|
|
)
|
|
from dump_things_service.collection_endpoints import CollectionRequest
|
|
from dump_things_service.instance_state import get_mapping_function_by_name
|
|
from dump_things_service.model import get_model_for_schema
|
|
from dump_things_service.resolve_curie import resolve_curie
|
|
from dump_things_service.tests.create_store import (
|
|
pid,
|
|
pid_curated,
|
|
pid_trr,
|
|
test_record,
|
|
test_record_curated,
|
|
test_record_trr,
|
|
)
|
|
from dump_things_service.token_endpoints import TokenRequest
|
|
|
|
# String representation of curated- and incoming-path
|
|
curated = 'curated'
|
|
incoming = 'incoming'
|
|
|
|
# Path to a local simple test schema
|
|
test_schema_location = str((Path(__file__).parent / 'testschema.yaml').absolute())
|
|
flat_social_schema_location = (
|
|
'https://concepts.datalad.org/s/flat-social/unreleased.yaml'
|
|
)
|
|
|
|
|
|
# The test store is created empty and collections are added via the admin
|
|
# web interface.
|
|
g_default_collections = [
|
|
CollectionRequest(
|
|
name=f'collection_{i}',
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/collection_{i}'),
|
|
schema=test_schema_location,
|
|
incoming=PurePosixPath(f'{incoming}/collection_{i}'),
|
|
)
|
|
for i in range(1, 8)
|
|
]
|
|
|
|
g_default_collections[6].submission_tags = TagSpec(
|
|
submitter_id_tag='abc:id',
|
|
submission_time_tag='abc:time',
|
|
)
|
|
|
|
g_default_collections.append(
|
|
CollectionRequest(
|
|
name='collection_8',
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/collection_8'),
|
|
schema=test_schema_location,
|
|
incoming=PurePosixPath(f'{incoming}/collection_8'),
|
|
backend=SQLiteBackendConfig(
|
|
type='sqlite',
|
|
),
|
|
submission_tags=TagSpec(
|
|
submitter_id_tag='no_default_id_tag',
|
|
submission_time_tag='no_default_time_tag',
|
|
),
|
|
)
|
|
)
|
|
|
|
g_default_collections.extend(
|
|
[
|
|
CollectionRequest(
|
|
name='collection_dlflatsocial-1',
|
|
schema=flat_social_schema_location,
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/collection_dlflatsocial-1'),
|
|
incoming=PurePosixPath(f'{incoming}/collection_dlflatsocial-1'),
|
|
),
|
|
CollectionRequest(
|
|
name='collection_dlflatsocial-2',
|
|
schema=flat_social_schema_location,
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/collection_dlflatsocial-2'),
|
|
incoming=PurePosixPath(f'{incoming}/collection_dlflatsocial-2'),
|
|
backend=SQLiteBackendConfig(
|
|
type='sqlite',
|
|
),
|
|
use_classes=[
|
|
'Organization',
|
|
'Person',
|
|
'Project',
|
|
],
|
|
ignore_classes=[
|
|
'Organization',
|
|
'Project',
|
|
],
|
|
),
|
|
]
|
|
)
|
|
|
|
g_default_tokens = [
|
|
TokenRequest(
|
|
name='test_default_token',
|
|
user_id='basic_access_user',
|
|
hashed=False,
|
|
representation='basic_access',
|
|
collections={
|
|
**{
|
|
f'collection_{i}': TokenCollectionConfig(
|
|
mode=TokenModes.READ_CURATED,
|
|
)
|
|
for i in range(1, 9)
|
|
},
|
|
**{
|
|
f'collection_dlflatsocial-{i}': TokenCollectionConfig(
|
|
mode=TokenModes.READ_CURATED,
|
|
)
|
|
for i in range(1, 3)
|
|
},
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test token for some collections',
|
|
user_id='test_user_1',
|
|
hashed=False,
|
|
representation='token-1',
|
|
collections={
|
|
collection_name: TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label='in_token_1',
|
|
)
|
|
for collection_name in (
|
|
'collection_1',
|
|
'collection_dlflatsocial-1',
|
|
'collection_dlflatsocial-2',
|
|
)
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test token for collection_2',
|
|
user_id='test_user_2',
|
|
hashed=False,
|
|
representation='token-2',
|
|
collections={
|
|
'collection_2': TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label='in_token-2',
|
|
)
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test token for collection_8',
|
|
user_id='test_user_8',
|
|
hashed=False,
|
|
representation='token-8',
|
|
collections={
|
|
'collection_8': TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label='test_user_8',
|
|
)
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test token for all collections',
|
|
user_id='user_all',
|
|
hashed=False,
|
|
representation='token-all',
|
|
collections={
|
|
**{
|
|
f'collection_{i}': TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label='token-all:user_all',
|
|
)
|
|
for i in range(1, 9)
|
|
},
|
|
**{
|
|
f'collection_dlflatsocial-{i}': TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label='token-all:user_all',
|
|
)
|
|
for i in range(1, 3)
|
|
},
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test Curator Token',
|
|
user_id='test_curator',
|
|
representation='token_curator',
|
|
collections={
|
|
f'collection_{i}': TokenCollectionConfig(
|
|
mode=TokenModes.CURATOR,
|
|
incoming_label=f'admin_{i}' if i < 5 else 'admin_common',
|
|
)
|
|
for i in range(1, 9)
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test Hashed Token',
|
|
user_id='test_hashed',
|
|
representation='token-hashed',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label='token-hashed-1',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test XX000 (READ_COLLECTION)',
|
|
user_id='test_user_1_read_collection',
|
|
representation='token_1_xxooo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.READ_COLLECTION,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test XXX00 (WRITE_COLLECTION)',
|
|
user_id='test_user_1_write_collection',
|
|
representation='token_1_xxxoo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test 0X000 (READ_SUBMISSIONS)',
|
|
user_id='test_user_1_read_submissions',
|
|
representation='token_1_oxooo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.READ_SUBMISSIONS,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test 0XX00 (WRITE_SUBMISSIONS)',
|
|
user_id='test_user_1_write_submissions',
|
|
representation='token_1_oxxoo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_SUBMISSIONS,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test X0X00 (SUBMIT)',
|
|
user_id='test_user_1_submit',
|
|
representation='token_1_xoxoo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.SUBMIT,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test 00X00 (SUBMIT_ONLY)',
|
|
user_id='test_user_1_submit_only',
|
|
representation='token_1_ooxoo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.SUBMIT_ONLY,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test X0000 (READ_CURATED)',
|
|
user_id='test_user_1_read_curated',
|
|
representation='token_1_xoooo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.READ_CURATED,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test 00000 (NOTHING)',
|
|
user_id='test_user_1_nothing',
|
|
representation='token_1_ooooo',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.NOTHING,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
TokenRequest(
|
|
name='Test XXXXX (CURATOR)',
|
|
user_id='test_user_1_curator',
|
|
representation='token_1_xxxxx',
|
|
collections={
|
|
'collection_1': TokenCollectionConfig(
|
|
mode=TokenModes.CURATOR,
|
|
incoming_label='modes',
|
|
),
|
|
'collection_8': TokenCollectionConfig(
|
|
mode=TokenModes.CURATOR,
|
|
incoming_label='modes',
|
|
),
|
|
},
|
|
),
|
|
]
|
|
|
|
g_default_entries = {
|
|
f'collection_{i}': [('Person', pid, test_record)] for i in range(1, 9)
|
|
}
|
|
for collection_id in range(1, 9):
|
|
g_default_entries[f'collection_{collection_id}'].extend(
|
|
[
|
|
('Person', pid_curated, test_record_curated),
|
|
(
|
|
'Person',
|
|
'abc:mode_test',
|
|
'pid: abc:mode_test\ngiven_name: mode_curated\nschema_type: abc:Person\n',
|
|
),
|
|
]
|
|
)
|
|
|
|
g_default_entries['collection_dlflatsocial-1'] = [('Person', pid_trr, test_record_trr)]
|
|
g_default_entries['collection_dlflatsocial-2'] = [('Person', pid_trr, test_record_trr)]
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def dump_stores_simple(tmp_path_factory):
|
|
tmp_path = tmp_path_factory.mktemp('dump_store')
|
|
audit_store_path = tmp_path_factory.mktemp('audit_store')
|
|
return tmp_path, audit_store_path
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def fastapi_app_simple(dump_stores_simple):
|
|
tmp_path, audit_tmp_path = dump_stores_simple
|
|
|
|
admin_token = 'admin-1'
|
|
old_sys_argv = sys.argv
|
|
sys.argv = [
|
|
'test-runner',
|
|
'--admin-token-hash',
|
|
hash_token_representation(admin_token),
|
|
'--ignore-default-config-file',
|
|
str(tmp_path),
|
|
]
|
|
from dump_things_service.main import app
|
|
|
|
sys.argv = old_sys_argv
|
|
return app, tmp_path, audit_tmp_path, admin_token
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def fastapi_client_simple(fastapi_app_simple):
|
|
from fastapi.testclient import TestClient
|
|
|
|
test_client = TestClient(fastapi_app_simple[0])
|
|
store_path = fastapi_app_simple[1]
|
|
audit_path = fastapi_app_simple[2]
|
|
admin_token = fastapi_app_simple[3]
|
|
|
|
# Add an audit backend to the first collection in g_default_collections
|
|
assert g_default_collections[0].name == 'collection_1'
|
|
g_default_collections[0].audit_backends = [
|
|
GitAuditBackendConfig(
|
|
type='gitaudit',
|
|
path=Path(audit_path),
|
|
auto_flush_timeout=2,
|
|
)
|
|
]
|
|
|
|
# Add collections via the Web-API
|
|
for collection_config in g_default_collections:
|
|
response = test_client.post(
|
|
'/collections',
|
|
json=collection_config.model_dump(
|
|
exclude_unset=True,
|
|
mode='json',
|
|
by_alias=True,
|
|
),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
# Add tokens via Web-API
|
|
for token_config in g_default_tokens:
|
|
response = test_client.post(
|
|
'/tokens',
|
|
json=token_config.model_dump(exclude_unset=True, mode='json'),
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
# Add default content via backend instances
|
|
for collection_config in g_default_collections:
|
|
curated_path = Path(store_path / collection_config.curated)
|
|
backend_config = collection_config.backend
|
|
if backend_config.type.startswith('sqlite'):
|
|
backend = SQLiteBackend(curated_path / sqlite_db_filename)
|
|
else:
|
|
backend = RecordDirStore(
|
|
curated_path,
|
|
pid_mapping_function=get_mapping_function_by_name(
|
|
backend_config.mapping_method,
|
|
),
|
|
suffix='yaml',
|
|
)
|
|
pydantic_module = get_model_for_schema(collection_config.schema_location)[0]
|
|
add_records_to_backend(
|
|
backend,
|
|
pydantic_module,
|
|
g_default_entries[collection_config.name],
|
|
)
|
|
return test_client, store_path, admin_token
|
|
|
|
|
|
def add_records_to_backend(
|
|
backend: StorageBackend,
|
|
pydantic_module: ModuleType,
|
|
record_infos: list[tuple[str, str, str]],
|
|
):
|
|
for class_name, record_pid, yaml_stream in record_infos:
|
|
json_object = yaml.load(yaml_stream, Loader=yaml.SafeLoader)
|
|
assert record_pid == json_object['pid']
|
|
backend.add_record(
|
|
iri=resolve_curie(pydantic_module, json_object['pid']),
|
|
class_name=class_name,
|
|
json_object=json_object,
|
|
)
|