entrouvert/wcs
entrouvert
/
wcs
14
1
Fork 0
Code Issues Pull Requests 30 Releases Activity

tests: use responses module to mock HTTP responses (#68527)

This commit is contained in:
Frédéric Péters 2022-08-29 20:01:05 +02:00
parent 34bf72a718
commit 74f0c9e8f0
9 changed files with 514 additions and 513 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
tests/admin_pages/test_datasource.py
View File

@ -1,11 +1,11 @@
import io
import json
import os
import re
import xml.etree.ElementTree as ET
from unittest import mock
import pytest
import responses
from webtest import Upload
from wcs import fields
@ -17,7 +17,7 @@ from wcs.formdef import FormDef
from wcs.qommon.http_request import HTTPRequest
from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef
from ..utilities import HttpRequestsMocking, clean_temporary_pub, create_temporary_pub, get_app, login
from ..utilities import clean_temporary_pub, create_temporary_pub, get_app, login
from .test_all import create_superuser
@ -452,26 +452,23 @@ def test_data_sources_view(pub):
assert 'Domicilié' in resp.text
# check json
json_file_path = os.path.join(pub.app_dir, 'test.json')
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}, json_file)
data_source.data_source = {'type': 'json', 'value': 'file://%s' % json_file_path}
data_source.data_source = {'type': 'json', 'value': 'https://example.net'}
data_source.store()
with HttpRequestsMocking():
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net', json={'data': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}
)
resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)
assert 'Preview' in resp.text
assert 'foo' in resp.text
# with other attributes
with open(json_file_path, 'w') as json_file:
json.dump({'results': [{'pk': '1', 'label': 'foo'}, {'pk': '2'}]}, json_file)
data_source.data_attribute = 'results'
data_source.id_attribute = 'pk'
data_source.text_attribute = 'label'
data_source.store()
with HttpRequestsMocking():
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={'results': [{'pk': '1', 'label': 'foo'}, {'pk': '2'}]})
resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)
assert 'Preview' in resp.text
assert '<tt>1</tt>: foo</li>' in resp.text
@ -482,34 +479,34 @@ def test_data_sources_view(pub):
data_source.data_attribute = None
data_source.data_source = {'type': 'json', 'value': '{{ site_url }}/foo/bar'}
data_source.store()
with HttpRequestsMocking():
resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)
resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)
assert '<a href="http://example.net/foo/bar"' in resp.text
assert 'Preview' not in resp.text
# errors
data_source.data_source = {'type': 'json', 'value': 'http://remote.example.net/404'}
data_source.notify_on_errors = True
data_source.record_on_errors = True
data_source.store()
resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)
with responses.RequestsMock() as rsps:
rsps.get('http://remote.example.net/404', status=404)
resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)
assert 'Preview' not in resp.text
assert pub.loggederror_class.count() == 0 # error not recorded
# check geojson
geojson_file_path = os.path.join(pub.app_dir, 'test.geojson')
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
data_source.data_source = {'type': 'geojson', 'value': 'https://example.net'}
data_source.store()
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'id': '1', 'text': 'foo', 'label': 'foo'}},
{'properties': {'id': '2', 'text': 'bar', 'label': 'bar'}},
]
},
geojson_file,
)
data_source.data_source = {'type': 'geojson', 'value': 'file://%s' % geojson_file_path}
data_source.store()
with HttpRequestsMocking():
resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)
assert 'Preview' in resp.text
assert 'foo' in resp.text

2
tests/form_pages/test_formdata.py
View File

@ -1522,7 +1522,7 @@ def test_formdata_named_wscall_in_conditions(http_requests, pub):
assert '>2nd page<' in resp.text
assert '>3rd page<' not in resp.text
assert '>4th page<' in resp.text
assert len(http_requests.requests) == 1
assert http_requests.count() == 1
def test_formdata_named_wscall_in_comment(pub):

2
tests/form_pages/test_live.py
View File

@ -505,7 +505,7 @@ def test_field_live_select_content_based_on_prefill(pub, http_requests):
assert resp.html.find('div', {'data-field-id': '1'}).attrs['data-live-source'] == 'true'
assert resp.html.find('div', {'data-field-id': '2'}).find('select')
assert resp.html.find('option', {'value': 'a'})
assert http_requests.get_last('url') == 'http://remote.example.net/json-list?plop=HELLO WORLD'
assert http_requests.get_last('url') == 'http://remote.example.net/json-list?plop=HELLO%20WORLD'
# check with autocomplete and a remote source with id
NamedDataSource.wipe()

727
tests/test_datasource.py
View File

@ -7,6 +7,7 @@ import xml.etree.ElementTree as ET
from unittest import mock
import pytest
import responses
from wcs import data_sources, fields
from wcs.categories import DataSourceCategory
@ -218,214 +219,232 @@ def test_python_datasource_with_evalutils(pub):
]
def test_json_datasource(pub, requests_pub, http_requests):
def test_json_datasource(pub, requests_pub):
get_request().datasources_cache = {}
datasource = {'type': 'json', 'value': ''}
assert data_sources.get_items(datasource) == []
# missing file
get_request().datasources_cache = {}
json_file_path = os.path.join(pub.app_dir, 'test.json')
datasource = {'type': 'json', 'value': 'file://%s' % json_file_path}
assert data_sources.get_items(datasource) == []
datasource = {'type': 'json', 'value': 'https://example.net'}
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', status=404)
assert data_sources.get_items(datasource) == []
# invalid json file
get_request().datasources_cache = {}
with open(json_file_path, 'wb') as json_file:
json_file.write(codecs.encode(b'foobar', 'zlib_codec'))
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', body=codecs.encode(b'foobar', 'zlib_codec'))
assert data_sources.get_items(datasource) == []
# empty json file
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({}, json_file)
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={})
assert data_sources.get_items(datasource) == []
# unrelated json file
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump('foobar', json_file)
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json='foobar')
assert data_sources.get_items(datasource) == []
# another unrelated json file
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': 'foobar'}, json_file)
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={'data': 'foobar'})
assert data_sources.get_items(datasource) == []
# json file not using dictionaries
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [['1', 'foo'], ['2', 'bar']]}, json_file)
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={'data': [['1', 'foo'], ['2', 'bar']]})
assert data_sources.get_items(datasource) == []
# a good json file
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}, json_file)
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo'},
{'id': '2', 'text': 'bar'},
]
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net', json={'data': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}
)
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo'},
{'id': '2', 'text': 'bar'},
]
# a json file with additional keys
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump(
{'data': [{'id': '1', 'text': 'foo', 'more': 'xxx'}, {'id': '2', 'text': 'bar', 'more': 'yyy'}]},
json_file,
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'data': [{'id': '1', 'text': 'foo', 'more': 'xxx'}, {'id': '2', 'text': 'bar', 'more': 'yyy'}]
},
)
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'more': 'xxx'},
{'id': '2', 'text': 'bar', 'more': 'yyy'},
]
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'more': 'xxx'},
{'id': '2', 'text': 'bar', 'more': 'yyy'},
]
# json specified with a variadic url
get_request().datasources_cache = {}
# json specified with a variadic url
get_request().datasources_cache = {}
class JsonUrlPath:
def get_substitution_variables(self):
return {'json_url': 'file://%s' % json_file_path}
class JsonUrlPath:
def get_substitution_variables(self):
return {'json_url': 'https://example.net'}
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': '[json_url]'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': '[json_url]'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': '{{ json_url }}'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': '{{ json_url }}'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
# json specified with a variadic url with an erroneous space
get_request().datasources_cache = {}
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': ' [json_url]'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
# json specified with a variadic url with an erroneous space
get_request().datasources_cache = {}
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': ' [json_url]'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': ' {{ json_url }}'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(JsonUrlPath())
datasource = {'type': 'json', 'value': ' {{ json_url }}'}
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'more': 'xxx'}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'more': 'yyy'}),
]
# a json file with integer as 'id'
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'id': 1, 'text': 'foo'}, {'id': 2, 'text': 'bar'}]}, json_file)
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': 1, 'text': 'foo'}),
('2', 'bar', '2', {'id': 2, 'text': 'bar'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': 1, 'text': 'foo'},
{'id': 2, 'text': 'bar'},
]
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={'data': [{'id': 1, 'text': 'foo'}, {'id': 2, 'text': 'bar'}]})
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': 1, 'text': 'foo'}),
('2', 'bar', '2', {'id': 2, 'text': 'bar'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': 1, 'text': 'foo'},
{'id': 2, 'text': 'bar'},
]
# a json file with empty or no text values
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'id': '1', 'text': ''}, {'id': '2'}]}, json_file)
assert data_sources.get_items(datasource) == [
('1', '', '1', {'id': '1', 'text': ''}),
('2', '2', '2', {'id': '2', 'text': '2'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': ''},
{'id': '2', 'text': '2'},
]
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={'data': [{'id': '1', 'text': ''}, {'id': '2'}]})
assert data_sources.get_items(datasource) == [
('1', '', '1', {'id': '1', 'text': ''}),
('2', '2', '2', {'id': '2', 'text': '2'}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': ''},
{'id': '2', 'text': '2'},
]
# a json file with empty or no id
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'id': '', 'text': 'foo'}, {'text': 'bar'}, {'id': None}]}, json_file)
assert data_sources.get_items(datasource) == []
assert data_sources.get_structured_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net', json={'data': [{'id': '', 'text': 'foo'}, {'text': 'bar'}, {'id': None}]}
)
assert data_sources.get_items(datasource) == []
assert data_sources.get_structured_items(datasource) == []
# a json file with invalid datatype for the text entry, (list in text key),
# the invalid entry will be skipped
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'id': '1', 'text': ['foo']}, {'id': '2', 'text': 'bar'}]}, json_file)
assert data_sources.get_items(datasource) == [('2', 'bar', '2', {'id': '2', 'text': 'bar'})]
assert data_sources.get_structured_items(datasource) == [{'id': '2', 'text': 'bar'}]
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net', json={'data': [{'id': '1', 'text': ['foo']}, {'id': '2', 'text': 'bar'}]}
)
assert data_sources.get_items(datasource) == [('2', 'bar', '2', {'id': '2', 'text': 'bar'})]
assert data_sources.get_structured_items(datasource) == [{'id': '2', 'text': 'bar'}]
# specify data_attribute
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'data_attribute': 'results'}
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'results': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}, json_file)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo'},
{'id': '2', 'text': 'bar'},
]
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net', json={'results': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}
)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo'},
{'id': '2', 'text': 'bar'},
]
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'data_attribute': 'data'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == []
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'data_attribute': 'data'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == []
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': {'results': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}}, json_file)
assert data_sources.get_structured_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={'data': {'results': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}},
)
assert data_sources.get_structured_items(datasource) == []
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'data_attribute': 'data.results'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo'},
{'id': '2', 'text': 'bar'},
]
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'data_attribute': 'data.results'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo'},
{'id': '2', 'text': 'bar'},
]
# specify id_attribute
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'id_attribute': 'pk'}
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'pk': '1', 'text': 'foo'}, {'pk': '2', 'text': 'bar'}]}, json_file)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'pk': '1'},
{'id': '2', 'text': 'bar', 'pk': '2'},
]
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net', json={'data': [{'pk': '1', 'text': 'foo'}, {'pk': '2', 'text': 'bar'}]}
)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'pk': '1'},
{'id': '2', 'text': 'bar', 'pk': '2'},
]
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'id_attribute': 'id'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == []
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'id_attribute': 'id'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == []
# specify text_attribute
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'text_attribute': 'label'}
get_request().datasources_cache = {}
with open(json_file_path, 'w') as json_file:
json.dump({'data': [{'id': '1', 'label': 'foo'}, {'id': '2', 'label': 'bar'}]}, json_file)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'label': 'foo'},
{'id': '2', 'text': 'bar', 'label': 'bar'},
]
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net', json={'data': [{'id': '1', 'label': 'foo'}, {'id': '2', 'label': 'bar'}]}
)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'label': 'foo'},
{'id': '2', 'text': 'bar', 'label': 'bar'},
]
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'text_attribute': 'text'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1', 'label': 'foo'},
{'id': '2', 'text': '2', 'label': 'bar'},
]
datasource = {'type': 'json', 'value': ' {{ json_url }}', 'text_attribute': 'text'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1', 'label': 'foo'},
{'id': '2', 'text': '2', 'label': 'bar'},
]
def test_json_datasource_bad_url(pub, error_email, http_requests, emails):
@ -441,15 +460,16 @@ def test_json_datasource_bad_url(pub, error_email, http_requests, emails):
}
assert data_sources.get_items(datasource) == []
assert emails.count() == 1
assert 'error in HTTP request to http://remote.example.net/404 (status: 404)' in emails.get_latest(
'subject'
assert (
'[ERROR] [DATASOURCE] Error loading JSON data source (error in HTTP request to (status: 404))'
in emails.get_latest('subject')
)
assert pub.loggederror_class.count() == 1
logged_error = pub.loggederror_class.select()[0]
assert logged_error.workflow_id is None
assert (
logged_error.summary
== "[DATASOURCE] Error loading JSON data source (error in HTTP request to http://remote.example.net/404 (status: 404))"
== "[DATASOURCE] Error loading JSON data source (error in HTTP request to (status: 404))"
)
datasource = {
@ -480,7 +500,7 @@ def test_json_datasource_bad_url(pub, error_email, http_requests, emails):
assert pub.loggederror_class.count() == 3
logged_error = pub.loggederror_class.select(order_by='id')[2]
assert logged_error.workflow_id is None
assert logged_error.summary == "[DATASOURCE] Error loading JSON data source (error)"
assert logged_error.summary.startswith("[DATASOURCE] Error loading JSON data source (error")
datasource = {
'type': 'json',
@ -555,266 +575,267 @@ def test_json_datasource_bad_qs_data(pub, error_email, emails, notify, record):
assert pub.loggederror_class.count() == 0
def test_geojson_datasource(pub, requests_pub, http_requests):
def test_geojson_datasource(pub, requests_pub):
get_request()
get_request().datasources_cache = {}
datasource = {'type': 'geojson', 'value': ''}
datasource = {'type': 'geojson', 'value': 'https://example.net'}
assert data_sources.get_items(datasource) == []
# missing file
get_request().datasources_cache = {}
geojson_file_path = os.path.join(pub.app_dir, 'test.geojson')
datasource = {'type': 'geojson', 'value': 'file://%s' % geojson_file_path}
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', status=404)
assert data_sources.get_items(datasource) == []
# invalid geojson file
get_request().datasources_cache = {}
with open(geojson_file_path, 'wb') as geojson_file:
geojson_file.write(codecs.encode(b'foobar', 'zlib_codec'))
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', body=codecs.encode(b'foobar', 'zlib_codec'))
assert data_sources.get_items(datasource) == []
# empty geojson file
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump({}, geojson_file)
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={})
assert data_sources.get_items(datasource) == []
# unrelated geojson file
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump('foobar', geojson_file)
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json='foobar')
assert data_sources.get_items(datasource) == []
# another unrelated geojson file
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump({'features': 'foobar'}, geojson_file)
assert data_sources.get_items(datasource) == []
with responses.RequestsMock() as rsps:
rsps.get('https://example.net', json={'features': 'foobar'})
assert data_sources.get_items(datasource) == []
# a good geojson file
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'id': '1', 'text': 'foo'}},
{'properties': {'id': '2', 'text': 'bar'}},
]
},
geojson_file,
)
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo'}}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar'}}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar'}},
]
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo'}}),
('2', 'bar', '2', {'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar'}}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar'}},
]
# a geojson file with additional keys
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
{'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
]
},
geojson_file,
)
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
]
]
# geojson specified with a variadic url
get_request().datasources_cache = {}
# geojson specified with a variadic url
get_request().datasources_cache = {}
class GeoJSONUrlPath:
def get_substitution_variables(self):
return {'geojson_url': 'file://%s' % geojson_file_path}
class GeoJSONUrlPath:
def get_substitution_variables(self):
return {'geojson_url': 'https://example.net'}
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': '[geojson_url]'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': '[geojson_url]'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': '{{ geojson_url }}'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': '{{ geojson_url }}'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
# geojson specified with a variadic url with an erroneous space
get_request().datasources_cache = {}
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': ' [geojson_url]'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
# geojson specified with a variadic url with an erroneous space
get_request().datasources_cache = {}
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': ' [geojson_url]'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
# same with django templated url
get_request().datasources_cache = {}
pub.substitutions.feed(GeoJSONUrlPath())
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}'}
assert data_sources.get_items(datasource) == [
(
'1',
'foo',
'1',
{'id': '1', 'text': 'foo', 'properties': {'id': '1', 'text': 'foo', 'more': 'xxx'}},
),
(
'2',
'bar',
'2',
{'id': '2', 'text': 'bar', 'properties': {'id': '2', 'text': 'bar', 'more': 'yyy'}},
),
]
# a geojson file with integer as 'id'
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'id': 1, 'text': 'foo'}},
{'properties': {'id': 2, 'text': 'bar'}},
]
},
geojson_file,
)
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': 1, 'text': 'foo', 'properties': {'id': 1, 'text': 'foo'}}),
('2', 'bar', '2', {'id': 2, 'text': 'bar', 'properties': {'id': 2, 'text': 'bar'}}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': 1, 'text': 'foo', 'properties': {'id': 1, 'text': 'foo'}},
{'id': 2, 'text': 'bar', 'properties': {'id': 2, 'text': 'bar'}},
]
assert data_sources.get_items(datasource) == [
('1', 'foo', '1', {'id': 1, 'text': 'foo', 'properties': {'id': 1, 'text': 'foo'}}),
('2', 'bar', '2', {'id': 2, 'text': 'bar', 'properties': {'id': 2, 'text': 'bar'}}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': 1, 'text': 'foo', 'properties': {'id': 1, 'text': 'foo'}},
{'id': 2, 'text': 'bar', 'properties': {'id': 2, 'text': 'bar'}},
]
# a geojson file with empty or no text values
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{'features': [{'properties': {'id': '1', 'text': ''}}, {'properties': {'id': '2'}}]}, geojson_file
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={'features': [{'properties': {'id': '1', 'text': ''}}, {'properties': {'id': '2'}}]},
)
assert data_sources.get_items(datasource) == [
('1', '1', '1', {'id': '1', 'text': '1', 'properties': {'id': '1', 'text': ''}}),
('2', '2', '2', {'id': '2', 'text': '2', 'properties': {'id': '2'}}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1', 'properties': {'id': '1', 'text': ''}},
{'id': '2', 'text': '2', 'properties': {'id': '2'}},
]
assert data_sources.get_items(datasource) == [
('1', '1', '1', {'id': '1', 'text': '1', 'properties': {'id': '1', 'text': ''}}),
('2', '2', '2', {'id': '2', 'text': '2', 'properties': {'id': '2'}}),
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1', 'properties': {'id': '1', 'text': ''}},
{'id': '2', 'text': '2', 'properties': {'id': '2'}},
]
# a geojson file with empty or no id
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'id': '', 'text': 'foo'}},
{'properties': {'text': 'bar'}},
{'properties': {'id': None}},
]
},
geojson_file,
)
assert data_sources.get_items(datasource) == []
assert data_sources.get_structured_items(datasource) == []
assert data_sources.get_items(datasource) == []
assert data_sources.get_structured_items(datasource) == []
# specify id_property
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}', 'id_property': 'gid'}
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'gid': '1', 'text': 'foo'}},
{'properties': {'gid': '2', 'text': 'bar'}},
]
},
geojson_file,
)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'gid': '1', 'text': 'foo'}},
{'id': '2', 'text': 'bar', 'properties': {'gid': '2', 'text': 'bar'}},
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'gid': '1', 'text': 'foo'}},
{'id': '2', 'text': 'bar', 'properties': {'gid': '2', 'text': 'bar'}},
]
# check with missing id property
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}', 'id_property': 'id'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == []
# check with missing id property
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}', 'id_property': 'id'}
get_request().datasources_cache = {}
assert data_sources.get_structured_items(datasource) == []
# check with feature IDs
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'id': '1', 'properties': {'text': 'foo'}},
{'id': '2', 'properties': {'text': 'bar'}},
]
},
geojson_file,
)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'text': 'foo'}},
{'id': '2', 'text': 'bar', 'properties': {'text': 'bar'}},
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'foo', 'properties': {'text': 'foo'}},
{'id': '2', 'text': 'bar', 'properties': {'text': 'bar'}},
]
# specify label_template_property
datasource = {
@ -823,52 +844,56 @@ def test_geojson_datasource(pub, requests_pub, http_requests):
'label_template_property': '{{ id }}: {{ text }}',
}
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'id': '1', 'text': 'foo'}},
{'properties': {'id': '2', 'text': 'bar'}},
]
},
geojson_file,
)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1: foo', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': '2: bar', 'properties': {'id': '2', 'text': 'bar'}},
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1: foo', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': '2: bar', 'properties': {'id': '2', 'text': 'bar'}},
]
# wrong template
get_request().datasources_cache = {}
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}', 'label_template_property': '{{ text }'}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '{{ text }', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': '{{ text }', 'properties': {'id': '2', 'text': 'bar'}},
]
get_request().datasources_cache = {}
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}', 'label_template_property': 'text'}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'text', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': 'text', 'properties': {'id': '2', 'text': 'bar'}},
]
# wrong template
get_request().datasources_cache = {}
datasource = {
'type': 'geojson',
'value': ' {{ geojson_url }}',
'label_template_property': '{{ text }',
}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '{{ text }', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': '{{ text }', 'properties': {'id': '2', 'text': 'bar'}},
]
get_request().datasources_cache = {}
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}', 'label_template_property': 'text'}
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': 'text', 'properties': {'id': '1', 'text': 'foo'}},
{'id': '2', 'text': 'text', 'properties': {'id': '2', 'text': 'bar'}},
]
# unknown property or empty value
datasource = {'type': 'geojson', 'value': ' {{ geojson_url }}', 'label_template_property': '{{ label }}'}
get_request().datasources_cache = {}
with open(geojson_file_path, 'w') as geojson_file:
json.dump(
{
with responses.RequestsMock() as rsps:
rsps.get(
'https://example.net',
json={
'features': [
{'properties': {'id': '1', 'text': 'foo', 'label': ''}},
{'properties': {'id': '2', 'text': 'bar'}},
]
},
geojson_file,
)
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1', 'properties': {'id': '1', 'text': 'foo', 'label': ''}},
{'id': '2', 'text': '2', 'properties': {'id': '2', 'text': 'bar'}},
]
assert data_sources.get_structured_items(datasource) == [
{'id': '1', 'text': '1', 'properties': {'id': '1', 'text': 'foo', 'label': ''}},
{'id': '2', 'text': '2', 'properties': {'id': '2', 'text': 'bar'}},
]
def test_geojson_datasource_bad_url(pub, http_requests, error_email, emails):
@ -886,7 +911,7 @@ def test_geojson_datasource_bad_url(pub, http_requests, error_email, emails):
assert logged_error.workflow_id is None
assert (
logged_error.summary
== "[DATASOURCE] Error loading JSON data source (error in HTTP request to http://remote.example.net/404 (status: 404))"
== "[DATASOURCE] Error loading JSON data source (error in HTTP request to (status: 404))"
)
datasource = {
@ -918,7 +943,7 @@ def test_geojson_datasource_bad_url(pub, http_requests, error_email, emails):
assert pub.loggederror_class.count() == 3
logged_error = pub.loggederror_class.select(order_by='id')[2]
assert logged_error.workflow_id is None
assert logged_error.summary == "[DATASOURCE] Error loading JSON data source (error)"
assert logged_error.summary.startswith("[DATASOURCE] Error loading JSON data source (error")
datasource = {
'type': 'geojson',

3
tests/test_hobo.py
View File

@ -421,6 +421,9 @@ def test_configure_authentication_methods(setuptest, http_requests):
hobo_cmd.configure_authentication_methods(service, pub)
hobo_cmd.configure_site_options(service, pub)
# reload options
pub.load_site_options()
idp_keys = list(pub.cfg['idp'].keys())
assert len(idp_keys) == 1
assert pub.cfg['idp'][idp_keys[0]]['metadata_url'] == 'http://authentic.example.net/idp/saml2/metadata'

12
tests/test_wscall.py
View File

@ -210,16 +210,20 @@ def test_webservice_on_error_with_sql(http_requests, emails, notify_on_errors, r
assert pub.loggederror_class.count() == 0
for url_part in ['400', '400-json', '404', '404-json', '500', 'json-err1', 'json-errheader1']:
status_code = 200
if not url_part.startswith('json'):
status_code = url_part[:3]
wscall.request = {'url': 'http://remote.example.net/%s' % url_part}
wscall.store()
resp = get_app(pub).get('/foobar/')
assert 'Foo Bar ' in resp.text
if notify_on_errors:
assert emails.count() == 1
assert "[ERROR] [WSCALL] %s whatever" % status_code in emails.emails
if url_part.startswith('400'):
assert '[ERROR] [WSCALL] 400 Bad Request' in emails.emails
elif url_part.startswith('404'):
assert '[ERROR] [WSCALL] 404 Not Found' in emails.emails
elif url_part.startswith('500'):
assert '[ERROR] [WSCALL] 500 Internal Server Error' in emails.emails
else:
assert '[ERROR] [WSCALL] 200 OK' in emails.emails
emails.empty()
else:
assert emails.count() == 0

214
tests/utilities.py
View File

@ -1,15 +1,13 @@
import http.cookies
import json
import os
import random
import shutil
import tempfile
import urllib.parse
import psycopg2
import responses
from django.conf import settings
from django.core import mail
from django.utils.encoding import force_bytes
from quixote import cleanup, get_publisher
from webtest import TestApp
@ -20,7 +18,6 @@ import wcs.qommon.sms
import wcs.wsgi
from wcs import compat, custom_views, sessions, sql
from wcs.qommon import force_str
from wcs.qommon.errors import ConnectionError
from wcs.qommon.tokens import Token
from wcs.roles import Role
from wcs.tracking_code import TrackingCode
@ -317,6 +314,10 @@ class EmailsMocking:
def emails(self):
return Emails()
def get_subjects(self):
for em in mail.outbox:
yield em.subject
def __enter__(self):
return self
@ -334,134 +335,103 @@ class HttpRequestsMocking:
self.requests = []
def __enter__(self):
import wcs.qommon.misc
self.requests_mock = responses.RequestsMock(assert_all_requests_are_fired=False)
self.requests_mock.get('http://remote.example.net/')
self.requests_mock.post('http://remote.example.net/')
self.requests_mock.put('http://remote.example.net/')
self.requests_mock.patch('http://remote.example.net/')
self.requests_mock.get('http://remote.example.net/204', status=204)
self.requests_mock.get('http://remote.example.net/400', status=400, body='bad request')
self.requests_mock.get(
'http://remote.example.net/400-json', status=400, json={"err": 1, "err_desc": ":("}
)
self.requests_mock.get('http://remote.example.net/404', status=404, body='page not found')
self.requests_mock.get('http://remote.example.net/404-json', status=404, json={"err": 1})
self.requests_mock.get('http://remote.example.net/500', status=500, body='internal server error')
self.requests_mock.get('http://remote.example.net/json-err1', json={'data': '', 'err': 1})
self.requests_mock.get('http://remote.example.net/json', json={"foo": "bar"})
self.requests_mock.post('http://remote.example.net/json', json={"foo": "bar"})
self.requests_mock.delete('http://remote.example.net/json', json={"foo": "bar"})
self.requests_mock.get(
'http://remote.example.net/json-list', json={"data": [{"id": "a", "text": "b"}]}
)
self.requests_mock.get(
'http://remote.example.net/json-list-extra',
json={"data": [{"id": "a", "text": "b", "foo": "bar"}]},
)
self.requests_mock.get(
'http://remote.example.net/json-list-extra-with-disabled',
json={
"data": [
{"id": "a", "text": "b", "foo": "bar"},
{"id": "c", "text": "d", "foo": "baz", "disabled": True},
]
},
)
self.requests_mock.get(
'http://remote.example.net/xml', body='<?xml version="1.0"><foo/>', content_type='text/xml'
)
self.requests_mock.get(
'http://remote.example.net/xml-errheader',
body='<?xml version="1.0"><foo/>',
content_type='text/xml',
headers={'x-error-code': '1'},
)
self.requests_mock.get('http://remote.example.net/json-err0', json={"data": "foo", "err": 0})
self.requests_mock.get('http://remote.example.net/json-err1', json={"data": "", "err": 1})
self.requests_mock.get(
'http://remote.example.net/json-list-err1', json={"data": [{"id": "a", "text": "b"}], "err": 1}
)
self.requests_mock.get('http://remote.example.net/json-errstr', json={"data": "", "err": "bug"})
self.requests_mock.get(
'http://remote.example.net/json-errheader0', json={"foo": "bar"}, headers={'x-error-code': '0'}
)
self.requests_mock.get(
'http://remote.example.net/json-errheader1', json={"foo": "bar"}, headers={'x-error-code': '1'}
)
self.requests_mock.get(
'http://remote.example.net/json-errheaderstr',
json={"foo": "bar"},
headers={'x-error-code': 'bug'},
)
self.requests_mock.get(
'http://remote.example.net/geojson',
json={
'features': [
{
'properties': {'id': '1', 'text': 'foo'},
'geometry': {'type': 'Point', 'coordinates': [1, 2]},
},
{
'properties': {'id': '2', 'text': 'bar'},
'geometry': {'type': 'Point', 'coordinates': [3, 4]},
},
]
},
)
self.requests_mock.post('https://portal/api/notification/add/', json={})
self.requests_mock.post('https://interco-portal/api/notification/add/', json={})
self.wcs_qommon_misc_http_request = wcs.qommon.misc._http_request
wcs.qommon.misc._http_request = self.http_request
with open(os.path.join(os.path.dirname(__file__), 'idp_metadata.xml')) as fd:
self.requests_mock.get('http://authentic.example.net/idp/saml2/metadata', body=fd.read())
with open(os.path.join(os.path.dirname(__file__), 'idp2_metadata.xml')) as fd:
self.requests_mock.get('http://authentic2.example.net/idp/saml2/metadata', body=fd.read())
self.requests_mock.start()
return self
def __exit__(self, exc_type, exc_value, tb):
import wcs.qommon.misc
wcs.qommon.misc._http_request = self.wcs_qommon_misc_http_request
del self.wcs_qommon_misc_http_request
def http_request(
self,
url,
method='GET',
body=None,
headers=None,
cert_file=None,
timeout=None,
raise_on_http_errors=False,
):
headers = headers or {}
self.requests.append(
{'url': url, 'method': method, 'body': body, 'headers': headers, 'timeout': timeout}
)
scheme, netloc, path = urllib.parse.urlparse(url)[:3]
base_url = urllib.parse.urlunparse((scheme, netloc, path, '', '', ''))
with open(os.path.join(os.path.dirname(__file__), 'idp_metadata.xml')) as fd:
metadata = fd.read()
with open(os.path.join(os.path.dirname(__file__), 'idp2_metadata.xml')) as fd:
metadata2 = fd.read()
geojson = {
'features': [
{
'properties': {'id': '1', 'text': 'foo'},
'geometry': {'type': 'Point', 'coordinates': [1, 2]},
},
{
'properties': {'id': '2', 'text': 'bar'},
'geometry': {'type': 'Point', 'coordinates': [3, 4]},
},
]
}
status, data, headers = {
'http://remote.example.net/204': (204, None, None),
'http://remote.example.net/400': (400, 'bad request', None),
'http://remote.example.net/400-json': (400, '{"err": 1, "err_desc": ":("}', None),
'http://remote.example.net/404': (404, 'page not found', None),
'http://remote.example.net/404-json': (404, '{"err": 1}', None),
'http://remote.example.net/500': (500, 'internal server error', None),
'http://remote.example.net/json': (200, '{"foo": "bar"}', None),
'http://remote.example.net/json-list': (200, '{"data": [{"id": "a", "text": "b"}]}', None),
'http://remote.example.net/json-list-extra': (
200,
'{"data": [{"id": "a", "text": "b", "foo": "bar"}]}',
None,
),
'http://remote.example.net/json-list-extra-with-disabled': (
200,
'{"data": [{"id": "a", "text": "b", "foo": "bar"}, {"id": "c", "text": "d", "foo": "baz", "disabled": true}]}',
None,
),
'http://remote.example.net/geojson': (200, json.dumps(geojson), None),
'http://remote.example.net/json-err0': (200, '{"data": "foo", "err": 0}', None),
'http://remote.example.net/json-err1': (200, '{"data": "", "err": 1}', None),
'http://remote.example.net/json-list-err1': (
200,
'{"data": [{"id": "a", "text": "b"}], "err": 1}',
None,
),
'http://remote.example.net/json-errstr': (200, '{"data": "", "err": "bug"}', None),
'http://remote.example.net/json-errheader0': (200, '{"foo": "bar"}', {'x-error-code': '0'}),
'http://remote.example.net/json-errheader1': (200, '{"foo": "bar"}', {'x-error-code': '1'}),
'http://remote.example.net/json-errheaderstr': (200, '{"foo": "bar"}', {'x-error-code': 'bug'}),
'http://remote.example.net/xml': (
200,
'<?xml version="1.0"><foo/>',
{'content-type': 'text/xml'},
),
'http://remote.example.net/xml-errheader': (
200,
'<?xml version="1.0"><foo/>',
{'content-type': 'text/xml', 'x-error-code': '1'},
),
'http://remote.example.net/connection-error': (None, None, None),
'http://authentic.example.net/idp/saml2/metadata': (200, metadata, None),
'http://authentic2.example.net/idp/saml2/metadata': (200, metadata2, None),
'https://portal/api/notification/add/': (200, {}, {}),
'https://interco-portal/api/notification/add/': (200, {}, {}),
}.get(base_url, (200, '', {}))
if url.startswith('file://'):
try:
with open(url[7:]) as fd:
status, data = 200, fd.read()
except OSError:
status = 404
data = force_bytes(data)
class FakeResponse:
def __init__(self, status, data, headers):
self.status_code = status
self.reason = 'whatever'
self.data = data
self.headers = headers or {}
self.length = len(data or '')
if status is None:
raise ConnectionError('error')
if raise_on_http_errors and not (200 <= status < 300):
raise ConnectionError('error in HTTP request to %s (status: %s)' % (url, status))
return FakeResponse(status, data, headers), status, data, None
self.requests_mock.stop()
def get_last(self, attribute):
return self.requests[-1][attribute]
if attribute in ('timeout', 'verify'):
return self.requests_mock.calls[-1].request.req_kwargs[attribute]
return getattr(self.requests_mock.calls[-1].request, attribute)
def empty(self):
self.requests = []
self.requests_mock.calls.reset()
def count(self):
return len(self.requests)
return len(self.requests_mock.calls)
class SMSMocking(wcs.qommon.sms.PasserelleSMS):

22
tests/workflow/test_all.py
View File

@ -1603,14 +1603,14 @@ def test_webservice_call(http_requests, pub):
item = WebserviceCallStatusItem()
item.url = 'http://remote.example.net'
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'GET'
item = WebserviceCallStatusItem()
item.url = 'http://remote.example.net'
item.post = True
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'POST'
payload = json.loads(http_requests.get_last('body'))
assert payload['url'] == 'http://example.net/baz/%s/' % formdata.id
@ -1627,7 +1627,7 @@ def test_webservice_call(http_requests, pub):
}
pub.substitutions.feed(formdata)
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'POST'
payload = json.loads(http_requests.get_last('body'))
assert payload == {
@ -1649,7 +1649,7 @@ def test_webservice_call(http_requests, pub):
}
pub.substitutions.feed(formdata)
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'POST'
payload = json.loads(http_requests.get_last('body'))
assert payload['extra'] == {'one': 1, 'str': 'abcd', 'decimal': '2', 'evalme': formdata.get_display_id()}
@ -1807,7 +1807,7 @@ def test_webservice_call(http_requests, pub):
item.action_on_4xx = ':stop'
with pytest.raises(AbortActionException):
item.perform(formdata)
assert formdata.evolution[-1].parts[-1].summary == '404 whatever'
assert formdata.evolution[-1].parts[-1].summary == '404 Not Found'
item = WebserviceCallStatusItem()
item.url = 'http://remote.example.net/xml'
@ -1902,7 +1902,7 @@ def test_webservice_call(http_requests, pub):
item.post_data = {'str': 'abcd', 'one': '=1', 'evalme': '=form_number', 'error': '=1=3'}
pub.substitutions.feed(formdata)
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'PUT'
payload = json.loads(http_requests.get_last('body'))
assert payload == {'one': 1, 'str': 'abcd', 'evalme': formdata.get_display_id()}
@ -1914,7 +1914,7 @@ def test_webservice_call(http_requests, pub):
item.post_data = {'str': 'abcd', 'one': '=1', 'evalme': '=form_number', 'error': '=1=3'}
pub.substitutions.feed(formdata)
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'PATCH'
payload = json.loads(http_requests.get_last('body'))
assert payload == {'one': 1, 'str': 'abcd', 'evalme': formdata.get_display_id()}
@ -2156,8 +2156,8 @@ def test_webservice_call_error_handling(http_requests, pub):
item.record_errors = True
item.action_on_network_errors = ':pass'
item.perform(formdata)
assert 'ConnectionError: error\n' in formdata.evolution[-1].parts[-1].summary
assert formdata.workflow_data['plop_connection_error'] == 'error'
assert 'ConnectionError: error' in formdata.evolution[-1].parts[-1].summary
assert formdata.workflow_data['plop_connection_error'].startswith('error')
def test_webservice_call_store_in_backoffice_filefield(http_requests, pub):
@ -2343,7 +2343,7 @@ def test_webservice_with_complex_data(http_requests, pub):
pub.substitutions.feed(formdata)
with get_publisher().complex_data():
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'POST'
payload = json.loads(http_requests.get_last('body'))
assert payload == {
@ -2371,7 +2371,7 @@ def test_webservice_with_complex_data(http_requests, pub):
del formdata.data['6']
with get_publisher().complex_data():
item.perform(formdata)
assert http_requests.get_last('url') == 'http://remote.example.net'
assert http_requests.get_last('url') == 'http://remote.example.net/'
assert http_requests.get_last('method') == 'POST'
payload = json.loads(http_requests.get_last('body'))
assert payload['bool'] is False

2
tox.ini
View File

@ -30,6 +30,7 @@ deps =
pyquery
mock
requests
responses
vobject
qrcode
Pillow
@ -67,6 +68,7 @@ deps =
pyquery
mock
requests
responses
vobject
qrcode
Pillow