|
1 |
# -*- coding: utf-8 -*-
|
1 |
2 |
import mock
|
2 |
3 |
import pytest
|
3 |
4 |
|
... | ... | |
39 |
40 |
<ows:ServiceType>WFS</ows:ServiceType><ows:ServiceTypeVersion>2.0.0</ows:ServiceTypeVersion><ows:Fees/>
|
40 |
41 |
<ows:AccessConstraints/>
|
41 |
42 |
</ows:ServiceIdentification>
|
|
43 |
<ows:ServiceProvider/>
|
|
44 |
<ows:OperationsMetadata>
|
|
45 |
<ows:Operation name="GetFeature">
|
|
46 |
<ows:DCP/>
|
|
47 |
<ows:Parameter name="outputFormat">
|
|
48 |
<ows:AllowedValues>
|
|
49 |
<ows:Value>text/xml; subtype=gml/3.2</ows:Value>
|
|
50 |
<ows:Value>application/json</ows:Value>
|
|
51 |
<ows:Value>json</ows:Value>
|
|
52 |
</ows:AllowedValues>
|
|
53 |
</ows:Parameter>
|
|
54 |
</ows:Operation>
|
|
55 |
</ows:OperationsMetadata>
|
42 |
56 |
</wfs:WFS_Capabilities>'''
|
43 |
57 |
|
44 |
58 |
FAKE_SERVICE_CAPABILITIES_V1_0_0 = '''<?xml version="1.0" encoding="UTF-8"?>
|
... | ... | |
69 |
83 |
"properties": {
|
70 |
84 |
"nom": "Champagnier"
|
71 |
85 |
},
|
72 |
|
"type": "Feature"
|
|
86 |
"type": "Feature"
|
73 |
87 |
},
|
74 |
88 |
{
|
75 |
89 |
"geometry": null,
|
... | ... | |
115 |
129 |
"type": "FeatureCollection"
|
116 |
130 |
}'''
|
117 |
131 |
|
|
132 |
FAKE_FEATURES_XML = u'''<?xml version="1.0" encoding="UTF-8"?>
|
|
133 |
<wfs:FeatureCollection xmlns:gml="http://www.opengis.net/gml"
|
|
134 |
xmlns:wfs="http://www.opengis.net/wfs" xmlns:xxx="http://www.somewhere.net/xxx">
|
|
135 |
<gml:boundedBy/>
|
|
136 |
<gml:featureMember>
|
|
137 |
<xxx:points_apport_volontaire_dmt gml:id="points_apport_volontaire_dmt.e097e421cc609bcc752c43cd4e4c9d992cb0ee3d">
|
|
138 |
<xxx:geo_shape>
|
|
139 |
<gml:Point srsName="urn:x-ogc:def:crs:EPSG:4326">
|
|
140 |
<gml:pos>1914018.64 4224644.61</gml:pos>
|
|
141 |
</gml:Point>
|
|
142 |
</xxx:geo_shape>
|
|
143 |
<numero>4</numero>
|
|
144 |
<xxx:nom_voie>place victor hugo</xxx:nom_voie>
|
|
145 |
<xxx:code_post>38000</xxx:code_post>
|
|
146 |
<xxx:nom_commune>Grenoble</xxx:nom_commune>
|
|
147 |
</xxx:points_apport_volontaire_dmt>
|
|
148 |
</gml:featureMember>
|
|
149 |
</wfs:FeatureCollection>'''
|
|
150 |
|
118 |
151 |
FAKE_ERROR = u'''<ows:ExceptionReport
|
119 |
152 |
xmlns:xs="http://www.w3.org/2001/XMLSchema"
|
120 |
153 |
xmlns:ows="http://www.opengis.net/ows/1.1"
|
... | ... | |
241 |
274 |
def geoserver_responses_v1_0_0(url, **kwargs):
|
242 |
275 |
if kwargs['params'].get('request') == 'GetCapabilities':
|
243 |
276 |
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES_V1_0_0)
|
244 |
|
return utils.FakedResponse(status_code=200, content=FAKE_FEATURES_JSON)
|
|
277 |
return utils.FakedResponse(status_code=200, content=FAKE_FEATURES_XML)
|
|
278 |
|
245 |
279 |
|
246 |
280 |
def geoserver_responses_errors(url, **kwargs):
|
247 |
281 |
if kwargs['params'].get('request') == 'GetCapabilities':
|
248 |
282 |
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
249 |
283 |
return utils.FakedResponse(status_code=200, content=FAKE_ERROR)
|
250 |
284 |
|
|
285 |
def geoserver_responses_errors_v1_0_0(url, **kwargs):
|
|
286 |
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
287 |
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES_V1_0_0)
|
|
288 |
return utils.FakedResponse(status_code=200, content=FAKE_ERROR)
|
|
289 |
|
251 |
290 |
|
252 |
291 |
def geoserver_responses_errors_unparsable(url, **kwargs):
|
253 |
292 |
if kwargs['params'].get('request') == 'GetCapabilities':
|
254 |
293 |
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
255 |
294 |
return utils.FakedResponse(status_code=200, content=FAKE_ERROR[:10])
|
256 |
295 |
|
|
296 |
def geoserver_responses_errors_unparsable_v1_0_0(url, **kwargs):
|
|
297 |
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
298 |
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES_V1_0_0)
|
|
299 |
return utils.FakedResponse(status_code=200, content=FAKE_ERROR[:10])
|
|
300 |
|
257 |
301 |
|
258 |
302 |
@mock.patch('passerelle.utils.Request.get')
|
259 |
303 |
def test_feature_info(mocked_get, app, connector):
|
... | ... | |
318 |
362 |
assert resp.json['err_desc'] == 'no wfs URL declared'
|
319 |
363 |
|
320 |
364 |
|
|
365 |
@pytest.mark.parametrize("server_responses, nb_res", [
|
|
366 |
(geoserver_responses, 7),
|
|
367 |
(geoserver_responses_v1_0_0, 1),
|
|
368 |
])
|
321 |
369 |
@mock.patch('passerelle.utils.Request.get')
|
322 |
|
def test_get_feature(mocked_get, app, connector):
|
|
370 |
def test_get_feature(mocked_get, server_responses, nb_res, app, connector):
|
323 |
371 |
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
324 |
372 |
assert endpoint == '/opengis/test/features'
|
325 |
|
mocked_get.side_effect = geoserver_responses
|
|
373 |
mocked_get.side_effect = server_responses
|
326 |
374 |
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
327 |
375 |
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
328 |
376 |
assert mocked_get.call_args[1]['params']['PROPERTYNAME'] == 'nom'
|
329 |
|
assert mocked_get.call_args[1]['params']['TYPENAMES'] == 'ref_metro_limites_communales'
|
330 |
|
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == 'json'
|
|
377 |
assert mocked_get.call_args[1]['params'][connector.get_wfs_typename_label()] == 'ref_metro_limites_communales'
|
|
378 |
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == connector.get_wfs_outputformat()
|
331 |
379 |
assert mocked_get.call_args[1]['params']['SERVICE'] == 'WFS'
|
332 |
380 |
assert mocked_get.call_args[1]['params']['VERSION'] == connector.get_wfs_service_version()
|
333 |
|
assert len(resp.json['data']) == 7
|
|
381 |
assert len(resp.json['data']) == nb_res
|
334 |
382 |
for item in resp.json['data']:
|
335 |
383 |
assert 'text' in item
|
336 |
384 |
|
... | ... | |
368 |
416 |
assert 'CQL_FILTER' not in mocked_get.call_args[1]['params']
|
369 |
417 |
|
370 |
418 |
|
|
419 |
@pytest.mark.parametrize("server_responses", [
|
|
420 |
(geoserver_responses_errors),
|
|
421 |
(geoserver_responses_errors_v1_0_0),
|
|
422 |
])
|
371 |
423 |
@mock.patch('passerelle.utils.Request.get')
|
372 |
|
def test_get_feature_error(mocked_get, app, connector):
|
|
424 |
def test_get_feature_error(mocked_get, server_responses, app, connector):
|
373 |
425 |
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
374 |
426 |
assert endpoint == '/opengis/test/features'
|
375 |
|
mocked_get.side_effect = geoserver_responses_errors
|
|
427 |
mocked_get.side_effect = server_responses
|
376 |
428 |
resp = app.get(endpoint, params={
|
377 |
429 |
'type_names': 'ref_metro_limites_communales',
|
378 |
430 |
'property_name': 'nom'
|
379 |
431 |
})
|
380 |
432 |
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
381 |
433 |
assert mocked_get.call_args[1]['params']['PROPERTYNAME'] == 'nom'
|
382 |
|
assert mocked_get.call_args[1]['params']['TYPENAMES'] == 'ref_metro_limites_communales'
|
383 |
|
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == 'json'
|
|
434 |
assert mocked_get.call_args[1]['params'][connector.get_wfs_typename_label()] == 'ref_metro_limites_communales'
|
|
435 |
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == connector.get_wfs_outputformat()
|
384 |
436 |
assert mocked_get.call_args[1]['params']['SERVICE'] == 'WFS'
|
385 |
437 |
assert mocked_get.call_args[1]['params']['VERSION'] == connector.get_wfs_service_version()
|
386 |
438 |
result = resp.json
|
... | ... | |
389 |
441 |
assert 'Could not parse' in result['data']['text']
|
390 |
442 |
|
391 |
443 |
|
|
444 |
@pytest.mark.parametrize("server_responses, error_msg", [
|
|
445 |
(geoserver_responses_errors_unparsable, 'unparsable error'),
|
|
446 |
(geoserver_responses_errors_unparsable_v1_0_0, 'unclosed token: line 1, column 0'),
|
|
447 |
])
|
392 |
448 |
@mock.patch('passerelle.utils.Request.get')
|
393 |
|
def test_get_feature_error2(mocked_get, app, connector):
|
|
449 |
def test_get_feature_error2(mocked_get, server_responses, error_msg, app, connector):
|
394 |
450 |
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
395 |
451 |
assert endpoint == '/opengis/test/features'
|
396 |
|
mocked_get.side_effect = geoserver_responses_errors_unparsable
|
|
452 |
mocked_get.side_effect = server_responses
|
397 |
453 |
resp = app.get(endpoint, params={
|
398 |
454 |
'type_names': 'ref_metro_limites_communales',
|
399 |
455 |
'property_name': 'nom'
|
400 |
456 |
})
|
401 |
457 |
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
402 |
458 |
assert mocked_get.call_args[1]['params']['PROPERTYNAME'] == 'nom'
|
403 |
|
assert mocked_get.call_args[1]['params']['TYPENAMES'] == 'ref_metro_limites_communales'
|
404 |
|
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == 'json'
|
|
459 |
assert mocked_get.call_args[1]['params'][connector.get_wfs_typename_label()] == 'ref_metro_limites_communales'
|
|
460 |
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == connector.get_wfs_outputformat()
|
405 |
461 |
assert mocked_get.call_args[1]['params']['SERVICE'] == 'WFS'
|
406 |
462 |
assert mocked_get.call_args[1]['params']['VERSION'] == connector.get_wfs_service_version()
|
407 |
463 |
result = resp.json
|
408 |
464 |
assert result['err'] == 1
|
409 |
|
assert result['err_desc'] == 'OpenGIS Error: unparsable error'
|
|
465 |
assert result['err_desc'] == 'OpenGIS Error: %s' % error_msg
|
410 |
466 |
assert '<ows:' in result['data']['content']
|
411 |
467 |
|
412 |
468 |
|
413 |
469 |
@pytest.mark.parametrize("server_responses, version, typename_label", [
|
|
470 |
(geoserver_responses, '2.0.0', 'TYPENAMES'),
|
414 |
471 |
(geoserver_responses_v1_0_0, '1.0.0', 'TYPENAME'),
|
415 |
|
(geoserver_responses, '2.0.0', 'TYPENAMES')])
|
|
472 |
])
|
416 |
473 |
@mock.patch('passerelle.utils.Request.get')
|
417 |
474 |
def test_typename_parameter_upgrade(mocked_get, server_responses, version, typename_label, app, connector):
|
418 |
475 |
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
419 |
476 |
assert endpoint == '/opengis/test/features'
|
420 |
477 |
mocked_get.side_effect = server_responses
|
421 |
|
resp = app.get(endpoint, params={'type_names': '...', 'property_name': '...'})
|
|
478 |
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
422 |
479 |
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
423 |
480 |
assert mocked_get.call_args[1]['params']['VERSION'] == version
|
424 |
481 |
assert typename_label in mocked_get.call_args[1]['params'].keys()
|
|
482 |
assert mocked_get.call_args[1]['params'][connector.get_wfs_typename_label()] == 'ref_metro_limites_communales'
|
|
483 |
|
|
484 |
|
|
485 |
@pytest.mark.parametrize("server_responses, service_format", [
|
|
486 |
(geoserver_responses, 'json'),
|
|
487 |
(geoserver_responses_v1_0_0, 'xml'),
|
|
488 |
])
|
|
489 |
@mock.patch('passerelle.utils.Request.get')
|
|
490 |
def test_outputFormat_possible_values(mocked_get, server_responses, service_format, app, connector):
|
|
491 |
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
492 |
assert endpoint == '/opengis/test/features'
|
|
493 |
mocked_get.side_effect = server_responses
|
|
494 |
resp = app.get(endpoint, params={'type_names': '...', 'property_name': '...'})
|
|
495 |
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
|
496 |
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == service_format
|
425 |
497 |
|
426 |
498 |
|
|
499 |
@pytest.mark.parametrize("resp1, resp2, resp3, resp4", [
|
|
500 |
(FAKE_SERVICE_CAPABILITIES, FAKE_GEOLOCATED_FEATURE, '{"features": []}', '{}'),
|
|
501 |
(FAKE_SERVICE_CAPABILITIES_V1_0_0, FAKE_FEATURES_XML, '<?xml version="1.0"?><empty/>', ''),
|
|
502 |
])
|
427 |
503 |
@mock.patch('passerelle.utils.Request.get')
|
428 |
|
def test_reverse_geocoding(mocked_get, app, connector):
|
|
504 |
def test_reverse_geocoding(mocked_get, resp1, resp2, resp3, resp4, app, connector):
|
429 |
505 |
connector.search_radius = 45
|
430 |
506 |
connector.projection = 'EPSG:3945'
|
431 |
507 |
connector.save()
|
... | ... | |
434 |
510 |
|
435 |
511 |
def side_effect(url, **kwargs):
|
436 |
512 |
if kwargs['params'].get('request') == 'GetCapabilities':
|
437 |
|
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
513 |
return utils.FakedResponse(status_code=200, content=resp1)
|
438 |
514 |
return mock.DEFAULT
|
439 |
515 |
mocked_get.side_effect = side_effect
|
440 |
|
mocked_get.return_value = utils.FakedResponse(content=FAKE_GEOLOCATED_FEATURE, status_code=200)
|
|
516 |
mocked_get.return_value = utils.FakedResponse(content=resp2, status_code=200)
|
441 |
517 |
resp = app.get(endpoint,
|
442 |
518 |
params={
|
443 |
519 |
'lat': '45.1893469606986',
|
... | ... | |
455 |
531 |
connector.projection = 'EPSG:4326'
|
456 |
532 |
connector.search_radius = 10
|
457 |
533 |
connector.save()
|
458 |
|
mocked_get.return_value = utils.FakedResponse(content='{"features": []}', status_code=200)
|
|
534 |
mocked_get.return_value = utils.FakedResponse(content=resp3, status_code=200)
|
459 |
535 |
resp = app.get(
|
460 |
536 |
endpoint,
|
461 |
537 |
params={
|
... | ... | |
466 |
542 |
assert resp.json['err'] == 1
|
467 |
543 |
assert resp.json['err_desc'] == 'Unable to geocode'
|
468 |
544 |
|
469 |
|
mocked_get.return_value = utils.FakedResponse(status_code=404, content='{}', ok=False)
|
|
545 |
mocked_get.return_value = utils.FakedResponse(status_code=404, content=resp4, ok=False)
|
470 |
546 |
resp = app.get(endpoint,
|
471 |
547 |
params={
|
472 |
548 |
'lat': '45.183784',
|
473 |
|
-
|