dump-things-server/dump_things_service/tests/test_collection_administration.py

from pathlib import (
    Path,
    PurePosixPath,
)

from starlette.testclient import TestClient

from dump_things_service import (
    HTTP_201_CREATED,
    HTTP_200_OK,
    HTTP_404_NOT_FOUND,
    HTTP_401_UNAUTHORIZED,
)
from dump_things_service.abstract_config import (
    TokenCollectionConfig,
    TokenModes,
    hash_token_representation,
)
from dump_things_service.collection_endpoints import CollectionRequest
from dump_things_service.token_endpoints import (
    TokenRequest,
    AdminTokenRequest,
)
from dump_things_service.utils import cleaned_json

# String representation of curated- and incoming-path
curated = 'admin_test_curated'
incoming = 'admin_test_incoming'

# Path to a local simple test schema
test_schema_location = str((Path(__file__).parent / 'testschema.yaml').absolute())

new_collection_name = 'admin_test_collection'
new_token_name = 'admin_test_token'
new_token_representation = 'admin_test_token'
new_collection_request = CollectionRequest(
    name=new_collection_name,
    default_token='test_default_token',
    curated=PurePosixPath(f'{curated}/admin_test_collection'),
    schema=test_schema_location,
    incoming=PurePosixPath(f'{incoming}/admin_test_collection'),
)

new_token_request = TokenRequest(
    name=new_token_name,
    user_id='admin_test_token_user',
    hashed=False,
    representation=new_token_representation,
    collections={
        new_collection_name: TokenCollectionConfig(
            mode=TokenModes.WRITE_COLLECTION,
            incoming_label=f'{new_collection_name}_label',
        )
    },
)

new_admin_token_name='New_Admin_Token'
plain_new_admin_token = 'admin-XXX'
new_admin_token_request = AdminTokenRequest(
    name=new_admin_token_name,
    representation=hash_token_representation(plain_new_admin_token),
)


def _name_in_openapi_paths(
        test_client: TestClient,
        name: str,
) -> bool:
    response = test_client.get('/openapi.json')
    open_api = response.json()
    for path in open_api['paths'].keys():
        if name in path:
            return True
    return False


def test_collection_adding(fastapi_client_simple):
    test_client, _, admin_token = fastapi_client_simple

    # Check that the collection does not yet exist
    response = test_client.get(
        f'/collections/{new_collection_name}',
        headers={'x-dumpthings-token': admin_token},
    )
    assert response.status_code == HTTP_404_NOT_FOUND
    assert not _name_in_openapi_paths(test_client, new_collection_name)

    # Add a new collection
    response = test_client.post(
        '/collections',
        headers={'x-dumpthings-token': admin_token},
        json=new_collection_request.model_dump(mode='json', by_alias=True),
    )
    assert response.status_code == HTTP_201_CREATED
    assert _name_in_openapi_paths(test_client, new_collection_name)

    response = test_client.get(
        f'/collections/{new_collection_name}',
        headers={'x-dumpthings-token': admin_token},
    )
    assert response.status_code == HTTP_200_OK
    new_collection_config = new_collection_request.model_dump(mode='json', by_alias=True)
    del new_collection_config['name']
    assert response.json() == new_collection_config

    # Add a token to the collection
    response = test_client.post(
        '/tokens',
        headers={'x-dumpthings-token': admin_token},
        json=new_token_request.model_dump(mode='json'),
    )
    assert response.status_code == HTTP_201_CREATED

    # Read the token back
    response = test_client.get(
        f'/tokens/{new_token_name}',
        headers={'x-dumpthings-token': admin_token},
    )
    assert response.status_code == HTTP_200_OK
    assert response.json() == {
        'name': new_token_request.name,
        'user_id': new_token_request.user_id,
        'collections': new_token_request.model_dump(mode='json')['collections'],
    }

    new_record = {
        'pid': 'http://example.com/admin-test-1',
        'given_name': 'Admin Test 1',
        'schema_type': 'abc:Person',
    }

    # Add a record to the collection
    response = test_client.post(
        f'/{new_collection_name}/record/Person',
        headers={'x-dumpthings-token': new_token_representation},
        json=new_record,
    )
    assert response.status_code == HTTP_200_OK

    # Read the record back
    response = test_client.get(
        f'/{new_collection_name}/records/Person',
        headers={'x-dumpthings-token': new_token_representation},
    )
    assert response.status_code == HTTP_200_OK
    assert cleaned_json(response.json()[0], ('annotations',)) == new_record

    # Remove the token
    response = test_client.delete(
        f'/tokens/{new_token_name}',
        headers={'x-dumpthings-token': admin_token},
    )
    assert response.status_code == HTTP_200_OK

    # Check that posting is not possible with the removed token
    response = test_client.post(
        f'/{new_collection_name}/record/Person',
        headers={'x-dumpthings-token': new_token_representation},
        json=new_record,
    )
    assert response.status_code == HTTP_401_UNAUTHORIZED

    # Remove the collection
    response = test_client.delete(
        f'/collections/{new_collection_name}',
        headers={'x-dumpthings-token': admin_token},
    )
    assert response.status_code == HTTP_200_OK

    # Check that the collection endpoints are not found
    response = test_client.get(
        f'/{new_collection_name}/records/Person',
        headers={'x-dumpthings-token': new_token_representation},
    )
    assert response.status_code == HTTP_404_NOT_FOUND

    # Check that the openapi document is adjusted
    assert not _name_in_openapi_paths(test_client, new_collection_name)


def test_collection_reading(fastapi_client_simple):
    test_client, _, admin_token = fastapi_client_simple

    # Check that the new admin token is not yet working
    response = test_client.get(
        f'/collections',
        headers={'x-dumpthings-token': admin_token},
    )
    assert response.status_code == HTTP_200_OK
    response_object = response.json()
    assert isinstance(response_object, dict)
    assert len(response_object) == 10


def test_admin_token_management(fastapi_client_simple):
    test_client, _, admin_token = fastapi_client_simple

    # Check that the new admin token is not yet working
    response = test_client.get(
        f'/collections/collection_1',
        headers={'x-dumpthings-token': plain_new_admin_token},
    )
    assert response.status_code == HTTP_401_UNAUTHORIZED

    # Add a new admin token
    response = test_client.post(
        '/admin_tokens',
        headers={'x-dumpthings-token': admin_token},
        json=new_admin_token_request.model_dump(mode='json'),
    )
    assert response.status_code == HTTP_201_CREATED

    # Try the new token
    response = test_client.get(
        f'/collections/collection_1',
        headers={'x-dumpthings-token': plain_new_admin_token},
    )
    assert response.status_code == HTTP_200_OK

    # Check that the token shows up in the token list
    response = test_client.get(
        f'/admin_tokens',
        headers={'x-dumpthings-token': plain_new_admin_token},
    )
    assert response.status_code == HTTP_200_OK
    assert new_admin_token_name in response.json()

    # Delete the new admin token
    response = test_client.delete(
        f'/admin_tokens/{new_admin_token_name}',
        headers={'x-dumpthings-token': plain_new_admin_token},
    )
    assert response.status_code == HTTP_200_OK

    response = test_client.get(
        f'/admin_tokens',
        headers={'x-dumpthings-token': admin_token},
    )
    assert response.status_code == HTTP_200_OK
    assert new_admin_token_name not in response.json()