-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_api_errors.py
197 lines (176 loc) · 8.53 KB
/
test_api_errors.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
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (C) 2016 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 querying & editing Projects with webgateway json api."""
from omeroweb.testlib import IWebTest, post_json, put_json, get_json
from django.core.urlresolvers import reverse
from omeroweb.api import api_settings
import pytest
from test_api_projects import get_connection
from omero.model import ProjectI, TagAnnotationI
from omero.rtypes import rstring
from omero_marshal import get_encoder, get_decoder, OME_SCHEMA_URL
from omero import ValidationException
class TestErrors(IWebTest):
"""Tests the response status with various error types."""
# Create a read-annotate group
@pytest.fixture(scope='function')
def group_A(self):
"""Return a new read-only group."""
return self.new_group(perms='rwra--')
# Create a read-only group
@pytest.fixture(scope='function')
def group_B(self):
"""Return a new read-only group."""
return self.new_group(perms='rwr---')
# Create users in the read-only group
@pytest.fixture()
def user_A(self, group_A, group_B):
"""Return a new user in the group_A group and also add to group_B."""
user = self.new_client_and_user(group=group_A)
self.add_groups(user[1], [group_B])
return user
def test_save_post_no_id(self):
"""If POST to /save/ data shouldn't contain @id."""
django_client = self.django_root_client
version = api_settings.API_VERSIONS[-1]
save_url = reverse('api_save', kwargs={'api_version': version})
payload = {'Name': 'test_save_post_no_id',
'@type': '%s#Project' % OME_SCHEMA_URL,
'@id': 1}
rsp = post_json(django_client, save_url, payload, status_code=400)
assert (rsp['message'] ==
"Object has '@id' attribute. Use PUT to update objects")
def test_save_put_id(self):
"""If PUT to /save/ to update, data must contain @id."""
django_client = self.django_root_client
version = api_settings.API_VERSIONS[-1]
save_url = reverse('api_save', kwargs={'api_version': version})
payload = {'Name': 'test_save_put_id',
'@type': '%s#Project' % OME_SCHEMA_URL}
rsp = put_json(django_client, save_url, payload, status_code=400)
assert (rsp['message'] ==
"No '@id' attribute. Use POST to create new objects")
def test_marshal_type(self):
"""If no decoder found for @type, get suitable message."""
django_client = self.django_root_client
version = api_settings.API_VERSIONS[-1]
save_url = reverse('api_save', kwargs={'api_version': version})
objType = 'SomeInvalidSchema#Project'
payload = {'Name': 'test_marshal_type',
'@type': objType}
rsp = post_json(django_client, save_url, payload, status_code=400)
assert (rsp['message'] ==
'No decoder found for type: %s' % objType)
def test_invalid_parameter(self):
"""Test that invalid query parameter gives suitable message."""
django_client = self.django_root_client
version = api_settings.API_VERSIONS[-1]
projects_url = reverse('api_projects', kwargs={'api_version': version})
payload = {'limit': 'foo'}
rsp = get_json(django_client, projects_url, payload, status_code=400)
assert (rsp['message'] ==
"invalid literal for int() with base 10: 'foo'")
def test_marshal_validation(self):
"""Test that we get expected error with invalid @type in json."""
django_client = self.django_root_client
version = api_settings.API_VERSIONS[-1]
save_url = reverse('api_save', kwargs={'api_version': version})
payload = {'Name': 'test_marshal_validation',
'@type': OME_SCHEMA_URL + '#Project',
'omero:details': {'@type': 'foo'}}
rsp = post_json(django_client, save_url, payload, status_code=400)
assert (rsp['message'] ==
"Error in decode of json data by omero_marshal")
assert rsp['stacktrace'].startswith(
'Traceback (most recent call last):')
def test_security_violation(self, group_B, user_A):
"""Test saving to incorrect group."""
conn = get_connection(user_A)
group_A_id = conn.getEventContext().groupId
userName = conn.getUser().getName()
django_client = self.new_django_client(userName, userName)
version = api_settings.API_VERSIONS[-1]
group_B_id = group_B.id.val
save_url = reverse('api_save', kwargs={'api_version': version})
# Create project in group_A (default group)
payload = {'Name': 'test_security_violation',
'@type': OME_SCHEMA_URL + '#Project'}
save_url_grp_A = save_url + '?group=' + str(group_A_id)
rsp = post_json(django_client, save_url_grp_A, payload,
status_code=201)
pr_json = rsp['data']
projectId = pr_json['@id']
# Try to save again into group B
save_url_grp_B = save_url + '?group=' + str(group_B_id)
rsp = put_json(django_client, save_url_grp_B, pr_json, status_code=403)
assert 'message' in rsp
msg = "Cannot read ome.model.containers.Project:Id_%s" % projectId
assert msg in rsp['message']
assert rsp['stacktrace'].startswith(
'Traceback (most recent call last):')
def test_validation_exception(self, user_A):
"""Test handling when we try to save something invalid."""
conn = get_connection(user_A)
group = conn.getEventContext().groupId
userName = conn.getUser().getName()
django_client = self.new_django_client(userName, userName)
version = api_settings.API_VERSIONS[-1]
save_url = reverse('api_save', kwargs={'api_version': version})
save_url += '?group=' + str(group)
# Create Tag
tag = TagAnnotationI()
tag.textValue = rstring('test_tag')
tag = conn.getUpdateService().saveAndReturnObject(tag)
tag_json = {'Value': 'test_tag',
'@id': tag.id.val,
'@type': OME_SCHEMA_URL + '#TagAnnotation'}
# Add Tag twice to Project to get Validation Exception
payload = {'Name': 'test_validation',
'@type': OME_SCHEMA_URL + '#Project',
'Annotations': [tag_json, tag_json]}
rsp = post_json(django_client, save_url, payload, status_code=400)
# NB: message contains whole stack trace
assert "ValidationException" in rsp['message']
assert rsp['stacktrace'].startswith(
'Traceback (most recent call last):')
def test_project_validation(self, user_A):
"""
Test to demonstrate details bug on encode->decode.
Test illustrates the ValidationException we see when
Project is encoded to dict then decoded back to Project
and saved.
No exception is seen if the original Project is simply
saved without encode & decode OR if the details are unloaded
before saving
"""
conn = get_connection(user_A)
project = ProjectI()
project.name = rstring('test_project_validation')
project = conn.getUpdateService().saveAndReturnObject(project)
# Saving original Project again is OK
conn.getUpdateService().saveObject(project)
# encode and decode before Save raises Validation Exception
project_json = get_encoder(project.__class__).encode(project)
decoder = get_decoder(project_json['@type'])
p = decoder.decode(project_json)
with pytest.raises(ValidationException):
conn.getUpdateService().saveObject(p)
p = decoder.decode(project_json)
# Unloading details allows Save without exception
p.unloadDetails()
conn.getUpdateService().saveObject(p)