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

from __future__ import annotations

import pytest
import time
import yaml
from itertools import count

from dump_things_service import (
    HTTP_200_OK,
    HTTP_404_NOT_FOUND,
)
#from dump_things_service.config import get_config

delete_record = {
    'schema_type': 'abc:Person',
    'pid': 'abc:delete-me',
    'given_name': 'Detlef',
}


@pytest.mark.parametrize('paginate', ('', 'p/'))
@pytest.mark.parametrize('class_name', ('', 'Person'))
def test_read_curated_records(
    fastapi_client_simple,
    paginate,
    class_name,
):
    test_client, _ = fastapi_client_simple

    response = test_client.get(
        f'/collection_1/curated/records/{paginate}{class_name}',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
    )
    assert response.status_code == HTTP_200_OK
    json_object = response.json()

    if 'items' in json_object:
        assert len(json_object['items']) == 3
    else:
        assert len(json_object) == 3

    for pattern, count in (('%25wolf%25', 1), ('%25cura%25', 2)):
        test_client, _ = fastapi_client_simple
        response = test_client.get(
            f'/collection_8/curated/records/{paginate}{class_name}?matching={pattern}',
            headers={'x-dumpthings-token': 'token_1_xxxxx'},
        )
        assert response.status_code == HTTP_200_OK
        json_object = response.json()
        if 'items' in json_object:
            assert len(json_object['items']) == count
        else:
            assert len(json_object) == count


pytest.mark.parametrize(
    'pid',
    ('abc:mode_test', 'abc:some_timee@x.com', 'abc:curated'),
)
def test_read_curated_records_by_pid(fastapi_client_simple):
    test_client, _ = fastapi_client_simple

    response = test_client.get(
        '/no_such_collection/curated/records/',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
    )
    assert response.status_code == HTTP_404_NOT_FOUND


def test_unknown_collection(fastapi_client_simple):
    test_client, _ = fastapi_client_simple

    response = test_client.get(
        '/no_such_collection/curated/records/',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
    )
    assert response.status_code == HTTP_404_NOT_FOUND


def test_curated_delete(fastapi_client_simple):
    test_client, _ = fastapi_client_simple

    response = test_client.post(
        '/collection_8/curated/record/Person',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
        json=delete_record,
    )
    assert response.status_code == HTTP_200_OK

    response = test_client.get(
        '/collection_8/curated/record?pid=abc:delete-me',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
    )
    assert response.status_code == HTTP_200_OK
    assert response.json()['pid'] == 'abc:delete-me'

    response = test_client.delete(
        '/collection_8/curated/record?pid=abc:delete-me',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
    )
    assert response.status_code == HTTP_200_OK
    assert response.json() is True

    response = test_client.get(
        '/collection_8/curated/record?pid=abc:delete-me',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
    )
    assert response.status_code == HTTP_200_OK
    assert response.json() is None

    response = test_client.delete(
        '/collection_8/curated/record?pid=abc:delete-me',
        headers={'x-dumpthings-token': 'token_1_xxxxx'},
    )
    assert response.status_code == HTTP_404_NOT_FOUND


def test_audit_backend(fastapi_client_simple):
    test_client, _ = fastapi_client_simple

    record_id = 'abc:audit-trailed'
    names = 'Frederick', 'Johny'
    tokens = 'token_1_xxxxx', 'token_admin'
    user_names = 'test_user_1_curated', 'test_admin'
    json_objects = tuple(
        {
            'schema_type': 'abc:Person',
            'pid': record_id,
            'given_name': name,
        }
        for name in names
    )

    for i in range(2):
        response = test_client.post(
            f'/collection_1/curated/record/Person?author_id=author_{i}@www.org',
            headers={'x-dumpthings-token': tokens[i]},
            json=json_objects[i],
        )
        assert response.status_code == HTTP_200_OK

    config_instance = get_config()
    audit_backend = config_instance.audit_backends['collection_1'][0]

    changes = audit_backend.get_audit_log(record_id)
    assert len(changes) == 2
    values = tuple(changes.values())
    for i in range(2):
        assert values[i][0] == user_names[i]
        assert values[i][1] == f'author_{i}@www.org'
        assert yaml.safe_load(values[i][3]) == json_objects[i]


def test_audit_backend_auto_flush(fastapi_client_simple):
    test_client, _ = fastapi_client_simple

    record_id = 'abc:audit-trailed'
    names = 'Robert', 'Anton'
    tokens = 'token_1_xxxxx', 'token_admin'
    user_names = 'test_user_1_curated', 'test_admin'
    json_objects = tuple(
        {
            'schema_type': 'abc:Person',
            'pid': record_id,
            'given_name': name,
        }
        for name in names
    )

    for i in range(2):
        response = test_client.post(
            f'/collection_1/curated/record/Person?author_id=author_{i}@www.org',
            headers={'x-dumpthings-token': tokens[i]},
            json=json_objects[i],
        )
        assert response.status_code == HTTP_200_OK

    config_instance = get_config()
    audit_backend = config_instance.audit_backends['collection_1'][0]

    assert audit_backend.current_change_set, 'expected unpersisted changes in audit log'

    for i in count():
        if not audit_backend.current_change_set:
            break
        i += 1
        if i == 10:
            raise ValueError(f'auto flush did not trigger within 10 seconds')
        time.sleep(1)