test_api.py

# Copyright 2018 Markus Scheidgen
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an"AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import Any, Iterable
import pytest
import time
import json
import zipfile
import io
import inspect
import datetime
import os.path
from urllib.parse import urlencode
import base64
import itertools
from hashlib import md5

from nomad.app.common import rfc3339DateTime
from nomad.app.api.auth import generate_upload_token
from nomad import search, files, config, utils, infrastructure
from nomad.metainfo import search_extension
from nomad.files import UploadFiles, PublicUploadFiles
from nomad.processing import Upload, Calc, SUCCESS
from nomad.datamodel import EntryMetadata, User, Dataset

from tests.conftest import create_auth_headers, clear_elastic, clear_raw_files
from tests.test_files import example_file, example_file_mainfile, example_file_contents
from tests.test_files import create_staging_upload, create_public_upload, assert_upload_files
from tests.test_search import assert_search_upload
from tests.processing import test_data as test_processing

from tests.app.test_app import BlueprintClient

logger = utils.get_logger(__name__)

example_statistics = ['atoms', 'authors', 'dft.system']


@pytest.fixture(scope='function')
def api(client):
    return BlueprintClient(client, '/api')


@pytest.fixture(scope='function')
def test_user_signature_token(api, test_user_auth):
    rv = api.get('/auth/', headers=test_user_auth)
    assert rv.status_code == 200
    return json.loads(rv.data)['signature_token']


def get_upload_entries_metadata(upload: dict) -> Iterable[EntryMetadata]:
    ''' Create a iterable of :class:`EntryMetadata` from a API upload json record. '''
    return [
        EntryMetadata(domain='dft', calc_id=entry['calc_id'], mainfile=entry['mainfile'])
        for entry in upload['calcs']['results']]


def assert_zip_file(rv, files: int = -1, basename: bool = None):
    assert rv.status_code == 200
    assert len(rv.data) > 0
    with zipfile.ZipFile(io.BytesIO(rv.data)) as zip_file:
        assert zip_file.testzip() is None
        zip_files = zip_file.namelist()
        if files >= 0:
            assert len(zip_files) == files
        if basename is not None:
            if basename:
                assert all(
                    os.path.basename(name) == name
                    for name in zip_files if name != 'manifest.json')
            else:
                assert all(
                    os.path.basename(name) != name
                    for name in zip_files for name in zip_files if name != 'manifest.json')


class TestInfo:
    def test_info(self, api, elastic):
        rv = api.get('/info/')
        assert rv.status_code == 200

        data = json.loads(rv.data)
        assert 'codes' in data
        assert 'parsers' in data
        assert 'statistics' in data
        assert len(data['parsers']) >= len(data['codes'])
        assert len(data['domains']) >= 1
        assert rv.status_code == 200

        rv = api.get('/info/')
        assert rv.status_code == 200


class TestKeycloak:
    def test_auth_wo_credentials(self, api, keycloak, no_warn):
        rv = api.get('/auth/')
        assert rv.status_code == 401

    @pytest.fixture(scope='function')
    def auth_headers(self, api, keycloak):
        basic_auth = base64.standard_b64encode(b'sheldon.cooper@nomad-coe.eu:password')
        rv = api.get('/auth/', headers=dict(Authorization='Basic %s' % basic_auth.decode('utf-8')))
        assert rv.status_code == 200
        auth = json.loads(rv.data)
        assert 'access_token' in auth
        assert auth['access_token'] is not None
        return dict(Authorization='Bearer %s' % auth['access_token'])

    def test_auth_with_password(self, api, auth_headers):
        pass

    def test_auth_with_access_token(self, api, auth_headers):
        rv = api.get('/auth/', headers=auth_headers)
        assert rv.status_code == 200

    def assert_sheldon(self, user):
        assert user.email is not None
        assert user.name == 'Sheldon Cooper'
        assert user.first_name == 'Sheldon'
        assert user.last_name == 'Cooper'
        assert user.created is not None
        assert user.affiliation is not None
        assert user.affiliation_address is not None

    def test_get_user(self, keycloak):
        user = infrastructure.keycloak.get_user(username='scooper')
        self.assert_sheldon(user)

    def test_search_user(self, keycloak):
        users = infrastructure.keycloak.search_user(query='Sheldon')
        assert len(users) == 1
        self.assert_sheldon(users[0])


class TestAuth:
    def test_auth_wo_credentials(self, api, no_warn):
        rv = api.get('/auth/')
        assert rv.status_code == 401

    def test_auth_with_token(self, api, test_user_auth):
        rv = api.get('/auth/', headers=test_user_auth)
        assert rv.status_code == 200
        self.assert_auth(api, json.loads(rv.data))

    def assert_auth(self, api, auth):
        assert 'user' not in auth
        assert 'access_token' in auth
        assert 'upload_token' in auth
        assert 'signature_token' in auth

    def test_signature_token(self, test_user_signature_token, no_warn):
        assert test_user_signature_token is not None

    def test_users(self, api):
        rv = api.get('/auth/users?query=Sheldon')
        assert rv.status_code == 200
        data = json.loads(rv.data)
        assert len(data['users'])
        keys = data['users'][0].keys()
        required_keys = ['name', 'email', 'user_id']
        assert all(key in keys for key in required_keys)
        for key in keys:
            assert data['users'][0].get(key) is not None

    def test_invite(self, api, test_user_auth, no_warn):
        rv = api.put(
            '/auth/users', headers=test_user_auth, content_type='application/json',
            data=json.dumps({
                'first_name': 'John',
                'last_name': 'Doe',
                'affiliation': 'Affiliation',
                'email': 'john.doe@affiliation.edu'
            }))
        assert rv.status_code == 200
        data = json.loads(rv.data)
        keys = data.keys()
        required_keys = ['name', 'email', 'user_id']
        assert all(key in keys for key in required_keys)


class TestUploads:

    def assert_uploads(self, upload_json_str, count=0, **kwargs):
        data = json.loads(upload_json_str)
        assert 'pagination' in data
        assert 'page' in data['pagination']

        data = data['results']
        assert isinstance(data, list)
        assert len(data) == count

        if count > 0:
            self.assert_upload(json.dumps(data[0]), **kwargs)

    def assert_upload(self, upload_json_str, id=None, **kwargs):
        data = json.loads(upload_json_str)
        assert 'upload_id' in data
        if id is not None:
            assert id == data['upload_id']
        assert 'create_time' in data

        for key, value in kwargs.items():
            assert data.get(key, None) == value

        return data

    def assert_processing(self, api, test_user_auth, upload_id):
        upload_endpoint = '/uploads/%s' % upload_id

        # poll until completed
        upload = self.block_until_completed(api, upload_id, test_user_auth)

        assert len(upload['tasks']) == 4
        assert upload['tasks_status'] == SUCCESS
        assert upload['current_task'] == 'cleanup'
        assert not upload['process_running']

        calcs = upload['calcs']['results']
        for calc in calcs:
            assert calc['tasks_status'] == SUCCESS
            assert calc['current_task'] == 'archiving'
            assert len(calc['tasks']) == 3

            assert 'atoms' in calc['metadata']
            assert api.get('/archive/logs/%s/%s' % (calc['upload_id'], calc['calc_id']), headers=test_user_auth).status_code == 200

        if upload['calcs']['pagination']['total'] > 1:
            rv = api.get('%s?page=2&per_page=1&order_by=tasks_status' % upload_endpoint, headers=test_user_auth)
            assert rv.status_code == 200
            upload = self.assert_upload(rv.data)
            assert len(upload['calcs']['results']) == 1

        entries = get_upload_entries_metadata(upload)
        assert_upload_files(upload_id, entries, files.StagingUploadFiles)
        assert_search_upload(entries, additional_keys=['atoms', 'dft.system'])

    def assert_published(self, api, test_user_auth, upload_id, proc_infra, metadata={}):
        rv = api.get('/uploads/%s' % upload_id, headers=test_user_auth)
        upload = self.assert_upload(rv.data)

        rv = api.post(
            '/uploads/%s' % upload_id,
            headers=test_user_auth,
            data=json.dumps(dict(operation='publish', metadata=metadata)),
            content_type='application/json')
        assert rv.status_code == 200
        upload = self.assert_upload(rv.data)
        assert upload['current_process'] == 'publish_upload'
        assert upload['process_running']

        additional_keys = ['with_embargo']
        if 'external_id' in metadata:
            additional_keys.append('external_id')

        self.block_until_completed(api, upload_id, test_user_auth)

        upload_proc = Upload.objects(upload_id=upload_id).first()
        assert upload_proc is not None
        assert upload_proc.published is True
        assert upload_proc.embargo_length == min(36, metadata.get('embargo_length', 36))

        with upload_proc.entries_metadata() as entries:
            for entry in entries:
                for key, transform in {
                        'comment': lambda e: e.comment,
                        'with_embargo': lambda e: e.with_embargo,
                        'references': lambda e: e.references,
                        'coauthors': lambda e: [u.user_id for u in e.coauthors],
                        '_uploader': lambda e: e.uploader.user_id,
                        '_pid': lambda e: str(e.pid),
                        'external_id': lambda e: e.external_id}.items():
                    if key in metadata:
                        assert transform(entry) == metadata[key], key

        assert_upload_files(upload_id, entries, files.PublicUploadFiles, published=True)
        assert_search_upload(entries, additional_keys=additional_keys, published=True)

    def block_until_completed(self, api, upload_id: str, test_user_auth):
        while True:
            time.sleep(0.1)
            rv = api.get('/uploads/%s' % upload_id, headers=test_user_auth)
            if rv.status_code == 200:
                upload = self.assert_upload(rv.data)
                if not upload['process_running'] and not upload['tasks_running']:
                    return upload
            elif rv.status_code == 404:
                return None
            else:
                raise Exception(
                    'unexpected status code while blocking for upload processing: %s' %
                    str(rv.status_code))

    def assert_upload_does_not_exist(self, api, upload_id: str, test_user_auth):
        self.block_until_completed(api, upload_id, test_user_auth)

        rv = api.get('/uploads/%s' % upload_id, headers=test_user_auth)
        assert rv.status_code == 404
        assert Upload.objects(upload_id=upload_id).first() is None
        assert Calc.objects(upload_id=upload_id).count() is 0
        upload_files = UploadFiles.get(upload_id)
        assert upload_files is None or isinstance(upload_files, PublicUploadFiles)

    def test_get_command(self, api, test_user_auth, no_warn):
        rv = api.get('/uploads/command', headers=test_user_auth)
        assert rv.status_code == 200
        data = json.loads(rv.data)
        assert 'upload_command' in data
        assert '/api/uploads' in data['upload_command']
        assert 'upload_url' in data

    def test_get_empty(self, api, test_user_auth, no_warn):
        rv = api.get('/uploads/', headers=test_user_auth)

        assert rv.status_code == 200
        self.assert_uploads(rv.data, count=0)

    def test_get_not_existing(self, api, test_user_auth, no_warn):
        rv = api.get('/uploads/123456789012123456789012', headers=test_user_auth)
        assert rv.status_code == 404

    def test_put_upload_token(self, api, non_empty_example_upload, test_user):
        url = '/uploads/?token=%s&local_path=%s&name=test_upload' % (
            generate_upload_token(test_user), non_empty_example_upload)
        rv = api.put(url)
        assert rv.status_code == 200
        assert 'Thanks for uploading' in rv.data.decode('utf-8')

    @pytest.mark.parametrize('mode', ['multipart', 'stream', 'local_path'])
    @pytest.mark.parametrize('name', [None, 'test_name'])
    def test_put(self, api, test_user_auth, proc_infra, example_upload, mode, name, no_warn):
        file = example_upload
        if name:
            url = '/uploads/?name=%s' % name
        else:
            url = '/uploads/'

        if mode == 'multipart':
            rv = api.put(
                url, data=dict(file=(open(file, 'rb'), 'the_name')), headers=test_user_auth)
            if not name:
                name = 'the_name'
        elif mode == 'stream':
            with open(file, 'rb') as f:
                rv = api.put(url, data=f.read(), headers=test_user_auth)
        elif mode == 'local_path':
            url += '&' if name else '?'
            url += 'local_path=%s' % file
            rv = api.put(url, headers=test_user_auth)
        else:
            assert False

        assert rv.status_code == 200
        if mode == 'local_path':
            upload = self.assert_upload(rv.data, upload_path=file, name=name)
        else:
            upload = self.assert_upload(rv.data, name=name)
        assert upload['tasks_running']

        self.assert_processing(api, test_user_auth, upload['upload_id'])

    @pytest.mark.timeout(config.tests.default_timeout)
    def test_upload_limit(self, api, mongo, test_user, test_user_auth, proc_infra):
        for _ in range(0, config.services.upload_limit):
            Upload.create(user=test_user)
        file = example_file
        rv = api.put('/uploads/?local_path=%s' % file, headers=test_user_auth)
        assert rv.status_code == 400
        assert Upload.user_uploads(test_user).count() == config.services.upload_limit

    def test_delete_not_existing(self, api, test_user_auth, no_warn):
        rv = api.delete('/uploads/123456789012123456789012', headers=test_user_auth)
        assert rv.status_code == 404

    @pytest.fixture(scope='function')
    def slow_processing(self, monkeypatch):
        old_cleanup = Upload.cleanup

        def slow_cleanup(self):
            time.sleep(0.5)
            old_cleanup(self)

        monkeypatch.setattr('nomad.processing.data.Upload.cleanup', slow_cleanup)
        yield True
        monkeypatch.setattr('nomad.processing.data.Upload.cleanup', old_cleanup)

    def test_delete_published(self, api, test_user_auth, proc_infra, no_warn):
        rv = api.put('/uploads/?local_path=%s' % example_file, headers=test_user_auth)
        upload = self.assert_upload(rv.data)
        self.assert_processing(api, test_user_auth, upload['upload_id'])
        self.assert_published(api, test_user_auth, upload['upload_id'], proc_infra)
        rv = api.delete('/uploads/%s' % upload['upload_id'], headers=test_user_auth)
        assert rv.status_code == 400

    def test_delete(self, api, test_user_auth, proc_infra, no_warn):
        rv = api.put('/uploads/?local_path=%s' % example_file, headers=test_user_auth)
        upload = self.assert_upload(rv.data)
        self.assert_processing(api, test_user_auth, upload['upload_id'])
        rv = api.delete('/uploads/%s' % upload['upload_id'], headers=test_user_auth)
        assert rv.status_code == 200
        self.assert_upload_does_not_exist(api, upload['upload_id'], test_user_auth)

    def test_post_empty(self, api, test_user_auth, empty_upload, proc_infra, no_warn):
        rv = api.put('/uploads/?local_path=%s' % empty_upload, headers=test_user_auth)
        assert rv.status_code == 200
        upload = self.assert_upload(rv.data)
        self.assert_processing(api, test_user_auth, upload['upload_id'])
        rv = api.post(
            '/uploads/%s' % upload['upload_id'], headers=test_user_auth,
            data=json.dumps(dict(operation='publish')),
            content_type='application/json')
        assert rv.status_code == 400

    def test_post(self, api, test_user_auth, non_empty_example_upload, proc_infra, no_warn):
        rv = api.put('/uploads/?local_path=%s' % non_empty_example_upload, headers=test_user_auth)
        assert rv.status_code == 200
        upload = self.assert_upload(rv.data)
        self.assert_processing(api, test_user_auth, upload['upload_id'])
        self.assert_published(api, test_user_auth, upload['upload_id'], proc_infra)

        # still visible
        assert api.get('/uploads/%s' % upload['upload_id'], headers=test_user_auth).status_code == 200
        # still listed with all=True
        rv = api.get('/uploads/?state=all', headers=test_user_auth)
        assert rv.status_code == 200
        data = json.loads(rv.data)['results']
        assert len(data) > 0
        assert any(item['upload_id'] == upload['upload_id'] for item in data)
        # not listed with all=False
        rv = api.get('/uploads/', headers=test_user_auth)
        assert rv.status_code == 200
        data = json.loads(rv.data)['results']
        assert not any(item['upload_id'] == upload['upload_id'] for item in data)

    def test_post_metadata(
            self, api, proc_infra, admin_user_auth, test_user_auth, test_user,
            other_test_user, no_warn, example_user_metadata):
        rv = api.put('/uploads/?local_path=%s' % example_file, headers=test_user_auth)
        upload = self.assert_upload(rv.data)
        self.assert_processing(api, test_user_auth, upload['upload_id'])
        metadata = dict(**example_user_metadata)
        metadata['_upload_time'] = datetime.datetime.utcnow().isoformat()
        self.assert_published(api, admin_user_auth, upload['upload_id'], proc_infra, metadata)

    def test_post_metadata_forbidden(self, api, proc_infra, test_user_auth, no_warn):
        rv = api.put('/uploads/?local_path=%s' % example_file, headers=test_user_auth)
        upload = self.assert_upload(rv.data)
        self.assert_processing(api, test_user_auth, upload['upload_id'])
        rv = api.post(
            '/uploads/%s' % upload['upload_id'],
            headers=test_user_auth,
            data=json.dumps(dict(operation='publish', metadata=dict(_pid='256'))),
            content_type='application/json')
        assert rv.status_code == 401

    def test_post_metadata_and_republish(
            self, api, proc_infra, admin_user_auth, test_user_auth, test_user,
            other_test_user, no_warn, example_user_metadata):
        rv = api.put('/uploads/?local_path=%s' % example_file, headers=test_user_auth)
        upload = self.assert_upload(rv.data)
        self.assert_processing(api, test_user_auth, upload['upload_id'])
        metadata = dict(**example_user_metadata)
        metadata['_upload_time'] = datetime.datetime.utcnow().isoformat()
        self.assert_published(api, admin_user_auth, upload['upload_id'], proc_infra, metadata)
        self.assert_published(api, admin_user_auth, upload['upload_id'], proc_infra, {})

    def test_post_re_process(self, api, published, test_user_auth, monkeypatch):
        monkeypatch.setattr('nomad.config.meta.version', 're_process_test_version')
        monkeypatch.setattr('nomad.config.meta.commit', 're_process_test_commit')

        upload_id = published.upload_id
        rv = api.post(
            '/uploads/%s' % upload_id,
            headers=test_user_auth,
            data=json.dumps(dict(operation='re-process')),
            content_type='application/json')

        assert rv.status_code == 200
        assert self.block_until_completed(api, upload_id, test_user_auth) is not None

    # TODO validate metadata (or all input models in API for that matter)
    # def test_post_bad_metadata(self, api, proc_infra, test_user_auth):
    #     rv = api.put('/uploads/?local_path=%s' % example_file, headers=test_user_auth)
    #     upload = self.assert_upload(rv.data)
    #     self.assert_processing(api, test_user_auth, upload['upload_id'])
    #     rv = api.post(
    #         '/uploads/%s' % upload['upload_id'],
    #         headers=test_user_auth,
    #         data=json.dumps(dict(operation='publish', metadata=dict(doesnotexist='hi'))),
    #         content_type='application/json')
    #     assert rv.status_code == 400

    @pytest.mark.parametrize('upload_file, ending', [
        ('examples_potcar.zip', ''),
        ('examples_potcar_gz.tgz', '.gz'),
        ('examples_potcar_xz.tgz', '.xz')])
    def test_potcar(self, api, proc_infra, test_user_auth, upload_file, ending):
        # only the owner, shared with people are supposed to download the original potcar file
        example_file = 'tests/data/proc/%s' % upload_file
        rv = api.put('/uploads/?local_path=%s' % example_file, headers=test_user_auth)

        upload = self.assert_upload(rv.data)
        upload_id = upload['upload_id']
        self.assert_processing(api, test_user_auth, upload_id)
        self.assert_published(api, test_user_auth, upload_id, proc_infra)
        rv = api.get('/raw/%s/examples_potcar/POTCAR%s' % (upload_id, ending))
        assert rv.status_code == 401
        rv = api.get('/raw/%s/examples_potcar/POTCAR%s' % (upload_id, ending), headers=test_user_auth)
        assert rv.status_code == 200
        rv = api.get('/raw/%s/examples_potcar/POTCAR%s.stripped' % (upload_id, ending))
        assert rv.status_code == 200


today = datetime.datetime.utcnow().date()
today_datetime = datetime.datetime(*today.timetuple()[:6])


class UploadFilesBasedTests:

    @staticmethod
    def fix_signature(func, wrapper):
        additional_args = list(inspect.signature(func).parameters.values())[4:]
        wrapper_sig = inspect.signature(wrapper)
        wrapper_args = list(wrapper_sig.parameters.values())[:3] + additional_args
        wrapper_sig = wrapper_sig.replace(parameters=tuple(wrapper_args))
        wrapper.__signature__ = wrapper_sig

    @staticmethod
    def check_authorization(func):
        @pytest.mark.parametrize('test_data', [
            [True, None, True],      # in staging for upload
            [True, None, False],     # in staging for different user
            [True, None, None],      # in staging for guest
            [True, None, 'admin'],   # in staging, for admin
            [False, True, True],     # in public, restricted for uploader
            [False, True, False],    # in public, restricted for different user
            [False, True, None],     # in public, restricted for guest
            [False, True, 'admin'],  # in public, restricted for admin
            [False, False, True],    # in public, public, for uploader
            [False, False, False],   # in public, public, for different user
            [False, False, None]     # in public, public, for guest
        ], indirect=True)
        def wrapper(self, api, test_data, *args, **kwargs):
            upload, authorized, auth_headers = test_data
            try:
                func(self, api, upload, auth_headers, *args, **kwargs)
            except AssertionError as assertion:
                assertion_str = str(assertion)
                if not authorized:
                    if '0 == 5' in assertion_str:
                        # the user is not authorized an gets an empty zip as expected
                        return
                    if '401' in assertion_str:
                        # the user is not authorized and gets a 401 as expected
                        return
                raise assertion

            if not authorized:
                assert False
        UploadFilesBasedTests.fix_signature(func, wrapper)
        return wrapper

    @staticmethod
    def ignore_authorization(func):
        @pytest.mark.parametrize('test_data', [
            [True, None, True],      # in staging
            [False, False, None],    # in public
        ], indirect=True)
        def wrapper(self, api, test_data, *args, **kwargs):
            upload, _, auth_headers = test_data
            func(self, api, upload, auth_headers, *args, **kwargs)
        UploadFilesBasedTests.fix_signature(func, wrapper)
        return wrapper

    @pytest.fixture(scope='function')
    def test_data(self, request, mongo, raw_files, no_warn, test_user, other_test_user, admin_user):
        # delete potential old test files
        for _ in [0, 1]:
            upload_files = UploadFiles.get('test_upload')
            if upload_files:
                upload_files.delete()

        in_staging, restricted, for_uploader = request.param

        if in_staging:
            authorized = for_uploader is True or for_uploader == 'admin'
        else:
            authorized = not restricted or for_uploader is True or for_uploader == 'admin'

        if for_uploader is True:
            auth_headers = create_auth_headers(test_user)
        elif for_uploader is False:
            auth_headers = create_auth_headers(other_test_user)
        elif for_uploader == 'admin':
            auth_headers = create_auth_headers(admin_user)
        else:
            auth_headers = None

        calc_specs = 'r' if restricted else 'p'
        Upload.create(user=test_user, upload_id='test_upload')
        if in_staging:
            _, _, upload_files = create_staging_upload('test_upload', calc_specs=calc_specs)
        else:
            _, _, upload_files = create_public_upload('test_upload', calc_specs=calc_specs)

        yield 'test_upload', authorized, auth_headers

        upload_files.delete()


class TestArchive(UploadFilesBasedTests):
    @UploadFilesBasedTests.check_authorization
    def test_get(self, api, upload, auth_headers):
        rv = api.get('/archive/%s/0' % upload, headers=auth_headers)
        assert rv.status_code == 200
        data = json.loads(rv.data)
        assert data is not None
        assert 'section_metadata' in data
        assert 'section_run' in data

    @UploadFilesBasedTests.ignore_authorization
    def test_get_signed(self, api, upload, _, test_user_signature_token):
        rv = api.get('/archive/%s/0?signature_token=%s' % (upload, test_user_signature_token))
        assert rv.status_code == 200
        assert json.loads(rv.data) is not None

    @UploadFilesBasedTests.check_authorization
    def test_get_calc_proc_log(self, api, upload, auth_headers):
        rv = api.get('/archive/logs/%s/0' % upload, headers=auth_headers)
        assert rv.status_code == 200
        assert len(rv.data) > 0

    @UploadFilesBasedTests.ignore_authorization
    def test_get_calc_proc_log_signed(self, api, upload, _, test_user_signature_token):
        rv = api.get('/archive/logs/%s/0?signature_token=%s' % (upload, test_user_signature_token))
        assert rv.status_code == 200
        assert len(rv.data) > 0

    @UploadFilesBasedTests.ignore_authorization
    def test_get_non_existing_archive(self, api, upload, auth_headers):
        rv = api.get('/archive/%s' % 'doesnt/exist', headers=auth_headers)
        assert rv.status_code == 404

    @pytest.mark.parametrize('compress', [False, True])
    def test_archive_zip_dowload_upload_id(self, api, non_empty_processed, test_user_auth, compress):
        url = '/archive/download?upload_id=%s&compress=%s' % (non_empty_processed.upload_id, 'true' if compress else 'false')
        rv = api.get(url, headers=test_user_auth)

        assert rv.status_code == 200
        assert_zip_file(rv, files=2)

    @pytest.mark.parametrize('query_params', [
        {'atoms': 'Si'},
        {'authors': 'Sheldon Cooper'}
    ])
    def test_archive_zip_dowload(self, api, processeds, test_user_auth, query_params):

        url = '/archive/download?%s' % urlencode(query_params)
        rv = api.get(url, headers=test_user_auth)

        assert rv.status_code == 200
        assert_zip_file(rv, files=len(processeds) + 1)
        with zipfile.ZipFile(io.BytesIO(rv.data)) as zip_file:
            with zip_file.open('manifest.json', 'r') as f:
                manifest = json.load(f)
                assert len(manifest) == len(processeds)

    def test_archive_zip_dowload_empty(self, api, elastic):
        url = '/archive/download?upload_id=doesNotExist'
        rv = api.get(url)

        assert rv.status_code == 200
        assert_zip_file(rv, files=1)

    @pytest.mark.parametrize('required', [{'section_workflow': '*'}, {'section_run': '*'}])
    def test_archive_query_paginated(self, api, published_wo_user_metadata, required):
        data = {'required': required, 'pagination': {'per_page': 5}}
        uri = '/archive/query'
        rv = api.post(uri, content_type='application/json', data=json.dumps(data))

        assert rv.status_code == 200
        data = rv.get_json()

        assert data
        results = data.get('results', None)
        assert len(results) > 0
        for result in results:
            assert 'calc_id' in result
            assert 'parser_name' in result
            assert 'archive' in result

        # TODO assert archive contents

    def test_archive_not_exists(self, api, published_wo_user_metadata):
        entry_metadata = EntryMetadata(
            domain='dft', upload_id=published_wo_user_metadata.upload_id,
            calc_id='test_id', published=True, with_embargo=False)
        entry_metadata.a_elastic.index(refresh=True)

        data = {}
        uri = '/archive/query'
        rv = api.post(uri, content_type='application/json', data=json.dumps(data))

        rv = api.post(uri, content_type='application/json', data=json.dumps(dict(per_page=5, raise_errors=True)))
        assert rv.status_code == 404

        rv = api.post(uri, content_type='application/json', data=json.dumps(dict(per_page=5, raise_errors=False)))
        assert rv.status_code == 200

    def test_archive_query_aggregated(self, api, published_wo_user_metadata):
        uri = '/archive/query'
        schema = {
            'section_run': {
                'section_single_configuration_calculation': {
                    'energy_total': '*'}}}

        query = {'required': schema, 'aggregation': {'per_page': 1}}

        count = 0
        while True:
            rv = api.post(uri, content_type='application/json', data=json.dumps(query))
            assert rv.status_code == 200
            data = rv.get_json()
            results = data.get('results', None)
            count += len(results)
            after = data['aggregation']['after']
            if after is None:
                break

            query['aggregation']['after'] = after

        assert count > 0

    @pytest.mark.timeout(config.tests.default_timeout)
    @pytest.fixture(scope='function')
    def example_upload(self, proc_infra, test_user):
        path = 'tests/data/proc/example_vasp_with_binary.zip'
        results = []
        for uid in range(2):
            upload_id = 'vasp_%d' % uid
            processed = test_processing.run_processing((upload_id, path), test_user)

            processed.publish_upload()
            try:
                processed.block_until_complete(interval=.01)
            except Exception:
                pass

            results.append(processed)

        return results

    @pytest.mark.parametrize('query_expression, nresults', [
        pytest.param({}, 4, id='empty'),
        pytest.param({'dft.system': 'bulk'}, 4, id='match'),
        pytest.param({'$gte': {'n_atoms': 1}}, 4, id='comparison'),
        pytest.param({
            '$and': [
                {'dft.system': 'bulk'}, {'$not': [{'dft.compound_type': 'ternary'}]}
            ]
        }, 2, id="and-with-not"),
        pytest.param({
            '$or': [
                {'upload_id': ['vasp_0']}, {'$gte': {'n_atoms': 1}}
            ]
        }, 4, id="or-with-gte"),
        pytest.param({
            '$not': [{'dft.spacegroup': 221}, {'dft.spacegroup': 227}]
        }, 0, id="not"),
        pytest.param({
            '$and': [
                {'dft.code_name': 'VASP'},
                {'$gte': {'n_atoms': 3}},
                {'$lte': {'dft.workflow.section_geometry_optimization.final_energy_difference': 1e-24}}
            ]}, 0, id='client-example')
    ])
    def test_post_archive_query(self, api, example_upload, query_expression, nresults):
        data = {'pagination': {'per_page': 5}, 'query': query_expression}
        rv = api.post('/archive/query', content_type='application/json', data=json.dumps(data))

        assert rv.status_code == 200
        data = rv.get_json()

        assert data
        results = data.get('results', None)
        assert len(results) == nresults

    @pytest.mark.parametrize('query', [
        pytest.param({'$bad_op': {'n_atoms': 1}}, id='bad-op')
    ])
    def test_post_archive_bad_query(self, api, query):
        rv = api.post(
            '/archive/query', content_type='application/json',
            data=json.dumps(dict(query=query)))

        assert rv.status_code == 400


class TestMetainfo():
    @pytest.mark.parametrize('package', ['common', 'vasp'])
    def test_regular(self, api, package):
        rv = api.get('/metainfo/%s' % package)
        assert rv.status_code == 200
        assert len(rv.get_json()) > 0

    def test_full_name(self, api):
        rv = api.get('/metainfo/nomad.datamodel.metainfo.common')
        assert rv.status_code == 200

    def test_extension(self, api):
        rv = api.get('/metainfo/common.json')
        assert rv.status_code == 200

        rv = api.get('/metainfo/legacy/common.json')
        assert rv.status_code == 200

        rv = api.get('/metainfo/legacy/common.nomadmetainfo.json')
        assert rv.status_code == 200

    @pytest.mark.parametrize('package', ['common', 'vasp'])
    def test_legacy(self, api, package):
        rv = api.get('/metainfo/legacy/%s' % package)
        assert rv.status_code == 200
        assert len(rv.get_json().get('metaInfos')) > 0

    def test_does_not_exist(self, api):
        rv = api.get('/metainfo/doesnotexist')
        assert rv.status_code == 404
        rv = api.get('/metainfo/legacy/doesnotexist')
        assert rv.status_code == 404

    def test_all(self, api):
        rv = api.get('/metainfo/')
        rv.status_code == 200
        assert len(rv.get_json()) > 0


class TestRepo():
    @pytest.fixture(scope='class')
    def example_elastic_calcs(
            self, elastic_infra, raw_files_infra, normalized,
            test_user: User, other_test_user: User):
        clear_elastic(elastic_infra)

        example_dataset = Dataset(
            dataset_id='ds_id', name='ds_name', user_id=test_user.user_id, doi='ds_doi')
        example_dataset.a_mongo.create()

        entry_metadata = EntryMetadata(
            domain='dft', upload_id='example_upload_id', calc_id='0', upload_time=today_datetime)
        entry_metadata.files = ['test/mainfile.txt']
        entry_metadata.apply_domain_metadata(normalized)
        entry_metadata.encyclopedia = normalized.section_metadata.encyclopedia

        entry_metadata.m_update(datasets=[example_dataset.dataset_id])

        entry_metadata.m_update(
            calc_id='1', uploader=test_user.user_id, published=True, with_embargo=False)
        entry_metadata.a_elastic.index(refresh=True)

        entry_metadata.m_update(
            calc_id='2', uploader=other_test_user.user_id, published=True,