0001-workflows-let-anonymous-signed-calls-trigger-jumps-9.patch
| tests/test_api.py | ||
|---|---|---|
| 18 | 18 |
from wcs.categories import Category |
| 19 | 19 |
from wcs.data_sources import NamedDataSource |
| 20 | 20 |
from wcs.workflows import Workflow |
| 21 |
from wcs.wf.jump import JumpWorkflowStatusItem |
|
| 21 | 22 |
from wcs import fields |
| 22 | 23 |
from wcs.api import sign_url |
| 23 | 24 | |
| ... | ... | |
| 780 | 781 |
assert len(resp.json['data']) == 1 |
| 781 | 782 |
resp = get_app(pub).get(sign_uri('/api/users/?q=foobar'))
|
| 782 | 783 |
assert len(resp.json['data']) == 0 |
| 784 | ||
| 785 |
def test_workflow_trigger(pub, local_user): |
|
| 786 |
workflow = Workflow(name='test') |
|
| 787 |
st1 = workflow.add_status('Status1', 'st1')
|
|
| 788 |
jump = JumpWorkflowStatusItem() |
|
| 789 |
jump.trigger = 'XXX' |
|
| 790 |
jump.status = 'st2' |
|
| 791 |
st1.items.append(jump) |
|
| 792 |
jump.parent = st1 |
|
| 793 |
st2 = workflow.add_status('Status2', 'st2')
|
|
| 794 |
workflow.store() |
|
| 795 | ||
| 796 |
FormDef.wipe() |
|
| 797 |
formdef = FormDef() |
|
| 798 |
formdef.name = 'test' |
|
| 799 |
formdef.fields = [] |
|
| 800 |
formdef.workflow_id = workflow.id |
|
| 801 |
formdef.store() |
|
| 802 | ||
| 803 |
formdef.data_class().wipe() |
|
| 804 |
formdata = formdef.data_class()() |
|
| 805 |
formdata.just_created() |
|
| 806 |
formdata.store() |
|
| 807 |
assert formdef.data_class().get(formdata.id).status == 'wf-st1' |
|
| 808 | ||
| 809 |
resp = get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), |
|
| 810 |
status=200) |
|
| 811 |
assert formdef.data_class().get(formdata.id).status == 'wf-st2' |
|
| 812 | ||
| 813 |
Role.wipe() |
|
| 814 |
role = Role(name='xxx') |
|
| 815 |
role.store() |
|
| 816 | ||
| 817 |
jump.by = [role.id] |
|
| 818 |
workflow.store() |
|
| 819 | ||
| 820 |
formdata.store() # (will get back to wf-st1) |
|
| 821 |
resp = get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), |
|
| 822 |
status=403) |
|
| tests/test_form_pages.py | ||
|---|---|---|
| 2164 | 2164 |
resp = get_app(pub).get('/test/')
|
| 2165 | 2165 |
assert ').autocomplete({' in resp.body
|
| 2166 | 2166 |
assert 'http://example.net' in resp.body |
| 2167 | ||
| 2168 |
def test_form_workflow_trigger(pub): |
|
| 2169 |
user = create_user(pub) |
|
| 2170 | ||
| 2171 |
formdef = create_formdef() |
|
| 2172 |
formdef.fields = [] |
|
| 2173 |
formdef.store() |
|
| 2174 | ||
| 2175 |
workflow = Workflow(name='test') |
|
| 2176 |
st1 = workflow.add_status('Status1', 'st1')
|
|
| 2177 |
jump = JumpWorkflowStatusItem() |
|
| 2178 |
jump.trigger = 'XXX' |
|
| 2179 |
jump.status = 'st2' |
|
| 2180 |
st1.items.append(jump) |
|
| 2181 |
jump.parent = st1 |
|
| 2182 |
st2 = workflow.add_status('Status2', 'st2')
|
|
| 2183 |
workflow.store() |
|
| 2184 | ||
| 2185 |
formdef.workflow_id = workflow.id |
|
| 2186 |
formdef.store() |
|
| 2187 | ||
| 2188 |
formdef.data_class().wipe() |
|
| 2189 |
formdata = formdef.data_class()() |
|
| 2190 |
formdata.just_created() |
|
| 2191 |
formdata.store() |
|
| 2192 |
assert formdef.data_class().get(formdata.id).status == 'wf-st1' |
|
| 2193 | ||
| 2194 |
app = get_app(pub) |
|
| 2195 |
resp = login(app, username='foo', password='foo').get('/')
|
|
| 2196 |
resp = app.post(formdata.get_url() + 'jump/trigger/XXX', status=403) |
|
| 2197 | ||
| 2198 |
Role.wipe() |
|
| 2199 |
role = Role(name='xxx') |
|
| 2200 |
role.store() |
|
| 2201 | ||
| 2202 |
jump.by = [role.id] |
|
| 2203 |
workflow.store() |
|
| 2204 |
resp = app.post(formdata.get_url() + 'jump/trigger/XXX', status=403) |
|
| 2205 | ||
| 2206 |
user.roles = [role.id] |
|
| 2207 |
user.store() |
|
| 2208 |
resp = app.post(formdata.get_url() + 'jump/trigger/XXX', status=302) |
|
| 2209 | ||
| 2210 |
assert formdef.data_class().get(formdata.id).status == 'wf-st2' |
|
| wcs/wf/jump.py | ||
|---|---|---|
| 27 | 27 |
from qommon.cron import CronJob |
| 28 | 28 | |
| 29 | 29 |
from wcs.workflows import Workflow, WorkflowStatusJumpItem, register_item_class |
| 30 |
from wcs.api import get_user_from_api_query_string |
|
| 30 |
from wcs.api import get_user_from_api_query_string, is_url_signed
|
|
| 31 | 31 | |
| 32 | 32 |
def jump_and_perform(formdata, status, workflow_data=None): |
| 33 | 33 |
if workflow_data: |
| ... | ... | |
| 62 | 62 |
if not get_request().get_method() == 'POST': |
| 63 | 63 |
raise errors.AccessForbiddenError() |
| 64 | 64 | |
| 65 |
signed_request = is_url_signed() |
|
| 65 | 66 |
user = get_user_from_api_query_string() or get_request().user |
| 66 |
if not user: |
|
| 67 |
raise errors.AccessForbiddenError() |
|
| 68 | 67 |
for item in self.wfstatus.items: |
| 69 | 68 |
if not isinstance(item, JumpWorkflowStatusItem): |
| 70 | 69 |
continue |
| 71 | 70 |
if not hasattr(item, 'trigger'): |
| 72 | 71 |
continue |
| 73 | 72 |
if component == item.trigger: |
| 74 |
if not item.check_auth(self.formdata, user): |
|
| 73 |
if signed_request and not item.by: |
|
| 74 |
pass |
|
| 75 |
elif not item.check_auth(self.formdata, user): |
|
| 75 | 76 |
raise errors.AccessForbiddenError() |
| 76 | 77 |
get_request().trigger_name = component |
| 77 | 78 |
workflow_data = None |
| 78 |
- |
|