-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_links.py
224 lines (194 loc) · 8.2 KB
/
test_links.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (C) 2015 University of Dundee & Open Microscopy Environment.
# All rights reserved.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Tests creation and deletion of links between e.g. Projects & Datasets etc.
"""
import omero
from omero.rtypes import rstring
from omeroweb.testlib import IWebTest
from omeroweb.testlib import post_json, get_json, delete_json
from time import sleep
from django.core.urlresolvers import reverse
import pytest
class TestLinks(IWebTest):
"""
Tests creation and deletion of links between
e.g. Projects & Datasets etc.
"""
@pytest.fixture
def project(self):
"""Returns a new OMERO Project with required fields set."""
project = omero.model.ProjectI()
project.name = rstring(self.uuid())
return self.update.saveAndReturnObject(project)
@pytest.fixture
def dataset(self):
"""Returns a new OMERO Dataset with required fields set."""
dataset = omero.model.DatasetI()
dataset.name = rstring(self.uuid())
return self.update.saveAndReturnObject(dataset)
@pytest.fixture
def datasets(self):
"""Returns 2 new OMERO Datasets with required fields set."""
dataset = omero.model.DatasetI()
dataset.name = rstring("A_%s" % self.uuid())
dataset2 = omero.model.DatasetI()
dataset2.name = rstring("B_%s" % self.uuid())
return self.update.saveAndReturnArray([dataset, dataset2])
@pytest.fixture
def images(self):
image = self.new_image(name="A_%s" % self.uuid())
image2 = self.new_image(name="B_%s" % self.uuid())
return self.update.saveAndReturnArray([image, image2])
@pytest.fixture
def screens(self):
"""Returns 2 new OMERO Screens with required fields set."""
screen = omero.model.ScreenI()
screen.name = rstring("A_%s" % self.uuid())
screen2 = omero.model.ScreenI()
screen2.name = rstring("B_%s" % self.uuid())
return self.update.saveAndReturnArray([screen, screen2])
@pytest.fixture
def plates(self):
"""Returns 2 new OMERO Plates with required fields set."""
plate = omero.model.PlateI()
plate.name = rstring("A_%s" % self.uuid())
plate2 = omero.model.PlateI()
plate2.name = rstring("B_%s" % self.uuid())
return self.update.saveAndReturnArray([plate, plate2])
def test_link_project_datasets(self, project, datasets):
# Link Project to Datasets
request_url = reverse("api_links")
pid = project.id.val
dids = [d.id.val for d in datasets]
data = {
'project': {pid: {'dataset': dids}}
}
rsp = post_json(self.django_client, request_url, data)
assert rsp == {"success": True}
# Check links
request_url = reverse("api_datasets")
rsp = get_json(self.django_client, request_url, {'id': pid})
# Expect a single Dataset with correct id
assert len(rsp['datasets']) == 2
assert rsp['datasets'][0]['id'] == dids[0]
assert rsp['datasets'][1]['id'] == dids[1]
def test_link_datasets_images(self, datasets, images):
# Link Datasets to Images
request_url = reverse("api_links")
dids = [d.id.val for d in datasets]
iids = [i.id.val for i in images]
# Link first dataset to first image,
# Second dataset linked to both images
data = {
'dataset': {dids[0]: {'image': [iids[0]]},
dids[1]: {'image': iids}}
}
rsp = post_json(self.django_client, request_url, data)
assert rsp == {"success": True}
# Check links
images_url = reverse("api_images")
# First Dataset has single image
rsp = get_json(self.django_client, images_url, {'id': dids[0]})
assert len(rsp['images']) == 1
assert rsp['images'][0]['id'] == iids[0]
# Second Dataset has both images
rsp = get_json(self.django_client, images_url, {'id': dids[1]})
assert len(rsp['images']) == 2
assert rsp['images'][0]['id'] == iids[0]
assert rsp['images'][1]['id'] == iids[1]
# Link BOTH images to first Dataset. Shouldn't get any
# Validation Exception, even though first image is already linked
data = {
'dataset': {dids[0]: {'image': iids}}
}
rsp = post_json(self.django_client, request_url, data)
assert rsp == {"success": True}
# Check first Dataset now has both images
rsp = get_json(self.django_client, images_url, {'id': dids[0]})
assert len(rsp['images']) == 2
assert rsp['images'][0]['id'] == iids[0]
assert rsp['images'][1]['id'] == iids[1]
def test_link_unlink_tagset_tags(self):
"""
Tests linking of tagset to tag, then unlinking
"""
tag = self.make_tag()
tagset = self.make_tag(ns=omero.constants.metadata.NSINSIGHTTAGSET)
tagId = tag.id.val
tagsetId = tagset.id.val
links_url = reverse("api_links")
# Link tagset to tag
data = {
'tagset': {tagsetId: {'tag': [tagId]}}
}
rsp = post_json(self.django_client, links_url, data)
assert rsp == {"success": True}
# Check that tag is listed under tagset...
tags_url = reverse("api_tags_and_tagged")
r = get_json(self.django_client, tags_url, {'id': tagsetId})
assert len(r['tags']) == 1
assert r['tags'][0]['id'] == tagId
# Unlink first Tag from Tagset
# data {} is same as for creating link above
response = delete_json(self.django_client, links_url, data)
assert response["success"]
# Since the Delete is ansync - need to check repeatedly for deletion
for i in range(10):
rsp = get_json(self.django_client, tags_url, {'id': tagsetId})
if len(rsp['tags']) == 0:
break
sleep(0.5)
# Check that link has been deleted
assert len(rsp['tags']) == 0
def test_unlink_screen_plate(self, screens, plates):
# Link both plates to both screens
request_url = reverse("api_links")
sids = [s.id.val for s in screens]
pids = [p.id.val for p in plates]
data = {
'screen': {sids[0]: {'plate': pids},
sids[1]: {'plate': pids}}
}
rsp = post_json(self.django_client, request_url, data)
assert rsp == {"success": True}
# Confirm that first Screen linked to 2 Plates
plates_url = reverse("api_plates")
rsp = get_json(self.django_client, plates_url, {'id': sids[0]})
assert len(rsp['plates']) == 2
# Unlink first Plate from first Screen
request_url = reverse("api_links")
data = {
'screen': {sids[0]: {'plate': pids[:1]}}
}
response = delete_json(self.django_client, request_url, data)
# Returns remaining link from 2nd Screen to first Plate
assert response == {"success": True,
"screen": {str(sids[1]): {"plate": pids[:1]}}
}
# Since the Delete is ansync - need to check repeatedly for deletion
# by counting plates under screen...
plates_url = reverse("api_plates")
for i in range(10):
rsp = get_json(self.django_client, plates_url, {'id': sids[0]})
if len(rsp['plates']) == 1:
break
sleep(0.5)
# Check that link has been deleted, leaving 2nd plate under 1st screen
assert len(rsp['plates']) == 1
assert rsp['plates'][0]['id'] == pids[1]