0001-backoffice-add-storage-UI-to-store-assign-roles-to-A.patch
| tests/admin_pages/test_api_access.py | ||
|---|---|---|
| 161 | 161 |
resp = resp.form.submit('submit')
|
| 162 | 162 |
assert resp.location == 'http://example.net/backoffice/settings/api-access/' |
| 163 | 163 |
assert ApiAccess.count() == 0 |
| 164 | ||
| 165 | ||
| 166 |
def test_api_access_roles(pub, api_access): |
|
| 167 |
create_superuser(pub) |
|
| 168 | ||
| 169 |
pub.role_class.wipe() |
|
| 170 |
role_a = pub.role_class(name='a') |
|
| 171 |
role_a.store() |
|
| 172 |
role_b = pub.role_class(name='b') |
|
| 173 |
role_b.store() |
|
| 174 | ||
| 175 |
app = login(get_app(pub)) |
|
| 176 | ||
| 177 |
resp = app.get('/backoffice/settings/api-access/1/')
|
|
| 178 |
resp = resp.click(href='edit') |
|
| 179 |
resp.form['roles$element0'] = role_a.id |
|
| 180 |
resp = resp.form.submit('roles$add_element')
|
|
| 181 |
resp.form['roles$element1'] = role_b.id |
|
| 182 |
resp = resp.form.submit('submit')
|
|
| 183 | ||
| 184 |
api_access = ApiAccess.get(api_access.id) |
|
| 185 |
assert set(x.id for x in api_access.get_roles()) == {role_a.id, role_b.id}
|
|
| wcs/admin/api_access.py | ||
|---|---|---|
| 16 | 16 | |
| 17 | 17 |
import uuid |
| 18 | 18 | |
| 19 |
from quixote import get_response, redirect |
|
| 19 |
from quixote import get_publisher, get_response, redirect
|
|
| 20 | 20 |
from quixote.directory import Directory |
| 21 | 21 |
from quixote.html import TemplateIO, htmltext |
| 22 | 22 | |
| 23 | 23 |
from wcs.api_access import ApiAccess |
| 24 | 24 |
from wcs.qommon import _, errors, template |
| 25 | 25 |
from wcs.qommon.backoffice.menu import html_top |
| 26 |
from wcs.qommon.form import CheckboxWidget, Form, HtmlWidget, StringWidget, TextWidget |
|
| 26 |
from wcs.qommon.form import ( |
|
| 27 |
CheckboxWidget, |
|
| 28 |
Form, |
|
| 29 |
HtmlWidget, |
|
| 30 |
SingleSelectWidget, |
|
| 31 |
StringWidget, |
|
| 32 |
TextWidget, |
|
| 33 |
WidgetList, |
|
| 34 |
) |
|
| 27 | 35 | |
| 28 | 36 | |
| 29 | 37 |
class ApiAccessUI: |
| ... | ... | |
| 65 | 73 |
title=_('Restrict to anonymised data'),
|
| 66 | 74 |
value=self.api_access.restrict_to_anonymised_data, |
| 67 | 75 |
) |
| 76 |
roles = list(get_publisher().role_class.select(order_by='name')) |
|
| 77 |
form.add( |
|
| 78 |
WidgetList, |
|
| 79 |
'roles', |
|
| 80 |
title=_('Roles'),
|
|
| 81 |
element_type=SingleSelectWidget, |
|
| 82 |
value=self.api_access.roles, |
|
| 83 |
add_element_label=_('Add Role'),
|
|
| 84 |
element_kwargs={
|
|
| 85 |
'render_br': False, |
|
| 86 |
'options': [(None, '---', None)] + [(x, x.name, x.id) for x in roles if not x.is_internal()], |
|
| 87 |
}, |
|
| 88 |
hint=_('Apply access control related to these roles'),
|
|
| 89 |
) |
|
| 90 | ||
| 68 | 91 |
if not self.api_access.is_readonly(): |
| 69 | 92 |
form.add_submit('submit', _('Submit'))
|
| 70 | 93 |
form.add_submit('cancel', _('Cancel'))
|
| ... | ... | |
| 86 | 109 | |
| 87 | 110 |
self.api_access.name = name |
| 88 | 111 |
self.api_access.access_identifier = access_identifier |
| 89 |
for attribute in ('description', 'access_key', 'restrict_to_anonymised_data'):
|
|
| 112 |
for attribute in ('description', 'access_key', 'restrict_to_anonymised_data', 'roles'):
|
|
| 90 | 113 |
setattr(self.api_access, attribute, form.get_widget(attribute).parse()) |
| 91 | 114 |
self.api_access.store() |
| 92 | 115 | |
| wcs/api_access.py | ||
|---|---|---|
| 14 | 14 |
# You should have received a copy of the GNU General Public License |
| 15 | 15 |
# along with this program; if not, see <http://www.gnu.org/licenses/>. |
| 16 | 16 | |
| 17 |
import xml.etree.ElementTree as ET |
|
| 18 | ||
| 19 |
from quixote import get_publisher |
|
| 20 | ||
| 21 |
from wcs.qommon.misc import xml_node_text |
|
| 22 |
from wcs.qommon.storage import Equal, Or |
|
| 17 | 23 |
from wcs.qommon.xml_storage import XmlStorableObject |
| 18 | 24 | |
| 19 | 25 | |
| ... | ... | |
| 25 | 31 |
access_key = None |
| 26 | 32 |
description = None |
| 27 | 33 |
restrict_to_anonymised_data = False |
| 34 |
roles = None |
|
| 28 | 35 | |
| 29 | 36 |
# declarations for serialization |
| 30 | 37 |
XML_NODES = [ |
| ... | ... | |
| 33 | 40 |
('access_identifier', 'str'),
|
| 34 | 41 |
('access_key', 'str'),
|
| 35 | 42 |
('restrict_to_anonymised_data', 'bool'),
|
| 43 |
('roles', 'roles'),
|
|
| 36 | 44 |
] |
| 37 | 45 | |
| 38 | 46 |
@classmethod |
| ... | ... | |
| 48 | 56 |
if api_access: |
| 49 | 57 |
return api_access.access_key |
| 50 | 58 |
return None |
| 59 | ||
| 60 |
def get_roles(self): |
|
| 61 |
return self.roles or [] |
|
| 62 | ||
| 63 |
def export_roles_to_xml(self, element, attribute_name, include_id=False, **kwargs): |
|
| 64 |
for role in self.get_roles(): |
|
| 65 |
sub = ET.SubElement(element, 'role') |
|
| 66 |
if include_id: |
|
| 67 |
sub.attrib['role-id'] = role.id |
|
| 68 |
sub.attrib['role-slug'] = role.slug |
|
| 69 |
sub.text = role.name |
|
| 70 | ||
| 71 |
def import_roles_from_xml(self, element, include_id=False, **kwargs): |
|
| 72 |
criterias = [] |
|
| 73 |
for sub in element: |
|
| 74 |
if include_id and 'role-id' in sub.attrib: |
|
| 75 |
criterias.append(Equal('id', sub.attrib['role-id']))
|
|
| 76 |
elif 'role-slug' in sub.attrib: |
|
| 77 |
criterias.append(Equal('slug', sub.attrib['role-slug']))
|
|
| 78 |
else: |
|
| 79 |
role_name = xml_node_text(sub) |
|
| 80 |
if role_name: |
|
| 81 |
criterias.append(Equal('name', role_name))
|
|
| 82 |
return get_publisher().role_class.select([Or(criterias)], order_by='name') |
|
| wcs/data_sources.py | ||
|---|---|---|
| 523 | 523 |
section = 'settings' |
| 524 | 524 |
return '%s/%s/data-sources/%s/' % (base_url, section, self.id) |
| 525 | 525 | |
| 526 |
def export_data_source_to_xml(self, element, attribute_name, charset): |
|
| 526 |
def export_data_source_to_xml(self, element, attribute_name, charset, **kwargs):
|
|
| 527 | 527 |
data_source = getattr(self, attribute_name) |
| 528 | 528 |
ET.SubElement(element, 'type').text = data_source.get('type')
|
| 529 | 529 |
ET.SubElement(element, 'value').text = force_text(data_source.get('value') or '', charset)
|
| 530 | 530 | |
| 531 |
def import_data_source_from_xml(self, element, charset):
|
|
| 531 |
def import_data_source_from_xml(self, element, **kwargs):
|
|
| 532 | 532 |
return {
|
| 533 | 533 |
'type': force_str(element.find('type').text),
|
| 534 | 534 |
'value': force_str(element.find('value').text or ''),
|
| wcs/qommon/xml_storage.py | ||
|---|---|---|
| 50 | 50 |
if not getattr(self, attribute_name, None): |
| 51 | 51 |
continue |
| 52 | 52 |
element = ET.SubElement(root, attribute_name) |
| 53 |
getattr(self, 'export_%s_to_xml' % attribute_type)(element, attribute_name, charset=charset) |
|
| 53 |
export_method = getattr(self, 'export_%s_to_xml' % attribute_type) |
|
| 54 |
export_method(element, attribute_name, charset=charset, include_id=include_id) |
|
| 54 | 55 |
return root |
| 55 | 56 | |
| 56 |
def export_str_to_xml(self, element, attribute_name, charset): |
|
| 57 |
def export_str_to_xml(self, element, attribute_name, charset, **kwargs):
|
|
| 57 | 58 |
element.text = force_text(getattr(self, attribute_name), charset) |
| 58 | 59 | |
| 59 |
def export_int_to_xml(self, element, attribute_name, charset):
|
|
| 60 |
def export_int_to_xml(self, element, attribute_name, **kwargs):
|
|
| 60 | 61 |
element.text = str(getattr(self, attribute_name)) |
| 61 | 62 | |
| 62 |
def export_bool_to_xml(self, element, attribute_name, charset):
|
|
| 63 |
def export_bool_to_xml(self, element, attribute_name, **kwargs):
|
|
| 63 | 64 |
element.text = 'true' if getattr(self, attribute_name) else 'false' |
| 64 | 65 | |
| 65 |
def export_datetime_to_xml(self, element, attribute_name, charset):
|
|
| 66 |
def export_datetime_to_xml(self, element, attribute_name, **kwargs):
|
|
| 66 | 67 |
element.text = getattr(self, attribute_name).isoformat() |
| 67 | 68 | |
| 68 |
def export_str_list_to_xml(self, element, attribute_name, charset):
|
|
| 69 |
def export_str_list_to_xml(self, element, attribute_name, **kwargs):
|
|
| 69 | 70 |
for item in getattr(self, attribute_name, None) or []: |
| 70 | 71 |
ET.SubElement(element, 'item').text = item |
| 71 | 72 | |
| ... | ... | |
| 101 | 102 |
element = tree.find(attribute_name) |
| 102 | 103 |
if element is None: |
| 103 | 104 |
continue |
| 105 |
import_method = getattr(obj, 'import_%s_from_xml' % attribute_type) |
|
| 104 | 106 |
setattr( |
| 105 | 107 |
obj, |
| 106 | 108 |
attribute_name, |
| 107 |
getattr(obj, 'import_%s_from_xml' % attribute_type)(element, charset=charset),
|
|
| 109 |
import_method(element, charset=charset, include_id=include_id),
|
|
| 108 | 110 |
) |
| 109 | 111 |
return obj |
| 110 | 112 | |
| 111 |
def import_str_from_xml(self, element, charset):
|
|
| 113 |
def import_str_from_xml(self, element, **kwargs):
|
|
| 112 | 114 |
return xml_node_text(element) |
| 113 | 115 | |
| 114 |
def import_int_from_xml(self, element, charset):
|
|
| 116 |
def import_int_from_xml(self, element, **kwargs):
|
|
| 115 | 117 |
return int(element.text) |
| 116 | 118 | |
| 117 |
def import_bool_from_xml(self, element, charset):
|
|
| 119 |
def import_bool_from_xml(self, element, **kwargs):
|
|
| 118 | 120 |
return bool(element.text == 'true') |
| 119 | 121 | |
| 120 |
def import_datetime_from_xml(self, element, charset):
|
|
| 122 |
def import_datetime_from_xml(self, element, **kwargs):
|
|
| 121 | 123 |
return datetime.datetime.strptime(element.text[:19], '%Y-%m-%dT%H:%M:%S') |
| 122 | 124 | |
| 123 |
def import_str_list_from_xml(self, element, charset):
|
|
| 125 |
def import_str_list_from_xml(self, element, **kwargs):
|
|
| 124 | 126 |
value = [] |
| 125 | 127 |
for item in element.findall('item'):
|
| 126 | 128 |
value.append(item.text) |
| wcs/roles.py | ||
|---|---|---|
| 44 | 44 |
StorableObject.__init__(self, id=id) |
| 45 | 45 |
self.name = name |
| 46 | 46 | |
| 47 |
def __eq__(self, other): |
|
| 48 |
return bool(self.__class__ is other.__class__ and self.id == other.id) |
|
| 49 | ||
| 47 | 50 |
def migrate(self): |
| 48 | 51 |
changed = False |
| 49 | 52 |
if not self.slug: |
| wcs/templates/wcs/backoffice/api_access.html | ||
|---|---|---|
| 19 | 19 |
<li>{% trans "Access identifier:" %} {{ api_access.access_identifier }}</li>
|
| 20 | 20 |
<li>{% trans "Access key:" %} {{ api_access.access_key }}</li>
|
| 21 | 21 |
{% if api_access.restrict_to_anonymised_data %}<li>{% trans "Restricted to anonymised data" %}</li>{% endif %}
|
| 22 |
{% if api_access.get_roles %}
|
|
| 23 |
<li>{% trans "Roles:" %}
|
|
| 24 |
<ul> |
|
| 25 |
{% for role in api_access.get_roles %}<li>{{ role.name }}</li>{% endfor %}
|
|
| 26 |
</ul> |
|
| 27 |
</li> |
|
| 28 |
{% endif %}
|
|
| 22 | 29 |
</ul> |
| 23 | 30 |
</div> |
| 24 | 31 |
{% endblock %}
|
| wcs/wscalls.py | ||
|---|---|---|
| 201 | 201 |
base_url = get_publisher().get_backoffice_url() |
| 202 | 202 |
return '%s/settings/wscalls/%s/' % (base_url, self.slug) |
| 203 | 203 | |
| 204 |
def export_request_to_xml(self, element, attribute_name, charset): |
|
| 204 |
def export_request_to_xml(self, element, attribute_name, charset, **kwargs):
|
|
| 205 | 205 |
request = getattr(self, attribute_name) |
| 206 | 206 |
for attr in ('url', 'request_signature_key', 'method'):
|
| 207 | 207 |
ET.SubElement(element, attr).text = force_text(request.get(attr) or '', charset) |
| ... | ... | |
| 214 | 214 |
if request.get('post_formdata'):
|
| 215 | 215 |
ET.SubElement(element, 'post_formdata') |
| 216 | 216 | |
| 217 |
def import_request_from_xml(self, element, charset):
|
|
| 217 |
def import_request_from_xml(self, element, **kwargs):
|
|
| 218 | 218 |
request = {}
|
| 219 | 219 |
for attr in ('url', 'request_signature_key', 'method'):
|
| 220 | 220 |
request[attr] = '' |
| 221 |
- |
|