0002-workflows-hide-attachment-not-matching-target-roles-.patch

     import re
     import time
     import xml.etree.ElementTree as ET
     import zipfile
     import pytest
     from django.utils.encoding import force_bytes
     from quixote import get_publisher
     from webtest import Hidden, Upload
     from wcs import fields
     from wcs.api_access import ApiAccess
     from wcs.blocks import BlockDef
     from wcs.categories import Category
     from wcs.data_sources import NamedDataSource
     from wcs.formdata import Evolution
     from wcs.formdef import FormDef
     from wcs.qommon import ods
     from wcs.qommon.http_request import HTTPRequest
     from wcs.qommon.ident.password_accounts import PasswordAccount
     from wcs.qommon.upload_storage import PicklableUpload
     from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem
     from wcs.workflows import (
         AttachmentEvolutionPart,
         EditableWorkflowStatusItem,
         Workflow,
         WorkflowBackofficeFieldsFormDef,
+    )
     from ..utilities import clean_temporary_pub, create_temporary_pub, get_app, login
-...
         resp = get_app(pub).get(sign_uri('/api/forms/test/%s/' % formdata.id, user=local_user))
         assert len(resp.json['evolution']) == 1
         assert len(resp.json['evolution'][0]['parts']) == 1
         part = resp.json['evolution'][0]['parts'][0]
         assert part['filename'] == 'hello.txt'
         assert part['content_type'] == 'text/plain'
         assert 'content' in part
         assert 'to' in part
         assert base64.decodebytes(force_bytes(part['content'])) == b'test'
         resp = get_app(pub).get(sign_uri('/api/forms/test/%s/?anonymise' % formdata.id, user=local_user))
         assert len(resp.json['evolution']) == 1
         assert 'parts' not in resp.json['evolution'][0]
         # check this doesn't get into list of forms API
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?full=on', user=local_user))
         assert 'hello.txt' not in resp.text
     def test_formdata_with_evolution_part_attachment_to(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.id = '123'
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         workflow = Workflow(name='test')
         st1 = workflow.add_status('Status1', 'st1')
         add_to_journal = RegisterCommenterWorkflowStatusItem()
         add_to_journal.id = '_add_to_journal'
         add_to_journal.comment = 'HELLO WORLD'
         add_to_journal.attachments = ['form_var_file_raw']
         add_to_journal.to = [role.id]
         st1.items.append(add_to_journal)
         workflow.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_id = workflow.id
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = [fields.FileField(id='1', label='File1', type='file', varname='file')]
         formdef.store()
         formdef.data_class().wipe()
         resp = get_app(pub).get('/test/')
         resp.forms[0]['f1$file'] = Upload('hello.txt', b'foobar', 'text/plain')
         resp = resp.forms[0].submit('submit')
         assert 'Check values then click submit.' in resp.text
         resp = resp.forms[0].submit('submit')
         assert resp.status_int == 302
         resp = resp.follow()
         assert 'The form has been recorded' in resp.text
         formdata = formdef.data_class().select()[0]
         resp = get_app(pub).get(sign_uri('/api/forms/test/%s/' % formdata.id, user=local_user))
         assert len(resp.json['evolution']) == 1
         assert len(resp.json['evolution'][0]['parts']) == 2
         assert resp.json['evolution'][0]['parts'][1]['type'] == 'workflow-comment'
         part = resp.json['evolution'][0]['parts'][0]
         assert part['type'] == 'workflow-attachment'
         assert part['filename'] == 'hello.txt'
         assert part['content_type'] == 'text/plain'
         assert part['to'] == ['123']
         assert 'content' in part
         assert base64.decodebytes(force_bytes(part['content'])) == b'foobar'
         resp = get_app(pub).get(sign_uri('/api/forms/test/%s/?anonymise' % formdata.id, user=local_user))
         assert len(resp.json['evolution']) == 1
         assert len(resp.json['evolution'][0]['parts']) == 1
         assert resp.json['evolution'][0]['parts'][0]['type'] == 'workflow-comment'
         # check this doesn't get into list of forms API
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?full=on', user=local_user))
         assert len(resp.json[0]['evolution']) == 1
         assert len(resp.json[0]['evolution'][0]['parts']) == 1
         assert resp.json[0]['evolution'][0]['parts'][0]['type'] == 'workflow-comment'
     def test_api_list_formdata(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'

         assert 'The form has been recorded' in resp.text
         formdata = formdef.data_class().select()[0]
         assert formdata.evolution[0].parts[0].content == 'Hello World'
         assert formdata.evolution[0].parts[0].to == [role2.id]
         resp = app.get('/test/%s/' % formdata.id)
         resp.status_int = 200
         assert resp.html.find('div', {'id': 'evolution-log'}).find('p').text == 'Hello World'
     def test_formdata_evolution_registercommenter_to_with_attachment(pub):
         user = create_user(pub)
         pub.role_class.wipe()
         role1 = pub.role_class(name='role the user does not have')
         role1.store()
         role2 = pub.role_class(name='role the user does have')
         role2.store()
         user.roles = [role2.id]
         user.store()
         wf = Workflow(name='status')
         st1 = wf.add_status('Status1', 'st1')
         comment = RegisterCommenterWorkflowStatusItem()
         comment.id = '1'
         comment.comment = 'Hello all'
         comment.attachments = ['form_var_file1_raw']
         comment.to = None
         st1.items.append(comment)
         comment.parent = st1
         comment = RegisterCommenterWorkflowStatusItem()
         comment.id = '2'
         comment.comment = 'Hello role1'
         comment.attachments = ['form_var_file2_raw']
         comment.to = [role1.id]
         st1.items.append(comment)
         comment.parent = st1
         comment = RegisterCommenterWorkflowStatusItem()
         comment.id = '3'
         comment.comment = 'Hello role2'
         comment.attachments = ['form_var_file3_raw']
         comment.to = [role2.id]
         st1.items.append(comment)
         comment.parent = st1
         wf.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_id = wf.id
         formdef.fields = [
             fields.FileField(id='1', label='File1', type='file', varname='file1'),
             fields.FileField(id='2', label='File2', type='file', varname='file2'),
             fields.FileField(id='3', label='File3', type='file', varname='file3'),
+        ]
         formdef.store()
         formdef.data_class().wipe()
         app = login(get_app(pub), username='foo', password='foo')
         resp = app.get('/test/')
         resp.forms[0]['f1$file'] = Upload('to-all.txt', b'foobar', 'text/plain')
         resp.forms[0]['f2$file'] = Upload('to-role1.txt', b'foobar', 'text/plain')
         resp.forms[0]['f3$file'] = Upload('to-role2.txt', b'foobar', 'text/plain')
         resp = resp.forms[0].submit('submit')
         assert 'Check values then click submit.' in resp.text
         resp = resp.forms[0].submit('submit')
         assert resp.status_int == 302
         resp = resp.follow()
         assert 'The form has been recorded' in resp.text
         formdata = formdef.data_class().select()[0]
         assert len(formdata.evolution[0].parts) == 6
         resp = app.get('/test/%s/' % formdata.id)
         resp.status_int = 200
         assert [x.a.text for x in resp.html.find_all('p', {'class': 'wf-attachment'})] == [
             'to-all.txt',
             'to-role2.txt',
+        ]

         register_commenter2.to = [role.id, '_submitter']
         user.roles = [role.id, role2.id]
         register_commenter2.perform(formdata)
         assert len(formdata.evolution[-1].parts) == 6
         assert '<p>d1</p>' in [str(x) for x in display_parts()]
         assert '<p>d2</p>' in [str(x) for x in display_parts()]
     def test_register_comment_to_with_attachment(pub):
         workflow = Workflow(name='register comment to with attachment')
         st1 = workflow.add_status('Status1', 'st1')
         role = pub.role_class(name='foorole')
         role.store()
         role2 = pub.role_class(name='no-one-role')
         role2.store()
         user = pub.user_class(name='baruser')
         user.roles = []
         user.store()
         upload1 = PicklableUpload('all.txt', 'text/plain')
         upload1.receive([b'barfoo'])
         upload2 = PicklableUpload('to-role.txt', 'text/plain')
         upload2.receive([b'barfoo'])
         upload3 = PicklableUpload('to-submitter.txt', 'text/plain')
         upload3.receive([b'barfoo'])
         upload4 = PicklableUpload('to-role-or-submitter.txt', 'text/plain')
         upload4.receive([b'barfoo'])
         FormDef.wipe()
         formdef = FormDef()
         formdef.url_name = 'foobar'
         formdef.fields = [
             FileField(id='1', label='File1', type='file', varname='file1'),
             FileField(id='2', label='File2', type='file', varname='file2'),
             FileField(id='3', label='File3', type='file', varname='file3'),
             FileField(id='4', label='File4', type='file', varname='file4'),
+        ]
         formdef._workflow = workflow
         formdef.store()
         formdata = formdef.data_class()()
         formdata.data = {'1': upload1, '2': upload2, '3': upload3, '4': upload4}
         formdata.just_created()
         assert formdata.status == 'wf-st1'
         pub.substitutions.feed(formdata)
         register_commenter = RegisterCommenterWorkflowStatusItem()
         register_commenter.parent = st1
         st1.items.append(register_commenter)
         def display_parts():
             formdata.evolution[-1]._display_parts = None  # invalidate cache
             return [str(x) for x in formdata.evolution[-1].display_parts()]
         register_commenter.comment = 'all'
         register_commenter.attachments = ['form_var_file1_raw']
         register_commenter.to = None
         register_commenter.perform(formdata)
         register_commenter.comment = 'to-role'
         register_commenter.attachments = ['form_var_file2_raw']
         register_commenter.to = [role.id]
         register_commenter.perform(formdata)
         register_commenter.comment = 'to-submitter'
         register_commenter.attachments = ['form_var_file3_raw']
         register_commenter.to = ['_submitter']
         register_commenter.perform(formdata)
         register_commenter.comment = 'to-role-or-submitter'
         register_commenter.attachments = ['form_var_file4_raw']
         register_commenter.to = [role.id, '_submitter']
         register_commenter.perform(formdata)
         assert len(formdata.evolution[-1].parts) == 8
         assert user.roles == []
         assert len(display_parts()) == 2
         assert 'all.txt' in display_parts()[0]
         assert display_parts()[1] == '<p>all</p>'
         pub._request._user = user
         user.roles = [role.id]
         assert len(display_parts()) == 6
         assert 'all.txt' in display_parts()[0]
         assert 'to-role.txt' in display_parts()[2]
         assert 'to-role-or-submitter.txt' in display_parts()[4]
         user.roles = []
         formdata.user_id = user.id
         assert len(display_parts()) == 6
         assert 'all.txt' in display_parts()[0]
         assert 'to-submitter.txt' in display_parts()[2]
         assert 'to-role-or-submitter.txt' in display_parts()[4]
         user.roles = [role.id]
         assert len(display_parts()) == 8
         assert 'all.txt' in display_parts()[0]
         assert 'to-role.txt' in display_parts()[2]
         assert 'to-submitter.txt' in display_parts()[4]
         assert 'to-role-or-submitter.txt' in display_parts()[6]
     def test_email(pub, emails):
         pub.substitutions.feed(MockSubstitutionVariables())
         formdef = FormDef()
         formdef.name = 'baz'
         formdef.fields = []
         formdef.store()

                         'render_br': False,
                         'options': [(None, '---', None)] + self.get_list_of_roles(include_logged_in_users=False),
                     },
+                )
         def get_parameters(self):
             return ('comment', 'to', 'attachments', 'condition')
         def attach_uploads_to_formdata(self, formdata, uploads):
         def attach_uploads_to_formdata(self, formdata, uploads, to):
             if not formdata.evolution[-1].parts:
                 formdata.evolution[-1].parts = []
             for upload in uploads:
                 try:
                     # useless but required to restore upload.fp from serialized state,
                     # needed by AttachmentEvolutionPart.from_upload()
                     upload.get_file_pointer()
                     formdata.evolution[-1].add_part(AttachmentEvolutionPart.from_upload(upload))
                     formdata.evolution[-1].add_part(AttachmentEvolutionPart.from_upload(upload, to=to))
                 except Exception:
                     get_publisher().notify_of_exception(sys.exc_info(), context='[comment/attachments]')
                 continue
         def perform(self, formdata):
             if not formdata.evolution:
                 return
             # process attachments first, they might be used in the comment
             # (with substitution vars)
             if self.attachments:
                 uploads = self.convert_attachments_to_uploads()
                 self.attach_uploads_to_formdata(formdata, uploads)
                 self.attach_uploads_to_formdata(formdata, uploads, self.to)
                 formdata.store()  # store and invalidate cache, so references can be used in the comment message.
             # the comment can use attachments done above
             try:
                 formdata.evolution[-1].add_part(JournalEvolutionPart(formdata, self.comment, self.to))
                 formdata.store()
             except TemplateError as e:
                 url = formdata.get_url()

             base_filename,
             fp,
             orig_filename=None,
             content_type=None,
             charset=None,
             varname=None,
             storage=None,
             storage_attrs=None,
             to=None,
         ):
             self.base_filename = base_filename
             self.orig_filename = orig_filename or base_filename
             self.content_type = content_type
             self.charset = charset
             self.fp = fp
             self.varname = varname
             self.storage = storage
             self.storage_attrs = storage_attrs
             self.to = to
         @classmethod
         def from_upload(cls, upload, varname=None):
         def from_upload(cls, upload, varname=None, to=None):
             return AttachmentEvolutionPart(
                 upload.base_filename,
                 getattr(upload, 'fp', None),
                 upload.orig_filename,
                 upload.content_type,
                 upload.charset,
                 varname=varname,
                 storage=getattr(upload, 'storage', None),
                 storage_attrs=getattr(upload, 'storage_attrs', None),
                 to=to,
+            )
         def get_file_pointer(self):
             if self.filename.startswith('uuid-'):
                 return None
             return open(self.filename, 'rb')  # pylint: disable=consider-using-with
         def __getstate__(self):
-...
         def get_json_export_dict(self, anonymise=False, include_files=True):
             if not include_files or anonymise:
                 return None
             d = {
                 'type': 'workflow-attachment',
                 'content_type': self.content_type,
                 'filename': self.base_filename,
                 'to': self.to,
+            }
             fd = self.get_file_pointer()
             if fd:
                 d['content'] = base64.encodebytes(fd.read())
                 fd.close()
             return d
         @classmethod
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.