0001-add-option-to-push-attached-document-to-portfolio-10.patch
tests/conftest.py | ||
---|---|---|
11 | 11 |
if 'postgresql' in item.keywords and item.config.option.without_postgresql_tests is True: |
12 | 12 |
pytest.skip('skipped (PostgreSQL are disabled on command line)') |
13 | 13 | |
14 | ||
14 | 15 |
def variable_url(request, pub, variable, url): |
16 |
return site_options(request, pub, 'options', variable, url) |
|
17 | ||
18 | ||
19 |
def site_options(request, pub, section, variable, value): |
|
15 | 20 |
config = ConfigParser.ConfigParser() |
16 | 21 |
path = os.path.join(pub.app_dir, 'site-options.cfg') |
17 | 22 |
if os.path.exists(path): |
18 | 23 |
config.read([path]) |
19 |
if not config.has_section('options'):
|
|
20 |
config.add_section('options')
|
|
21 |
config.set('options', variable, url)
|
|
24 |
if not config.has_section(section):
|
|
25 |
config.add_section(section)
|
|
26 |
config.set(section, variable, value)
|
|
22 | 27 |
with file(path, 'w') as site_option: |
23 | 28 |
config.write(site_option) |
24 | 29 | |
... | ... | |
26 | 31 |
config = ConfigParser.ConfigParser() |
27 | 32 |
if os.path.exists(path): |
28 | 33 |
config.read([path]) |
29 |
config.remove_option('options', variable)
|
|
34 |
config.remove_option(section, variable)
|
|
30 | 35 |
with file(path, 'w') as site_option: |
31 | 36 |
config.write(site_option) |
32 | 37 |
request.addfinalizer(fin) |
33 |
return url |
|
38 |
return value |
|
39 | ||
34 | 40 | |
35 | 41 |
@pytest.fixture |
36 | 42 |
def fargo_url(request, pub): |
37 | 43 |
return variable_url(request, pub, 'fargo_url', 'http://fargo.example.net') |
38 | 44 | |
45 | ||
46 |
@pytest.fixture |
|
47 |
def fargo_secret(request, pub): |
|
48 |
return site_options(request, pub, 'wscall-secrets', 'fargo.example.net', 'xxx') |
|
49 | ||
50 | ||
39 | 51 |
@pytest.fixture |
40 | 52 |
def welco_url(request, pub): |
41 | 53 |
return variable_url(request, pub, 'welco_url', 'http://welco.example.net') |
tests/test_form_pages.py | ||
---|---|---|
1904 | 1904 |
assert resp.content_type == 'application/pdf' |
1905 | 1905 |
assert resp.body.startswith('%PDF-') |
1906 | 1906 | |
1907 | ||
1908 |
@pytest.mark.skipif(transform_to_pdf is None, reason='libreoffice not found') |
|
1909 |
def test_formdata_generated_document_odt_to_pdf_download_push_to_portfolio(pub, fargo_url, |
|
1910 |
fargo_secret, caplog): |
|
1911 |
create_user(pub) |
|
1912 |
pub.cfg['debug'] = {'logger': True} |
|
1913 |
pub.write_cfg() |
|
1914 |
wf = Workflow(name='status') |
|
1915 |
st1 = wf.add_status('Status1', 'st1') |
|
1916 |
export_to = ExportToModel() |
|
1917 |
export_to.label = 'create doc' |
|
1918 |
export_to.varname = 'created_doc' |
|
1919 |
template_filename = os.path.join(os.path.dirname(__file__), 'template.odt') |
|
1920 |
template = open(template_filename).read() |
|
1921 |
upload = QuixoteUpload('/foo/template.odt', content_type='application/octet-stream') |
|
1922 |
upload.fp = StringIO.StringIO() |
|
1923 |
upload.fp.write(template) |
|
1924 |
upload.fp.seek(0) |
|
1925 |
export_to.model_file = UploadedFile(pub.app_dir, None, upload) |
|
1926 |
export_to.id = '_export_to' |
|
1927 |
export_to.by = ['_submitter'] |
|
1928 |
export_to.convert_to_pdf = True |
|
1929 |
export_to.push_to_portfolio = True |
|
1930 |
st1.items.append(export_to) |
|
1931 |
export_to.parent = st1 |
|
1932 |
wf.store() |
|
1933 | ||
1934 |
formdef = create_formdef() |
|
1935 |
formdef.workflow_id = wf.id |
|
1936 |
formdef.fields = [] |
|
1937 |
formdef.store() |
|
1938 |
formdef.data_class().wipe() |
|
1939 | ||
1940 |
resp = login(get_app(pub), username='foo', password='foo').get('/test/') |
|
1941 |
resp = resp.forms[0].submit('submit') |
|
1942 |
assert 'Check values then click submit.' in resp.body |
|
1943 |
resp = resp.forms[0].submit('submit') |
|
1944 |
assert resp.status_int == 302 |
|
1945 |
form_location = resp.location |
|
1946 |
resp = resp.follow() |
|
1947 |
assert 'The form has been recorded' in resp.body |
|
1948 | ||
1949 |
with mock.patch('wcs.file_validation.http_post_request') as http_post_request: |
|
1950 |
http_post_request.return_value = None, 200, 'null', None |
|
1951 |
resp = resp.form.submit('button_export_to') |
|
1952 |
assert http_post_request.call_count == 1 |
|
1953 |
assert ('file template.pdf pushed to portfolio of foo@localhost' |
|
1954 |
== caplog.records()[-1].message) |
|
1955 | ||
1956 |
resp = resp.follow() # $form/$id/create_doc |
|
1957 |
resp = resp.follow() # $form/$id/create_doc/ |
|
1958 |
assert resp.content_type == 'application/pdf' |
|
1959 |
assert 'PDF' in resp.body |
|
1960 | ||
1961 |
export_to.attach_to_history = True |
|
1962 |
wf.store() |
|
1963 | ||
1964 |
resp = login(get_app(pub), username='foo', password='foo').get(form_location) |
|
1965 |
with mock.patch('wcs.file_validation.http_post_request') as http_post_request: |
|
1966 |
http_post_request.return_value = None, 200, 'null', None |
|
1967 |
resp = resp.form.submit('button_export_to') |
|
1968 |
assert http_post_request.call_count == 1 |
|
1969 |
assert ('file template.pdf pushed to portfolio of foo@localhost' |
|
1970 |
== caplog.records()[-1].message) |
|
1971 |
assert resp.location == form_location |
|
1972 |
resp = resp.follow() # back to form page |
|
1973 | ||
1974 |
resp = resp.click('template.pdf') |
|
1975 |
assert resp.location.endswith('/template.pdf') |
|
1976 |
resp = resp.follow() |
|
1977 |
assert resp.content_type == 'application/pdf' |
|
1978 |
assert resp.body.startswith('%PDF-') |
|
1979 | ||
1980 | ||
1907 | 1981 |
def test_formdata_form_file_download(pub): |
1908 | 1982 |
create_user(pub) |
1909 | 1983 |
wf = Workflow(name='status') |
wcs/file_validation.py | ||
---|---|---|
18 | 18 |
import urlparse |
19 | 19 |
import hashlib |
20 | 20 |
import urllib |
21 |
import base64 |
|
21 | 22 | |
23 |
from qommon import get_logger |
|
22 | 24 |
from qommon.misc import http_get_page, json_loads, http_post_request |
23 |
from quixote import get_publisher, get_request |
|
25 |
from quixote import get_publisher, get_request, get_response
|
|
24 | 26 | |
25 | 27 |
from wcs.api_utils import get_secret_and_orig, sign_url |
26 | 28 | |
... | ... | |
56 | 58 |
return status, json_loads(response_payload) |
57 | 59 | |
58 | 60 | |
61 |
def fargo_post_json_async(url, payload): |
|
62 |
url = fargo_url(url) |
|
63 |
headers = {'Content-Type': 'application/json'} |
|
64 |
yield None |
|
65 |
response, status, response_payload, auth_header = http_post_request( |
|
66 |
url, json.dumps(payload), headers=headers) |
|
67 |
yield status, json_loads(response_payload) |
|
68 | ||
69 | ||
59 | 70 |
def sha256_of_upload(upload): |
60 | 71 |
return hashlib.sha256(upload.get_content()).hexdigest() |
61 | 72 | |
... | ... | |
129 | 140 |
upload.metadata = response['data'] |
130 | 141 |
filled.data['%s_structured' % field.id] = upload.metadata |
131 | 142 |
filled.store() |
143 | ||
144 | ||
145 |
def push_document(user, filename, content): |
|
146 |
if not user: |
|
147 |
return |
|
148 |
charset = get_publisher().site_charset |
|
149 |
payload = {} |
|
150 |
if user.name_identifiers: |
|
151 |
payload['user_nameid'] = unicode(user.name_identifiers[0], 'ascii') |
|
152 |
elif user.email: |
|
153 |
payload['user_email'] = unicode(user.email, 'ascii') |
|
154 |
payload['origin'] = get_request().get_server().split(':')[0] |
|
155 |
payload['file_name'] = unicode(filename, charset) |
|
156 |
payload['file_b64_content'] = base64.b64encode(content) |
|
157 |
async_post = fargo_post_json_async('/api/documents/push/', payload) |
|
158 |
next(async_post) |
|
159 | ||
160 |
def afterjob(job): |
|
161 |
status = 0 |
|
162 |
status, resp = next(async_post) |
|
163 |
import pdb |
|
164 |
pdb.set_trace() |
|
165 |
if status == 200: |
|
166 |
get_logger().info('file %s pushed to portfolio of %s' |
|
167 |
% (filename, user.display_name)) |
|
168 |
else: |
|
169 |
get_logger().error('failed %s failed to be pushed to portfolio of %s' |
|
170 |
% (filename, user.display_name)) |
|
171 | ||
172 |
get_response().add_after_job( |
|
173 |
N_('Sending file %s in portfolio of %s') % (filename, user.display_name), |
|
174 |
afterjob) |
wcs/wf/attachment.py | ||
---|---|---|
21 | 21 |
from qommon.errors import * |
22 | 22 | |
23 | 23 |
from wcs.forms.common import FormStatusPage, FileDirectory |
24 |
from wcs.file_validation import has_file_validation, push_document |
|
24 | 25 | |
25 | 26 |
def lookup_wf_attachment(self, filename): |
26 | 27 |
# supports for URLs such as /$formdata/$id/files/attachment/test.txt |
... | ... | |
79 | 80 |
by = [] |
80 | 81 |
backoffice_info_text = None |
81 | 82 |
varname = None |
83 |
push_to_portfolio = False |
|
82 | 84 | |
83 | 85 |
@classmethod |
84 | 86 |
def init(cls): |
... | ... | |
105 | 107 |
form.add_submit('button%s' % self.id, self.button_label or _('Upload File')) |
106 | 108 |
form.get_widget('button%s' % self.id).backoffice_info_text = self.backoffice_info_text |
107 | 109 | |
110 |
def push_file_to_portfolio(self, formdata, filename, outstream): |
|
111 |
outstream.seek(0) |
|
112 |
push_document(formdata.get_user(), filename, outstream.read()) |
|
113 | ||
108 | 114 |
def submit_form(self, form, formdata, user, evo): |
109 | 115 |
if form.get_widget('attachment%s' % self.id): |
110 | 116 |
f = form.get_widget('attachment%s' % self.id).parse() |
... | ... | |
112 | 118 |
if self.required: |
113 | 119 |
form.set_error('attachment%s' % self.id, _('Missing file')) |
114 | 120 |
return |
121 |
if self.push_to_portfolio: |
|
122 |
self.push_file_to_portfolio(formdata, upload.base_filename, upload.fp) |
|
115 | 123 |
evo.add_part(AttachmentEvolutionPart.from_upload(f, varname=self.varname)) |
116 | 124 | |
117 | 125 |
def get_parameters(self): |
118 |
return ('by', 'required', 'title', 'display_title', 'button_label',
|
|
126 |
parameters = ('by', 'required', 'title', 'display_title', 'button_label',
|
|
119 | 127 |
'display_button', 'hint', 'backoffice_info_text', 'varname') |
128 |
if has_file_validation(): |
|
129 |
parameters += ('push_to_portfolio',) |
|
130 |
return parameters |
|
120 | 131 | |
121 | 132 |
def add_parameters_widgets(self, form, parameters, prefix='', formdef=None): |
122 | 133 |
if 'by' in parameters: |
... | ... | |
146 | 157 |
if 'varname' in parameters: |
147 | 158 |
form.add(VarnameWidget, '%svarname' % prefix, |
148 | 159 |
title=_('Variable Name'), value=self.varname) |
160 |
if 'push_to_portfolio' in parameters: |
|
161 |
form.add(CheckboxWidget, '%spush_to_portfolio' % prefix, |
|
162 |
title=_('Push generated file to portfiolo'), |
|
163 |
value=self.push_to_portfolio) |
|
149 | 164 | |
150 | 165 | |
151 | 166 |
register_item_class(AddAttachmentWorkflowStatusItem) |
wcs/wf/export_to_model.py | ||
---|---|---|
37 | 37 |
from wcs.fields import SubtitleField, TitleField, CommentField, PageField |
38 | 38 |
from wcs.workflows import (WorkflowStatusItem, AttachmentEvolutionPart, |
39 | 39 |
template_on_formdata, register_item_class) |
40 |
from wcs.file_validation import has_file_validation, push_document |
|
40 | 41 | |
41 | 42 | |
42 | 43 |
try: |
... | ... | |
179 | 180 |
backoffice_info_text = None |
180 | 181 |
varname = None |
181 | 182 |
convert_to_pdf = False |
183 |
push_to_portfolio = False |
|
182 | 184 | |
183 | 185 |
def render_as_line(self): |
184 | 186 |
if self.label: |
... | ... | |
202 | 204 |
widget = form.get_widget('button%s' % self.id) |
203 | 205 |
widget.backoffice_info_text = self.backoffice_info_text |
204 | 206 | |
207 |
def push_file_to_portfolio(self, formdata, filename, outstream): |
|
208 |
outstream.seek(0) |
|
209 |
push_document(formdata.get_user(), filename, outstream.read()) |
|
210 | ||
205 | 211 |
def submit_form(self, form, formdata, user, evo): |
206 | 212 |
if form.get_submit() == 'button%s' % self.id: |
207 | 213 |
if not evo.comment: |
208 | 214 |
evo.comment = _('Form exported in a model') |
209 | 215 |
in_backoffice = get_request() and get_request().is_in_backoffice() |
216 |
outstream = self.apply_template_to_formdata(formdata) |
|
217 |
filename = self.model_file.base_filename |
|
218 |
content_type = self.model_file.content_type |
|
219 |
if self.convert_to_pdf: |
|
220 |
filename = filename.rsplit('.', 1)[0] + '.pdf' |
|
221 |
content_type = 'application/pdf' |
|
222 |
if self.push_to_portfolio: |
|
223 |
self.push_file_to_portfolio(formdata, filename, outstream) |
|
210 | 224 |
if self.attach_to_history: |
211 |
filename = self.model_file.base_filename |
|
212 |
content_type = self.model_file.content_type |
|
213 |
if self.convert_to_pdf: |
|
214 |
filename = filename.rsplit('.', 1)[0] + '.pdf' |
|
215 |
content_type = 'application/pdf' |
|
216 | 225 |
evo.add_part(AttachmentEvolutionPart( |
217 | 226 |
filename, |
218 |
self.apply_template_to_formdata(formdata),
|
|
227 |
outstream,
|
|
219 | 228 |
content_type=content_type, |
220 | 229 |
varname=self.varname)) |
221 | 230 |
return formdata.get_url(backoffice=in_backoffice) |
... | ... | |
309 | 318 |
form.add(CheckboxWidget, '%sconvert_to_pdf' % prefix, |
310 | 319 |
title=_('Convert generated file to PDF'), |
311 | 320 |
value=self.convert_to_pdf) |
321 |
if 'push_to_portfolio' in parameters: |
|
322 |
form.add(CheckboxWidget, '%spush_to_portfolio' % prefix, |
|
323 |
title=_('Push generated file to portfiolo'), |
|
324 |
value=self.push_to_portfolio) |
|
312 | 325 | |
313 | 326 |
def get_directory_name(self): |
314 | 327 |
return qommon.misc.simplify(self.label or 'export_to_model', space='_') |
... | ... | |
369 | 382 |
outstream.seek(0) |
370 | 383 |
return outstream |
371 | 384 | |
372 | ||
373 | 385 |
def get_parameters(self): |
374 |
parameters = ('by', 'label', 'model_file', 'attach_to_history', |
|
375 |
'backoffice_info_text', 'varname')
|
|
386 |
parameters = ('by', 'label', 'model_file', 'attach_to_history', 'backoffice_info_text',
|
|
387 |
'varname')
|
|
376 | 388 |
if transform_to_pdf is not None: |
377 | 389 |
parameters += ('convert_to_pdf',) |
390 |
if has_file_validation(): |
|
391 |
parameters += ('push_to_portfolio',) |
|
378 | 392 |
return parameters |
379 | 393 | |
380 | 394 |
def model_file_export_to_xml(self, xml_item, charset, include_id=False): |
381 |
- |