Rework integration tests to use pytest

parent ea0b7d60
......@@ -15,28 +15,3 @@
#
# You should have received a copy of the GNU General Public License
# along with gracedb. If not, see <http://www.gnu.org/licenses/>.
import unittest
from .test_events import TestEvents
from .test_labels import TestLabels
from .test_superevents import TestSuperevents
from .test_voevents import VOEventTestSuite
# List of TestCase classes to import tests from
test_classes = [TestEvents, TestSuperevents, TestLabels]
# List of test suites to add
test_suites = [VOEventTestSuite()]
# Define suite and loader
IntegrationTestSuite = unittest.TestSuite()
loader = unittest.TestLoader()
# Get suites from classes
for test_class in test_classes:
suite = loader.loadTestsFromTestCase(test_class)
test_suites.append(suite)
# Add custom test suites
for suite in test_suites:
IntegrationTestSuite.addTest(suite)
# -*- coding: utf-8 -*-
# Copyright (C) Brian Moe, Branson Stephens (2015)
#
# This file is part of gracedb
#
# gracedb is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# It is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with gracedb. If not, see <http://www.gnu.org/licenses/>.
from __future__ import print_function
import unittest
import os
from ligo.gracedb.rest import GraceDb
class TestGraceDb(unittest.TestCase):
"""Base class for gracedb-client integration tests"""
@classmethod
def setUpClass(cls):
# Define useful variables
# Test service URL
TEST_SERVICE = os.environ.get(
'TEST_SERVICE',
'https://gracedb-test.ligo.org/api/'
)
# Data directory
cls.TEST_DATA_DIR = os.environ.get(
'TEST_DATA_DIR',
os.path.join(os.path.dirname(__file__), "data")
)
# Set up client
cls._gracedb = GraceDb(TEST_SERVICE)
print("Using service {0}".format(cls._gracedb._versioned_service_url))
......@@ -22,3 +22,51 @@ def test_data_dir():
os.path.join(os.path.dirname(__file__), 'data')
)
return d
@pytest.fixture
def create_event(client, test_data_dir):
def _inner(pipeline='gstlal', search='LowMass',
filename=os.path.join(test_data_dir, 'cbc-lm.xml')):
response = client.createEvent('Test', pipeline, filename,
search=search)
return response.json()
return _inner
@pytest.fixture
def create_superevent(client, test_data_dir, create_event):
def _inner(
pipeline='gstlal', search='LowMass',
filename=os.path.join(test_data_dir, 'cbc-lm.xml'),
t_start=1, t_0=2, t_end=3
):
event = create_event(pipeline=pipeline, search=search,
filename=filename)
response = client.createSuperevent(t_start, t_0, t_end,
event['graceid'], category='T')
return response.json()
return _inner
@pytest.fixture
def create_obj(client, test_data_dir, create_event, create_superevent):
def _inner(
obj_type='event', pipeline='gstlal', search='LowMass',
filename=os.path.join(test_data_dir, 'cbc-lm.xml'),
t_start=1, t_0=2, t_end=3
):
event = create_event(pipeline=pipeline, search=search,
filename=filename)
if obj_type == 'superevent':
response = client.createSuperevent(t_start, t_0, t_end,
event['graceid'], category='T')
obj = response.json()
obj_id = obj['superevent_id']
elif obj_type == 'event':
obj = event
obj_id = event['graceid']
else:
raise ValueError("obj_type must be 'event' or 'superevent'")
return obj, obj_id
return _inner
import pytest
# Set module-level marks
pytestmark = pytest.mark.integration
def test_ping(client):
response = client.ping()
assert response.status == 200
@pytest.mark.parametrize(
"resource",
['api_versions', 'server_version', 'links', 'templates', 'groups',
'pipelines', 'searches', 'allowed_labels', 'superevent_categories',
'em_groups', 'voevent_types', 'signoff_types', 'signoff_statuses',
'instruments']
)
def test_api_root_content(client, resource):
assert getattr(client, resource) is not None
import datetime
import pytest
# Apply module-level marks
pytestmark = pytest.mark.integration
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_emobservations(client, create_obj, obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Compile EM observation data
emgroup = client.em_groups[0]
ra_list = [1, 2, 3, 4]
ra_width_list = [0.5] * len(ra_list)
dec_list = [5, 6, 7, 8]
dec_width_list = [0.7] * len(dec_list)
now = datetime.datetime.utcnow()
start_time_list = list(
map(lambda i: (now + datetime.timedelta(seconds=i)).isoformat(),
[0, 1, 2, 3])
)
duration_list = [1] * len(start_time_list)
comment = "test comment"
# Create EM observation and check results
response = client.writeEMObservation(
obj_id, emgroup, ra_list, ra_width_list, dec_list, dec_width_list,
start_time_list, duration_list, comment=comment
)
assert response.status == 201
data = response.json()
assert data['comment'] == comment
assert data['group'] == emgroup
assert len(data['footprints']) == len(ra_list)
for emf in data['footprints']:
N = emf['N']
assert emf['ra'] == ra_list[N - 1]
assert emf['dec'] == dec_list[N - 1]
assert emf['raWidth'] == ra_width_list[N - 1]
assert emf['decWidth'] == dec_width_list[N - 1]
assert emf['exposure_time'] == duration_list[N - 1]
emo_N = data['N']
# Get list of emobservations and check results
response = client.emobservations(obj_id)
assert response.status == 200
data = response.json()
assert len(data['observations']) == 1
assert data['observations'][0]['N'] == emo_N
# Retrieve the individual emobservation directly
response = client.emobservations(obj_id, emo_N)
assert response.status == 200
data = response.json()
assert data['comment'] == comment
assert data['group'] == emgroup
assert len(data['footprints']) == len(ra_list)
for emf in data['footprints']:
N = emf['N']
assert emf['ra'] == ra_list[N - 1]
assert emf['dec'] == dec_list[N - 1]
assert emf['raWidth'] == ra_width_list[N - 1]
assert emf['decWidth'] == dec_width_list[N - 1]
assert emf['exposure_time'] == duration_list[N - 1]
import os
import pytest
# Apply module-level marks
pytestmark = pytest.mark.integration
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_file_upload_and_retrieval(client, test_data_dir, create_obj,
obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Create a log with a file upload
comment = 'test file upload'
test_file = os.path.join(test_data_dir, 'test_file.txt')
response = client.writeLog(obj_id, comment, filename=test_file)
assert response.status == 201
data = response.json()
assert data['comment'] == comment
assert data['filename'] == os.path.basename(test_file)
# Check file list for obj
response = client.files(obj_id)
assert response.status == 200
data = response.json()
assert len(data) >= 1
# Download the file and compare it
response = client.files(obj_id, os.path.basename(test_file))
assert response.status == 200
file_from_server = response.read()
with open(test_file, 'rb') as f:
file_from_disk = f.read()
assert file_from_server == file_from_disk
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_upload_file_multiple_versions(client, test_data_dir, create_obj,
obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Create a log with a file upload
file_name = 'new_file_test.txt'
comment = 'test file upload'
fc1 = 'file 1 data'
response = client.writeLog(obj_id, comment, filename=file_name,
filecontents=fc1)
assert response.status == 201
data = response.json()
assert data['comment'] == comment
assert data['filename'] == file_name
assert data['file_version'] == 0
# Create another log using the same file name
fc2 = 'file 2 data'
response = client.writeLog(obj_id, comment, filename=file_name,
filecontents=fc2)
assert response.status == 201
data = response.json()
assert data['comment'] == comment
assert data['filename'] == file_name
assert data['file_version'] == 1
# Check the attached files
response = client.files(obj_id)
assert response.status == 200
data = response.json()
# There should be at least the two files we uploaded and the symlink to
# the latest one.
assert len(data) >= 3
for version in ['', ',0', ',1']:
assert (file_name + version) in data
# Download all files and compare the contents
file1_data = client.files(obj_id, file_name + ',0').read()
file2_data = client.files(obj_id, file_name + ',1').read()
file_data = client.files(obj_id, file_name).read()
assert file1_data.decode() == fc1
assert file2_data.decode() == fc2
assert file2_data == file_data
assert file1_data != file2_data
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_upload_large_file(client, test_data_dir, create_obj, obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Upload a large file (2 MB)
filename = os.path.join(test_data_dir, 'big.data')
response = client.writeLog(obj_id, "Large file upload test",
filename=filename)
assert response.status == 201
data = response.json()
assert data['filename'] == os.path.basename(filename)
# Download file and compare to file on disk
# Download the file and compare it
response = client.files(obj_id, os.path.basename(filename))
assert response.status == 200
file_from_server = response.read()
with open(filename, 'rb') as f:
file_from_disk = f.read()
assert file_from_server == file_from_disk
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_upload_binary_file(client, test_data_dir, create_obj, obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Upload a binary file
filename = os.path.join(test_data_dir, 'upload.data.gz')
response = client.writeLog(obj_id, "Binary file upload test",
filename=filename)
assert response.status == 201
data = response.json()
assert data['filename'] == os.path.basename(filename)
# -*- coding: utf-8 -*-
# Copyright (C) Brian Moe, Branson Stephens (2015)
#
# This file is part of gracedb
#
# gracedb is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# It is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with gracedb. If not, see <http://www.gnu.org/licenses/>.
from __future__ import print_function
import unittest
import os
from .base import TestGraceDb
class TestLabels(TestGraceDb):
"""A suite for carefully testing event creation with labels"""
# Helper functions --------------------------------------------------------
def label_creation_helper(self, input_labels=None):
eventFile = os.path.join(self.TEST_DATA_DIR, "burst-cwb.txt")
if input_labels:
r = self._gracedb.createEvent(
"Test", "CWB", eventFile, labels=input_labels
)
else:
r = self._gracedb.createEvent("Test", "CWB", eventFile)
result = r.json()
gid = result['graceid']
# Make sure event was created properly
self.assertEqual(r.status, 201)
# Make sure there are no warnings
self.assertEqual(result['warnings'], [])
# Get event and make sure labels match what we expect
event_labels = self._gracedb.labels(gid).json()['labels']
if input_labels:
# Make sure that input_labels is a list for ease of comparison
if isinstance(input_labels, str):
input_labels = [input_labels]
event_label_names = [l['name'] for l in event_labels]
for l in input_labels:
self.assertIn(l, event_label_names)
else:
self.assertEqual(event_labels, [])
def label_bad_args_helper(self, input_labels, error_class):
eventFile = os.path.join(self.TEST_DATA_DIR, "burst-cwb.txt")
with self.assertRaises(error_class):
self._gracedb.createEvent(
"Test", "CWB", eventFile, labels=input_labels
)
# Tests -------------------------------------------------------------------
def test_create_event_with_label(self):
"""Create an event with a single label"""
self.label_creation_helper("DQV")
def test_create_event_with_labels(self):
"""Create an event with multiple labels"""
self.label_creation_helper(["DQV", "INJ", "PE_READY"])
def test_create_event_with_label_blank_string(self):
"""Create an event with labels=''"""
self.label_creation_helper('')
def test_create_event_with_label_empty_list(self):
"""Create an event with labels=[]"""
self.label_creation_helper([])
def test_create_event_no_labels_defined(self):
"""Try to create an event with no labels defined"""
self.label_creation_helper()
def test_create_event_label_list_empty_string(self):
"""Try to create an event with labels=['']"""
self.label_bad_args_helper([''], NameError)
def test_create_event_label_string_bad(self):
"""Try to create an event with labels='BADLABEL'"""
self.label_bad_args_helper('BADLABEL', NameError)
def test_create_event_label_list_bad(self):
"""Try to create an event with labels=['BADLABEL']"""
self.label_bad_args_helper(['BADLABEL'], NameError)
def test_create_event_label_list_onebad(self):
"""Try to create an event with labels=['H1OPS', 'BADLABEL']"""
self.label_bad_args_helper(['H1OPS', 'BADLABEL'], NameError)
def test_create_event_label_list_allbad(self):
"""Try to create an event with labels=['BAD1', 'BAD2']"""
self.label_bad_args_helper(['BAD1', 'BAD2'], NameError)
def test_create_event_label_int(self):
"""Try to create an event with labels=123"""
self.label_bad_args_helper(123, TypeError)
if __name__ == "__main__":
unittest.main(verbosity=2, failfast=True)
import pytest
# Apply module-level mark
pytestmark = pytest.mark.integration
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_labels(client, create_obj, obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Add a label
label1 = 'DQV'
response = client.writeLabel(obj_id, label1)
assert response.status == 201
data = response.json()
# NOTE: events API returns empty response - should be fixed
if obj_type == 'superevent':
assert data['name'] == label1
elif obj_type == 'event':
assert data == '{}'
# Add another label
label2 = 'INJ'
response = client.writeLabel(obj_id, label2)
assert response.status == 201
data = response.json()
# NOTE: events API returns empty response - should be fixed
if obj_type == 'superevent':
assert data['name'] == label2
elif obj_type == 'event':
assert data == '{}'
# Get list of labels
response = client.labels(obj_id)
assert response.status == 200
data = response.json()
obj_labels = [l['name'] for l in data['labels']]
assert len(obj_labels) == 2
assert label1 in obj_labels
assert label2 in obj_labels
# Pull down an individual label
response = client.labels(obj_id, label2)
assert response.status == 200
data = response.json()
assert data['name'] == label2
# Remove a label
response = client.removeLabel(obj_id, label1)
assert response.status == 204
data = response.json()
assert data == {}
# Get event and double check
if obj_type == 'event':
method = 'event'
elif obj_type == 'superevent':
method = 'superevent'
response = getattr(client, method)(obj_id)
assert response.status == 200
data = response.json()
assert len(data['labels']) == 1
assert label1 not in data['labels']
assert label2 in data['labels']
import pytest
# Apply module-level marks
pytestmark = pytest.mark.integration
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_logs(client, create_obj, obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Create a log
comment = 'test comment'
response = client.writeLog(obj_id, comment)
assert response.status == 201
data = response.json()
assert data['comment'] == comment
log_N = data['N']
# Pull down list of logs for the event
response = client.logs(obj_id)
assert response.status == 200
data = response.json()
assert len(data['log']) >= 1
# Pull down individual log
response = client.logs(obj_id, log_N)
assert response.status == 200
data = response.json()
assert data['comment'] == comment
assert data['N'] == log_N
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_log_creation_with_tags(client, create_obj, obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Create a log with tags
comment = 'test log with tags'
tags = ['test_tag1', 'test_tag2']
response = client.writeLog(obj_id, comment, tag_name=tags)
assert response.status == 201
data = response.json()
assert data['comment'] == comment
for t in tags:
assert t in data['tag_names']
@pytest.mark.parametrize("obj_type", ['event', 'superevent'])
def test_log_tag_and_untag(client, create_obj, obj_type):
# Create event or superevent
obj, obj_id = create_obj(obj_type)
# Create a log
comment = 'test log, add/remove tags later'
response = client.writeLog(obj_id, comment)
assert response.status == 201
data = response.json()
assert data['comment'] == comment
log_N = data['N']
# Add a tag
tag = 'test_tag'
response = client.addTag(obj_id, log_N, tag)
assert response.status == 201
# Get log and check tag status
response = client.logs(obj_id, log_N)
assert response.status == 200
data = response.json()
assert len(data['tag_names']) == 1
assert tag in data['tag_names']
# Remove tag
response = client.removeTag(obj_id, log_N, tag)
assert response.status == 204
# Get log and check tag status
response = client.logs(obj_id, log_N)
assert response.status == 200
data = response.json()
assert len(data['tag_names']) == 0
......@@ -2,8 +2,10 @@ import os
import pytest
# Apply module-level marks
pytestmark = pytest.mark.integration
<