0005-implement-URL-signatures-in-the-file-validation-web-.patch
tests/test_backoffice_pages.py | ||
---|---|---|
1703 | 1703 |
assert not '0_structured' in formdata.data |
1704 | 1704 |
resp = app.get('/backoffice/management/form-title/%s/' % form_id) |
1705 | 1705 |
assert not 'Validate' in resp.body |
1706 |
with mock.patch('wcs.file_validation.http_post_request') as http_post_request:
|
|
1706 |
with mock.patch('wcs.file_validation.fargo_post_json') as fargo_post_json:
|
|
1707 | 1707 |
resp = app.get('/backoffice/management/form-title/%s/validate?field_id=0' % form_id) |
1708 |
assert http_post_request.call_count == 0
|
|
1708 |
assert fargo_post_json.call_count == 0
|
|
1709 | 1709 |
resp = resp.follow() |
1710 | 1710 |
assert not 'Valid ' in resp.body |
1711 | 1711 |
assert not 'Validate' in resp.body |
... | ... | |
1780 | 1780 | |
1781 | 1781 |
fragment = '%s : %s' % (metadata_field['label'], metadata[metadata_field['varname']]) |
1782 | 1782 |
assert fragment in resp.body |
1783 |
with mock.patch('wcs.file_validation.http_post_request') as http_post_request:
|
|
1783 |
with mock.patch('wcs.file_validation.fargo_post_json') as fargo_post_json:
|
|
1784 | 1784 |
payload = { |
1785 | 1785 |
'user_nameid': '12345', |
1786 | 1786 |
'origin': 'example.net', |
... | ... | |
1799 | 1799 |
'end': '1978-01-01', |
1800 | 1800 |
'display': 'John Doe, 169 rue du château, 75014 PARIS' |
1801 | 1801 |
}) |
1802 |
http_post_request.return_value = None, 201, json.dumps(result), None
|
|
1802 |
fargo_post_json.return_value = 201, result
|
|
1803 | 1803 |
resp = app.get('/backoffice/management/form-title/%s/validate?field_id=0' % form_id) |
1804 |
assert http_post_request.call_count == 1 |
|
1805 |
assert http_post_request.call_args[0][0] == 'http://fargo.example.net/api/validation/justificatif-de-domicile/' |
|
1806 |
assert json_loads(http_post_request.call_args[0][1]) == payload |
|
1807 |
assert http_post_request.call_args[1] == {'headers': {'Content-Type': 'application/json'}} |
|
1804 |
assert fargo_post_json.call_count == 1 |
|
1805 |
assert fargo_post_json.call_args[0][0] == '/api/validation/justificatif-de-domicile/' |
|
1806 |
assert fargo_post_json.call_args[0][1] == payload |
|
1808 | 1807 |
resp = resp.follow() |
1809 | 1808 | |
1810 | 1809 |
assert 'Valid from 1970-01-01 to 1978-01-01' in resp.body |
1811 | 1810 | |
1812 | 1811 | |
1813 |
def test_backoffice_file_validation_no_upload(pub, fargo_url): |
|
1812 |
def test_backoffice_file_validation_no_upload(pub, fargo_url, fargo_secret):
|
|
1814 | 1813 |
document_type = { |
1815 | 1814 |
'id': 'justificatif-de-domicile', |
1816 | 1815 |
'fargo': True, |
... | ... | |
1859 | 1858 |
with mock.patch('wcs.file_validation.http_get_page') as http_get_page: |
1860 | 1859 |
http_get_page.return_value = None, 200, json.dumps(return_value), None |
1861 | 1860 |
resp = app.get('/form-title/') |
1862 |
http_get_page.assert_called_once_with( |
|
1861 |
assert http_get_page.call_count == 1 |
|
1862 |
assert http_get_page.call_args[0][0].startswith( |
|
1863 | 1863 |
'http://fargo.example.net/api/validation/justificatif-de-domicile/?user_nameid=12345') |
1864 | 1864 |
assert validation['display'] in resp.body |
1865 | 1865 |
resp.forms[0]['f0$validation_url'] = 'zob' |
... | ... | |
1872 | 1872 |
} |
1873 | 1873 | |
1874 | 1874 |
def side_effect(url): |
1875 |
if url == 'zob':
|
|
1875 |
if url.startswith('http://fargo.example.net/zob'):
|
|
1876 | 1876 |
return None, 200, json.dumps(return_value2), None |
1877 | 1877 |
else: |
1878 | 1878 |
return None, 200, json.dumps(return_value), None |
1879 | 1879 |
http_get_page.side_effect = side_effect |
1880 | 1880 |
resp = resp.forms[0].submit('submit') |
1881 |
assert http_get_page.call_args_list[0][0][0].endswith(
|
|
1882 |
'api/validation/justificatif-de-domicile/?user_nameid=12345')
|
|
1883 |
assert http_get_page.call_args_list[1][0][0] == 'zob'
|
|
1884 |
assert http_get_page.call_args_list[2][0][0] == 'zob'
|
|
1881 |
assert http_get_page.call_count == 3
|
|
1882 |
assert 'api/validation/justificatif-de-domicile/?user_nameid=12345' in http_get_page.call_args_list[0][0][0]
|
|
1883 |
assert http_get_page.call_args_list[1][0][0].startswith('http://fargo.example.net/zob')
|
|
1884 |
assert http_get_page.call_args_list[2][0][0].startswith('http://fargo.example.net/zob')
|
|
1885 | 1885 |
for key in metadata: |
1886 | 1886 |
assert 'value="%s"' % metadata[key] in resp.body |
1887 | 1887 |
assert 'Check values then click submit.' in resp.body |
wcs/file_validation.py | ||
---|---|---|
23 | 23 |
from qommon.misc import http_get_page, json_loads, http_post_request, ConnectionError |
24 | 24 |
from quixote import get_publisher, get_request |
25 | 25 | |
26 |
from wcs.api_utils import get_secret_and_orig, sign_url |
|
27 | ||
26 | 28 | |
27 | 29 |
def has_file_validation(): |
28 | 30 |
return get_publisher().get_site_option('fargo_url') is not None |
29 | 31 | |
30 | 32 | |
31 |
def fargo_get(path):
|
|
33 |
def fargo_url(url):
|
|
32 | 34 |
fargo_url = get_publisher().get_site_option('fargo_url') |
33 |
url = urlparse.urljoin(fargo_url, path) |
|
35 |
url = urlparse.urljoin(fargo_url, url) |
|
36 |
secret, orig = get_secret_and_orig(url) |
|
37 |
if '?' in url: |
|
38 |
url += '&orig=%s' % orig |
|
39 |
else: |
|
40 |
url += '?orig=%s' % orig |
|
41 |
return sign_url(url, secret) |
|
42 | ||
43 | ||
44 |
def fargo_get(url): |
|
45 |
url = fargo_url(url) |
|
34 | 46 |
response, status, data, auth_header = http_get_page(url) |
35 | 47 |
if status == 200: |
36 | 48 |
return json_loads(data) |
37 | 49 |
return None |
38 | 50 | |
39 | 51 | |
52 |
def fargo_post_json(url, payload): |
|
53 |
url = fargo_url(url) |
|
54 |
headers = {'Content-Type': 'application/json'} |
|
55 |
response, status, response_payload, auth_header = http_post_request( |
|
56 |
url, json.dumps(payload), headers=headers) |
|
57 |
return status, json_loads(response_payload) |
|
58 | ||
59 | ||
40 | 60 |
def sha256_of_upload(upload): |
41 | 61 |
return hashlib.sha256(upload.get_content()).hexdigest() |
42 | 62 | |
... | ... | |
67 | 87 | |
68 | 88 |
def get_validation(url): |
69 | 89 |
try: |
70 |
response, status, data, auth_header = http_get_page(url)
|
|
90 |
result = fargo_get(url)
|
|
71 | 91 |
except ConnectionError: |
72 | 92 |
get_logger().warning('unable to retrieve validation from fargo') |
73 | 93 |
return None |
74 |
if status == 200: |
|
75 |
return json_loads(data)['data'] |
|
76 |
return None |
|
94 |
return result['data'] if result else None |
|
77 | 95 | |
78 | 96 | |
79 | 97 |
def get_validations(document_type): |
... | ... | |
107 | 125 |
def validate(filled, field, upload): |
108 | 126 |
'''Compute link to Fargo to validate the given document''' |
109 | 127 |
document_type_id = field.document_type['id'] |
110 |
path = 'api/validation/%s/' % urllib.quote(document_type_id) |
|
111 |
fargo_url = get_publisher().get_site_option('fargo_url') |
|
112 |
url = urlparse.urljoin(fargo_url, path) |
|
128 |
url = '/api/validation/%s/' % urllib.quote(document_type_id) |
|
113 | 129 |
payload = {} |
114 | 130 |
if filled.user: |
115 | 131 |
if filled.user.name_identifiers: |
... | ... | |
122 | 138 |
for meta_field in field.metadata: |
123 | 139 |
if 'varname' in meta_field: |
124 | 140 |
payload[meta_field['varname']] = upload.metadata.get(meta_field['varname'], '') |
125 |
headers = {'Content-Type': 'application/json'} |
|
126 | 141 |
try: |
127 |
response, status, response_payload, auth_header = http_post_request(url, |
|
128 |
json.dumps(payload), |
|
129 |
headers=headers) |
|
142 |
status, response = fargo_post_json(url, payload) |
|
130 | 143 |
except ConnectionError: |
131 | 144 |
get_logger().warning('unable to validate document on fargo for %s', filled.get_display_id()) |
132 | 145 |
return |
133 | 146 |
if status == 201: |
134 |
upload.metadata = json_loads(response_payload)['data']
|
|
147 |
upload.metadata = response['data']
|
|
135 | 148 |
filled.data['%s_structured' % field.id] = upload.metadata |
136 | 149 |
filled.store() |
137 |
- |