1161 lines
44 KiB
Python
1161 lines
44 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
|
|
import tests.utils
|
|
from passerelle.apps.opengis.models import FeatureCache, OpenGIS, Query
|
|
from passerelle.base.models import Job
|
|
from passerelle.utils import import_site
|
|
from tests.test_manager import login
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
FAKE_FEATURE_INFO = '''<?xml version="1.0" encoding="UTF-8"?>
|
|
<msGMLOutput
|
|
xmlns:gml="http://www.opengis.net/gml"
|
|
xmlns:xlink="http://www.w3.org/1999/xlink"
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
xmlns:extra="http://example.net/">
|
|
<cad_cadastre.cadparcelle_layer>
|
|
<gml:name>Parcelle cadastrale (Plan cadastral informatise du Grand Lyon)</gml:name>
|
|
<cad_cadastre.cadparcelle_feature>
|
|
<gml:boundedBy>
|
|
<gml:Box srsName="EPSG:4171">
|
|
<gml:coordinates>4.784140,45.796890 4.784834,45.797365</gml:coordinates>
|
|
</gml:Box>
|
|
</gml:boundedBy>
|
|
<identifiant>69040BD309</identifiant>
|
|
<codedgi>040000BD0309</codedgi>
|
|
<numero>309</numero>
|
|
<surfacecadastrale>2406</surfacecadastrale>
|
|
<natureproprietaire>Particulier</natureproprietaire>
|
|
<indice>Parcelle figuree au plan</indice>
|
|
<arpentage>Arpentee</arpentage>
|
|
<gid>75404</gid>
|
|
<extra:foo>bar</extra:foo>
|
|
</cad_cadastre.cadparcelle_feature>
|
|
</cad_cadastre.cadparcelle_layer>
|
|
</msGMLOutput>'''
|
|
|
|
FAKE_FEATURE_INFO_FEATURE_COLLECTION = '''<?xml version="1.0" encoding="UTF-8"?>
|
|
<wfs:FeatureCollection xmlns:xs="http://www.w3.org/2001/XMLSchema"
|
|
xmlns:wfs="http://www.opengis.net/wfs"
|
|
xmlns:gml="http://www.opengis.net/gml"
|
|
xmlns:metropole-de-lyon="http://metropole-de-lyon"
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
xsi:schemaLocation="..."
|
|
>
|
|
<gml:boundedBy>
|
|
<gml:Envelope srsDimension="2" srsName="urn:x-ogc:def:crs:EPSG:3857">
|
|
<gml:lowerCorner>531238.20604672 5735107.61605308</gml:lowerCorner>
|
|
<gml:upperCorner>546464.26335844 5751147.23178471</gml:upperCorner>
|
|
</gml:Envelope>
|
|
</gml:boundedBy>
|
|
<gml:featureMembers>
|
|
<metropole-de-lyon:eco_ecologie.zfe gml:id="eco_ecologie.zfe.200046977-ZFE-001">
|
|
<gml:boundedBy>
|
|
<gml:Envelope srsName="urn:x-ogc:def:crs:EPSG:3857" srsDimension="2">
|
|
<gml:lowerCorner>531238.20604672 5735107.61605308</gml:lowerCorner>
|
|
<gml:upperCorner>546464.26335844 5751147.23178471</gml:upperCorner>
|
|
</gml:Envelope>
|
|
</gml:boundedBy>
|
|
<metropole-de-lyon:gid>200046977-ZFE-001</metropole-de-lyon:gid>
|
|
<metropole-de-lyon:date_debut>2020-01-01</metropole-de-lyon:date_debut>
|
|
<metropole-de-lyon:vp_critair>V5</metropole-de-lyon:vp_critair>
|
|
<metropole-de-lyon:vp_horaires>24/7</metropole-de-lyon:vp_horaires>
|
|
<metropole-de-lyon:vul_critair>V3</metropole-de-lyon:vul_critair>
|
|
<metropole-de-lyon:vul_horaires>24/7</metropole-de-lyon:vul_horaires>
|
|
<metropole-de-lyon:pl_critair>V3</metropole-de-lyon:pl_critair>
|
|
<metropole-de-lyon:pl_horaires>24/7</metropole-de-lyon:pl_horaires>
|
|
<metropole-de-lyon:autobus_autocars_critair>V5</metropole-de-lyon:autobus_autocars_critair>
|
|
<metropole-de-lyon:autobus_autocars_horaires>24/7</metropole-de-lyon:autobus_autocars_horaires>
|
|
<metropole-de-lyon:deux_rm_critair>V5</metropole-de-lyon:deux_rm_critair>
|
|
<metropole-de-lyon:deux_rm_horaires>24/7</metropole-de-lyon:deux_rm_horaires>
|
|
<metropole-de-lyon:url_arrete>https://agora.grandlyon.com/webdelib/files/unzip//seance_264250/23_d1647427974276.pdf</metropole-de-lyon:url_arrete>
|
|
<metropole-de-lyon:url_site_information>https://www.grandlyon.com/actions/zfe.html#c20726</metropole-de-lyon:url_site_information>
|
|
<metropole-de-lyon:the_geom>
|
|
<gml:Polygon srsName="urn:x-ogc:def:crs:EPSG:3857" srsDimension="2">
|
|
<gml:exterior>
|
|
<gml:LinearRing>
|
|
<gml:posList>...</gml:posList>
|
|
</gml:LinearRing>
|
|
</gml:exterior>
|
|
</gml:Polygon>
|
|
</metropole-de-lyon:the_geom>
|
|
<metropole-de-lyon:id>200046977-ZFE-001</metropole-de-lyon:id>
|
|
</metropole-de-lyon:eco_ecologie.zfe>
|
|
</gml:featureMembers>
|
|
</wfs:FeatureCollection>'''
|
|
|
|
FAKE_SERVICE_CAPABILITIES = '''<?xml version="1.0" encoding="UTF-8"?>
|
|
<wfs:WFS_Capabilities version="2.0.0"
|
|
xmlns:wfs="http://www.opengis.net/wfs/2.0"
|
|
xmlns:ows="http://www.opengis.net/ows/1.1">
|
|
<ows:ServiceIdentification>
|
|
<ows:Title/><ows:Abstract/>
|
|
<ows:ServiceType>WFS</ows:ServiceType><ows:ServiceTypeVersion>2.0.0</ows:ServiceTypeVersion><ows:Fees/>
|
|
<ows:AccessConstraints/>
|
|
</ows:ServiceIdentification>
|
|
<ows:OperationsMetadata>
|
|
<ows:Operation name="GetFeature">
|
|
<ows:Parameter name="outputFormat">
|
|
<ows:AllowedValues>
|
|
<ows:Value>application/gml+xml; version=3.2</ows:Value>
|
|
<ows:Value>application/json; subtype=geojson</ows:Value>
|
|
</ows:AllowedValues>
|
|
</ows:Parameter>
|
|
</ows:Operation>
|
|
</ows:OperationsMetadata>
|
|
</wfs:WFS_Capabilities>'''
|
|
|
|
FAKE_SERVICE_CAPABILITIES_V1_0_0 = '''<?xml version="1.0" encoding="UTF-8"?>
|
|
<wfs:WFS_Capabilities version="1.0.0"
|
|
xmlns:wfs="http://www.opengis.net/wfs/2.0"
|
|
xmlns:ows="http://www.opengis.net/ows/1.1">
|
|
<ows:ServiceIdentification>
|
|
<ows:Title/><ows:Abstract/>
|
|
<ows:ServiceType>WFS</ows:ServiceType><ows:ServiceTypeVersion>1.0.0</ows:ServiceTypeVersion><ows:Fees/>
|
|
<ows:AccessConstraints/>
|
|
</ows:ServiceIdentification>
|
|
<ows:OperationsMetadata>
|
|
<ows:Operation name="GetFeature">
|
|
<ows:Parameter name="outputFormat">
|
|
<ows:AllowedValues>
|
|
<ows:Value>application/gml+xml; version=3.2</ows:Value>
|
|
<ows:Value>application/json; subtype=geojson</ows:Value>
|
|
</ows:AllowedValues>
|
|
</ows:Parameter>
|
|
</ows:Operation>
|
|
</ows:OperationsMetadata>
|
|
</wfs:WFS_Capabilities>'''
|
|
|
|
FAKE_FEATURES_JSON = '''
|
|
{
|
|
"features": [
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27be",
|
|
"properties": {
|
|
"nom": "Bri\u00e9-et-Angonnes"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27bd",
|
|
"properties": {
|
|
"nom": "Champagnier"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27bb",
|
|
"properties": {
|
|
"nom": "Claix"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27ba",
|
|
"properties": {
|
|
"nom": "Corenc"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27b9",
|
|
"properties": {
|
|
"nom": "\u00c9chirolles"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27b8",
|
|
"properties": {
|
|
"nom": "Eybens"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27b7",
|
|
"properties": {
|
|
"nom": "Fontaine"
|
|
},
|
|
"type": "Feature"
|
|
}
|
|
],
|
|
"type": "FeatureCollection"
|
|
}'''
|
|
|
|
FAKE_ERROR = '''<ows:ExceptionReport
|
|
xmlns:xs="http://www.w3.org/2001/XMLSchema"
|
|
xmlns:ows="http://www.opengis.net/ows/1.1"
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
version="2.0.0"
|
|
xsi:schemaLocation="http://www.opengis.net/ows/1.1 https://sigmetropole.lametro.fr/geoserver/schemas/ows/1.1.0/owsAll.xsd">
|
|
<ows:Exception exceptionCode="NoApplicableCode">
|
|
<ows:ExceptionText>Could not parse CQL filter list.
|
|
Encountered &quot;BIS&quot; at line 1, column 129.
|
|
Was expecting one of:
|
|
&lt;EOF&gt;
|
|
&quot;and&quot; ...
|
|
&quot;or&quot; ...
|
|
&quot;;&quot; ...
|
|
&quot;/&quot; ...
|
|
&quot;*&quot; ...
|
|
&quot;+&quot; ...
|
|
&quot;-&quot; ...
|
|
Parsing : strEqualsIgnoreCase(nom_commune, &apos;Grenoble&apos;) = true AND strEqualsIgnoreCase(nom_voie, &apos;rue albert recoura&apos;) = true AND numero=8 BIS.</ows:ExceptionText>
|
|
</ows:Exception>
|
|
</ows:ExceptionReport>
|
|
'''
|
|
|
|
FAKE_GEOLOCATED_FEATURE = '''{
|
|
"crs": {
|
|
"properties": {
|
|
"name": "urn:ogc:def:crs:EPSG::3945"
|
|
},
|
|
"type": "name"
|
|
},
|
|
"features": [
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914059.51,
|
|
4224699.2
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_afnor": "BOULEVARD EDOUARD REY",
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "boulevard \u00e9douard rey",
|
|
"numero": 17
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914042.47,
|
|
4224665.2
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "place victor hugo",
|
|
"numero": 2
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914035.7,
|
|
4224700.42
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "boulevard \u00e9douard rey",
|
|
"numero": 28
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914018.64,
|
|
4224644.61
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "place victor hugo",
|
|
"numero": 4
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
[
|
|
[1914018, 4224644],
|
|
[1914018, 4224844],
|
|
[1914318, 4224944],
|
|
[1914318, 4224544]
|
|
]
|
|
],
|
|
"type": "Polygon"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_square": "place trapeze"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
[
|
|
[1914059, 4224699],
|
|
[1914059, 4224899],
|
|
[1914259, 4224699]
|
|
]
|
|
],
|
|
"type": "Polygon"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_square": "place triangle"
|
|
},
|
|
"type": "Feature"
|
|
}
|
|
],
|
|
"totalFeatures": 6,
|
|
"type": "FeatureCollection"
|
|
}'''
|
|
|
|
|
|
FAKE_GEOLOCATED_FEATURE_CIRCLE = {
|
|
'type': 'FeatureCollection',
|
|
'features': [
|
|
{
|
|
'type': 'Feature',
|
|
'properties': {
|
|
'in-circle': False,
|
|
'in-bbox': False,
|
|
},
|
|
'geometry': {'type': 'Point', 'coordinates': [2.3555374145507812, 48.906705448392216]},
|
|
},
|
|
{
|
|
'type': 'Feature',
|
|
'properties': {
|
|
'in-circle': True,
|
|
'in-bbox': True,
|
|
},
|
|
'geometry': {'type': 'Point', 'coordinates': [2.3366546630859375, 48.86990906900767]},
|
|
},
|
|
{
|
|
'type': 'Feature',
|
|
'properties': {
|
|
'in-circle': True,
|
|
'in-bbox': True,
|
|
},
|
|
'geometry': {'type': 'Point', 'coordinates': [2.344207763671875, 48.82901755447848]},
|
|
},
|
|
{
|
|
'type': 'Feature',
|
|
'properties': {
|
|
'in-circle': False,
|
|
'in-bbox': True,
|
|
},
|
|
'geometry': {'type': 'Point', 'coordinates': [2.304, 48.8086]},
|
|
},
|
|
{
|
|
'type': 'Feature',
|
|
'properties': {
|
|
'in-circle': True,
|
|
'in-bbox': True,
|
|
},
|
|
'geometry': {
|
|
'type': 'Polygon',
|
|
'coordinates': [[[2.304, 48.8086], [2.304, 49.8086], [1.304, 48.8086]]],
|
|
},
|
|
},
|
|
{
|
|
'type': 'Feature',
|
|
'properties': {
|
|
'in-circle': False,
|
|
'in-bbox': False,
|
|
},
|
|
'geometry': {
|
|
'type': 'Polygon',
|
|
'coordinates': [[[-2.304, 48.8086], [-2.304, 49.8086], [-1.304, 48.8086]]],
|
|
},
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def connector():
|
|
return tests.utils.setup_access_rights(
|
|
OpenGIS.objects.create(
|
|
slug='test', wms_service_url='http://example.net/wms', wfs_service_url='http://example.net/wfs'
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def query(connector):
|
|
return Query.objects.create(
|
|
resource=connector,
|
|
name='Test Query',
|
|
slug='test_query',
|
|
description='Test query.',
|
|
typename='pvo_patrimoine_voirie.pvoparking',
|
|
filter_expression=(
|
|
'<Filter><PropertyIsEqualTo><PropertyName>typeparking'
|
|
'</PropertyName></PropertyIsEqualTo></Filter>'
|
|
),
|
|
)
|
|
|
|
|
|
def geoserver_responses(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
assert kwargs['params'].get('service')
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_FEATURES_JSON)
|
|
|
|
|
|
def geoserver_responses_v1_0_0(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
assert kwargs['params'].get('service')
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES_V1_0_0)
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_FEATURES_JSON)
|
|
|
|
|
|
def geoserver_responses_errors(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_ERROR)
|
|
|
|
|
|
def geoserver_responses_errors_unparsable(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_ERROR[:10])
|
|
|
|
|
|
def geoserver_geolocated_responses(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_GEOLOCATED_FEATURE)
|
|
|
|
|
|
def geoserver_circle_responses(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(status_code=200, content=json.dumps(FAKE_GEOLOCATED_FEATURE_CIRCLE))
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_feature_info(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'feature_info', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/feature_info'
|
|
mocked_get.return_value = tests.utils.FakedResponse(content=FAKE_FEATURE_INFO, status_code=200)
|
|
resp = app.get(endpoint, params={'lat': '45.796890', 'lon': '4.784140'})
|
|
assert (
|
|
mocked_get.call_args[1]['params']['bbox']
|
|
== '532556.896735,5747844.261214,532579.160633,5747876.194333'
|
|
)
|
|
assert mocked_get.call_args[1]['params']['crs'] == 'EPSG:3857'
|
|
assert (
|
|
resp.json['data']['cad_cadastrecadparcelle_layer']['cad_cadastrecadparcelle_feature'][
|
|
'natureproprietaire'
|
|
]
|
|
== 'Particulier'
|
|
)
|
|
assert 'name' not in resp.json['data']['cad_cadastrecadparcelle_layer']
|
|
assert (
|
|
'boundedBy'
|
|
not in resp.json['data']['cad_cadastrecadparcelle_layer']['cad_cadastrecadparcelle_feature']
|
|
)
|
|
assert 'foo' in resp.json['data']['cad_cadastrecadparcelle_layer']['cad_cadastrecadparcelle_feature']
|
|
connector.projection = 'EPSG:4326'
|
|
connector.save()
|
|
resp = app.get(endpoint, params={'lat': '45.796890', 'lon': '4.784140'})
|
|
assert mocked_get.call_args[1]['params']['bbox'] == '45.796790,4.784040,45.796990,4.784240'
|
|
assert mocked_get.call_args[1]['params']['crs'] == 'EPSG:4326'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_feature_info_feature_collection(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'feature_info', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/feature_info'
|
|
mocked_get.return_value = tests.utils.FakedResponse(
|
|
content=FAKE_FEATURE_INFO_FEATURE_COLLECTION, status_code=200
|
|
)
|
|
resp = app.get(endpoint, params={'lat': '45.796890', 'lon': '4.784140'})
|
|
assert (
|
|
mocked_get.call_args[1]['params']['bbox']
|
|
== '532556.896735,5747844.261214,532579.160633,5747876.194333'
|
|
)
|
|
assert mocked_get.call_args[1]['params']['crs'] == 'EPSG:3857'
|
|
assert resp.json['data']['autobus_autocars_critair'] == 'V5'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize(
|
|
'lat,lon',
|
|
[
|
|
('bad-value', '4.784140'),
|
|
('45.796890', 'bad-value'),
|
|
],
|
|
)
|
|
def test_feature_info_bad_request(mocked_get, app, connector, lat, lon):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'feature_info', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/feature_info'
|
|
mocked_get.return_value = tests.utils.FakedResponse(content=FAKE_FEATURE_INFO, status_code=200)
|
|
resp = app.get(endpoint, params={'lat': lat, 'lon': lon})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Bad coordinates format'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_tile(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'tile', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/tile'
|
|
mocked_get.return_value = tests.utils.FakedResponse(
|
|
content=b'\x89PNG\r\n\x1a\n\x00\x00...', status_code=200
|
|
)
|
|
resp = app.get(endpoint + '/16/33650/23378.png')
|
|
assert mocked_get.call_args[1]['params']['crs'] == 'EPSG:3857'
|
|
assert (
|
|
mocked_get.call_args[1]['params']['bbox']
|
|
== '539339.671580,5741338.068556,539951.167806,5741949.564782'
|
|
)
|
|
connector.projection = 'EPSG:4326'
|
|
connector.save()
|
|
resp = app.get(endpoint + '/16/33650/23378.png')
|
|
assert mocked_get.call_args[1]['params']['crs'] == 'EPSG:4326'
|
|
assert mocked_get.call_args[1]['params']['bbox'] == '45.756026,4.844971,45.759859,4.850464'
|
|
assert resp.content == b'\x89PNG\r\n\x1a\n\x00\x00...'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_feature_with_no_wfs_url(mocked_get, app, connector):
|
|
connector.wfs_service_url = ''
|
|
connector.save()
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/features'
|
|
mocked_get.side_effect = geoserver_responses
|
|
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
|
assert resp.json['data'] is None
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'no wfs URL declared'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_feature(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/features'
|
|
mocked_get.side_effect = geoserver_responses
|
|
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
|
assert mocked_get.call_args[1]['params']['request'] == 'GetFeature'
|
|
assert mocked_get.call_args[1]['params']['propertyName'] == 'nom'
|
|
assert mocked_get.call_args[1]['params']['typenames'] == 'ref_metro_limites_communales'
|
|
assert 'json' in mocked_get.call_args[1]['params']['outputFormat']
|
|
assert mocked_get.call_args[1]['params']['service'] == 'WFS'
|
|
assert mocked_get.call_args[1]['params']['version'] == connector.get_wfs_service_version()
|
|
assert len(resp.json['data']) == 7
|
|
for item in resp.json['data']:
|
|
assert 'text' in item
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_filtered_feature(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
mocked_get.side_effect = geoserver_responses
|
|
app.get(
|
|
endpoint,
|
|
params={
|
|
'type_names': 'ref_metro_limites_communales',
|
|
'property_name': 'nom',
|
|
'cql_filter': 'nom=\'Fontaine\'',
|
|
},
|
|
)
|
|
assert mocked_get.call_args[1]['params']['cql_filter'] == 'nom=\'Fontaine\''
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_filtered_by_property_feature(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
mocked_get.side_effect = geoserver_responses
|
|
params = {
|
|
'type_names': 'ref_metro_limites_communales',
|
|
'property_name': 'nom',
|
|
'cql_filter': 'nom=\'Fontaine\'',
|
|
'filter_property_name': 'nom',
|
|
}
|
|
app.get(endpoint, params=params)
|
|
assert mocked_get.call_args[1]['params']['cql_filter'] == 'nom=\'Fontaine\''
|
|
params['q'] = 'bens'
|
|
app.get(endpoint, params=params)
|
|
assert mocked_get.call_args[1]['params']['cql_filter'] == 'nom=\'Fontaine\' AND nom LIKE \'%bens%\''
|
|
params['case-insensitive'] = True
|
|
app.get(endpoint, params=params)
|
|
assert mocked_get.call_args[1]['params']['cql_filter'] == 'nom=\'Fontaine\' AND nom ILIKE \'%bens%\''
|
|
del params['case-insensitive']
|
|
params['case_insensitive'] = True
|
|
assert mocked_get.call_args[1]['params']['cql_filter'] == 'nom=\'Fontaine\' AND nom ILIKE \'%bens%\''
|
|
params.pop('cql_filter')
|
|
app.get(endpoint, params=params)
|
|
assert 'cql_filter' not in mocked_get.call_args[1]['params']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_feature_error(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/features'
|
|
mocked_get.side_effect = geoserver_responses_errors
|
|
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
|
assert mocked_get.call_args[1]['params']['request'] == 'GetFeature'
|
|
assert mocked_get.call_args[1]['params']['propertyName'] == 'nom'
|
|
assert mocked_get.call_args[1]['params']['typenames'] == 'ref_metro_limites_communales'
|
|
assert 'json' in mocked_get.call_args[1]['params']['outputFormat']
|
|
assert mocked_get.call_args[1]['params']['service'] == 'WFS'
|
|
assert mocked_get.call_args[1]['params']['version'] == connector.get_wfs_service_version()
|
|
result = resp.json
|
|
assert result['err'] == 1
|
|
assert result['err_desc'] == 'OpenGIS Error: NoApplicableCode'
|
|
assert 'Could not parse' in result['data']['text']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_feature_error2(mocked_get, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/features'
|
|
mocked_get.side_effect = geoserver_responses_errors_unparsable
|
|
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
|
assert mocked_get.call_args[1]['params']['request'] == 'GetFeature'
|
|
assert mocked_get.call_args[1]['params']['propertyName'] == 'nom'
|
|
assert mocked_get.call_args[1]['params']['typenames'] == 'ref_metro_limites_communales'
|
|
assert 'json' in mocked_get.call_args[1]['params']['outputFormat']
|
|
assert mocked_get.call_args[1]['params']['service'] == 'WFS'
|
|
assert mocked_get.call_args[1]['params']['version'] == connector.get_wfs_service_version()
|
|
result = resp.json
|
|
assert result['err'] == 1
|
|
assert result['err_desc'] == 'OpenGIS Error: unparsable error'
|
|
assert '<ows:' in result['data']['content']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_feature_bad_result(mocked_get, app, connector):
|
|
def keyerror(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
assert kwargs['params'].get('service')
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(status_code=200, content=json.dumps({}))
|
|
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
mocked_get.side_effect = keyerror
|
|
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'OpenGIS Error: bad result format'
|
|
assert resp.json['data'] == {'content': "'{}'"}
|
|
|
|
def typeerror(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
assert kwargs['params'].get('service')
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(status_code=200, content=json.dumps([]))
|
|
|
|
mocked_get.side_effect = typeerror
|
|
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'OpenGIS Error: bad result format'
|
|
assert resp.json['data'] == {'content': "'[]'"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'server_responses, version, typename_label',
|
|
[(geoserver_responses_v1_0_0, '1.0.0', 'typename'), (geoserver_responses, '2.0.0', 'typenames')],
|
|
)
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_typename_parameter_upgrade(mocked_get, server_responses, version, typename_label, app, connector):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/features'
|
|
mocked_get.side_effect = server_responses
|
|
app.get(endpoint, params={'type_names': '...', 'property_name': '...'})
|
|
assert mocked_get.call_args[1]['params']['request'] == 'GetFeature'
|
|
assert mocked_get.call_args[1]['params']['version'] == version
|
|
assert typename_label in mocked_get.call_args[1]['params'].keys()
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_reverse_geocoding(mocked_get, app, connector):
|
|
connector.search_radius = 45
|
|
connector.projection = 'EPSG:3945'
|
|
connector.save()
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'reverse', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/reverse'
|
|
|
|
def side_effect(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return mock.DEFAULT
|
|
|
|
mocked_get.side_effect = side_effect
|
|
mocked_get.return_value = tests.utils.FakedResponse(content=FAKE_GEOLOCATED_FEATURE, status_code=200)
|
|
resp = app.get(endpoint, params={'lat': '45.1893469606986', 'lon': '5.72462060798'})
|
|
assert (
|
|
mocked_get.call_args[1]['params']['cql_filter']
|
|
== 'DWITHIN(the_geom,Point(1914061.486036 4224640.457791),45,meters)'
|
|
)
|
|
assert resp.json['lon'] == '5.724077'
|
|
assert resp.json['lat'] == '45.189397'
|
|
assert resp.json['address']['house_number'] == '4'
|
|
assert resp.json['address']['road'] == 'place victor hugo'
|
|
assert resp.json['address']['postcode'] == '38000'
|
|
assert resp.json['address']['city'] == 'Grenoble'
|
|
|
|
resp = app.get(
|
|
endpoint,
|
|
params={
|
|
'lat': '45.1893469606986',
|
|
'lon': '5.72462060798',
|
|
'addressdetails': 'details',
|
|
'accept-language': 'fr',
|
|
},
|
|
)
|
|
assert not resp.json['err']
|
|
|
|
resp = app.get(endpoint, params={'lat': 'lat=45.1893469606986', 'lon': '5.72462060798'})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Wrong parameters lon, lat: 5.72462060798, lat=45.1893469606986'
|
|
|
|
connector.projection = 'EPSG:4326'
|
|
connector.search_radius = 10
|
|
connector.save()
|
|
mocked_get.return_value = tests.utils.FakedResponse(content='{"features": []}', status_code=200)
|
|
resp = app.get(endpoint, params={'lat': '45.183784', 'lon': '5.714885'})
|
|
assert (
|
|
mocked_get.call_args[1]['params']['cql_filter']
|
|
== 'DWITHIN(the_geom,Point(5.714885 45.183784),10,meters)'
|
|
)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Unable to geocode'
|
|
|
|
mocked_get.return_value = tests.utils.FakedResponse(status_code=404, content='{}', ok=False)
|
|
resp = app.get(endpoint, params={'lat': '45.183784', 'lon': '5.714885'})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Webservice returned status code 404'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_cache_update(mocked_get, app, connector, query):
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
query.update_cache()
|
|
|
|
assert mocked_get.call_args[1]['params']['filter'] == query.filter_expression
|
|
assert mocked_get.call_args[1]['params']['typenames'] == query.typename
|
|
assert FeatureCache.objects.count() == 6
|
|
|
|
feature = FeatureCache.objects.get(lon=1914059.51, lat=4224699.2)
|
|
assert feature.data['properties']['code_post'] == 38000
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_computed_properties(mocked_get, app, connector, query):
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
query.update_cache()
|
|
|
|
assert FeatureCache.objects.count() == 6
|
|
for feature in FeatureCache.objects.all():
|
|
assert 'foobar' not in feature.data['properties']
|
|
assert 'color' not in feature.data['properties']
|
|
|
|
query.computed_properties = {
|
|
'foobar': 'foo',
|
|
'color': '{% if properties.numero %}green{% else %}red{% endif %}',
|
|
}
|
|
query.save()
|
|
query.update_cache()
|
|
|
|
assert FeatureCache.objects.count() == 6
|
|
for feature in FeatureCache.objects.all():
|
|
assert feature.data['properties']['foobar'] == 'foo'
|
|
assert (
|
|
feature.data['properties']['color'] == 'green'
|
|
if feature.data['properties'].get('numero')
|
|
else 'red'
|
|
)
|
|
|
|
query.computed_properties = {
|
|
'foobar': '',
|
|
'color': '{% if properties.numero %}green{% else %}red{% endif }',
|
|
}
|
|
query.save()
|
|
query.update_cache()
|
|
assert FeatureCache.objects.count() == 6
|
|
for feature in FeatureCache.objects.all():
|
|
assert feature.data['properties']['foobar'] == ''
|
|
assert 'color' not in feature.data['properties'] # syntax error
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_q_endpoint(mocked_get, app, connector, query):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'query/test_query/', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/query/test_query/'
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
query.update_cache()
|
|
feature = FeatureCache.objects.get(lon=1914059.51, lat=4224699.2)
|
|
resp = app.get(endpoint)
|
|
|
|
assert len(resp.json['features']) == FeatureCache.objects.count()
|
|
|
|
feature_data = next(
|
|
feature for feature in resp.json['features'] if feature['geometry']['coordinates'][0] == 1914059.51
|
|
)
|
|
assert feature_data == feature.data
|
|
|
|
resp = app.get(endpoint + '?bbox=1914041,4224660,1914060,4224670')
|
|
assert len(resp.json['features']) == 2
|
|
|
|
resp = app.get(endpoint + '?bbox=wrong')
|
|
assert resp.json['err'] == 1
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_cache_update_change(mocked_get, app, connector, query):
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
query.update_cache()
|
|
assert FeatureCache.objects.count() == 6
|
|
|
|
def new_response(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
return tests.utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return tests.utils.FakedResponse(
|
|
content='{"features": [{"properties": {}, "geometry": {"coordinates": [1, 1], "type": "Point"}}]}',
|
|
status_code=200,
|
|
)
|
|
|
|
mocked_get.side_effect = new_response
|
|
query.update_cache()
|
|
assert FeatureCache.objects.count() == 1
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_q_endpoint_cache_empty(mocked_get, app, connector, query):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'query/test_query/', slug=connector.slug)
|
|
assert not FeatureCache.objects.exists()
|
|
resp = app.get(endpoint)
|
|
|
|
assert resp.json['err'] == 1
|
|
assert 'not synchronized' in resp.json['err_desc']
|
|
assert not FeatureCache.objects.exists()
|
|
assert mocked_get.call_count == 0
|
|
assert resp.json['features'] == []
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_cache_update_jobs(mocked_get, app, connector, query):
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
|
|
# fixtures created one query
|
|
job = Job.objects.get(method_name='update_queries')
|
|
assert not FeatureCache.objects.exists()
|
|
|
|
connector.jobs()
|
|
assert FeatureCache.objects.count() == 6
|
|
job.refresh_from_db()
|
|
assert job.status == 'completed'
|
|
|
|
# modifying a query triggers an update
|
|
query.save()
|
|
assert Job.objects.filter(method_name='update_queries', status='registered').count() == 1
|
|
connector.jobs()
|
|
|
|
# modifying the connector triggers an update
|
|
connector.save()
|
|
assert Job.objects.filter(method_name='update_queries', status='registered').count() == 1
|
|
connector.jobs()
|
|
|
|
# two queries to update
|
|
query.save()
|
|
query.pk = None
|
|
query.slug = query.name = 'test2'
|
|
query.save()
|
|
with mock.patch.object(Query, 'update_cache') as mocked:
|
|
connector.jobs()
|
|
assert mocked.call_count == 2
|
|
|
|
# now only one
|
|
query.save()
|
|
with mock.patch.object(Query, 'update_cache') as mocked:
|
|
connector.jobs()
|
|
assert mocked.call_count == 1
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_cache_update_daily(mocked_get, app, connector, query):
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
assert not FeatureCache.objects.exists()
|
|
call_command('cron', 'daily')
|
|
assert FeatureCache.objects.count() == 6
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_endpoint_documentation(mocked_get, app, connector, query):
|
|
resp = app.get(connector.get_absolute_url())
|
|
assert query.name in resp.text
|
|
assert query.description in resp.text
|
|
assert '/opengis/test/query/test_query/' in resp.text
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_text_search(mocked_get, app, connector, query):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'query/test_query/', slug=connector.slug)
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
query.update_cache()
|
|
|
|
resp = app.get(endpoint + '?q=grenoble')
|
|
assert len(resp.json['features']) == 0
|
|
|
|
query.indexing_template = '{{nom_commune}} {{nom_voie}}'
|
|
query.save()
|
|
query.update_cache()
|
|
|
|
resp = app.get(endpoint + '?q=grenoble')
|
|
assert len(resp.json['features']) == 6
|
|
|
|
resp = app.get(endpoint + '?q=victor')
|
|
assert len(resp.json['features']) == 2
|
|
|
|
resp = app.get(endpoint + '?q=hugo victor')
|
|
assert len(resp.json['features']) == 2
|
|
|
|
resp = app.get(endpoint + '?q=nomatch')
|
|
assert len(resp.json['features']) == 0
|
|
|
|
# add an explicit word, that will be found in all features
|
|
query.indexing_template = 'plop {{nom_commune}} {{nom_voie}}'
|
|
query.save()
|
|
query.update_cache()
|
|
|
|
resp = app.get(endpoint + '?q=plop')
|
|
assert len(resp.json['features']) == 6
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_circle_filtering(mocked_get, app, connector, query):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'query/test_query/', slug=connector.slug)
|
|
mocked_get.side_effect = geoserver_circle_responses
|
|
query.update_cache()
|
|
|
|
center_lon = 2.3488000
|
|
center_lat = 48.8534100
|
|
radius = 5000
|
|
|
|
bbox = Query.get_bbox_containing_circle(center_lon, center_lat, float(radius))
|
|
resp = app.get(endpoint + '?bbox=' + ','.join(str(x) for x in bbox))
|
|
features = resp.json['features']
|
|
assert len(features) == 4
|
|
assert all(feature['properties']['in-circle'] or feature['properties']['in-bbox'] for feature in features)
|
|
|
|
resp = app.get(endpoint + '?circle=%s,%s,%s' % (center_lon, center_lat, radius))
|
|
features = resp.json['features']
|
|
assert len(features) == 3
|
|
assert all(feature['properties']['in-circle'] for feature in features)
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_query_data_filter(mocked_get, app, connector, query):
|
|
endpoint = tests.utils.generic_endpoint_url('opengis', 'query/test_query/', slug=connector.slug)
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
query.update_cache()
|
|
|
|
resp = app.get(endpoint + '?property:code_insee=38185')
|
|
assert len(resp.json['features']) == 6
|
|
|
|
resp = app.get(endpoint + '?property:nom_voie=place victor hugo')
|
|
assert len(resp.json['features']) == 2
|
|
|
|
resp = app.get(endpoint + '?property:nom_voie=place victor hugo&property:numero=4')
|
|
assert len(resp.json['features']) == 1
|
|
|
|
boolean_properties = {'properties': {'exists': True, 'present': 'true'}}
|
|
FeatureCache.objects.create(query=query, lat=0, lon=0, data=boolean_properties)
|
|
resp = app.get(endpoint + '?property:exists=true&property:present=true')
|
|
assert len(resp.json['features']) == 1
|
|
|
|
resp = app.get(endpoint + '?text=place victor hugo')
|
|
assert resp.json['err'] == 1
|
|
assert 'extra parameter' in resp.json['err_desc']
|
|
|
|
|
|
def test_opengis_query_unicity(admin_user, app, connector, query):
|
|
connector2 = OpenGIS.objects.create(
|
|
slug='test2', wms_service_url='http://example.net/wms', wfs_service_url='http://example.net/wfs'
|
|
)
|
|
Query.objects.create(
|
|
resource=connector2,
|
|
name='Foo Bar',
|
|
slug='foo-bar',
|
|
)
|
|
|
|
app = login(app)
|
|
resp = app.get('/manage/opengis/%s/query/new/' % connector.slug)
|
|
resp.form['slug'] = query.slug
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp.form['typename'] = 'foo'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 200
|
|
assert Query.objects.filter(resource=connector).count() == 1
|
|
assert 'A query with this slug already exists' in resp.text
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = query.name
|
|
resp.form['typename'] = 'foo'
|
|
resp = resp.form.submit()
|
|
assert Query.objects.filter(resource=connector).count() == 1
|
|
assert resp.status_code == 200
|
|
assert 'A query with this name already exists' in resp.text
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp.form['typename'] = 'foo'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 302
|
|
assert Query.objects.filter(resource=connector).count() == 2
|
|
new_query = Query.objects.latest('pk')
|
|
assert new_query.resource == connector
|
|
|
|
resp = app.get('/manage/opengis/%s/query/%s/' % (connector.slug, new_query.pk))
|
|
resp.form['slug'] = query.slug
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 200
|
|
assert 'A query with this slug already exists' in resp.text
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = query.name
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 200
|
|
assert 'A query with this name already exists' in resp.text
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 302
|
|
|
|
|
|
def test_opengis_query_computed_properties_form(admin_user, app, connector):
|
|
app = login(app)
|
|
resp = app.get('/manage/opengis/%s/query/new/' % connector.slug)
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp.form['typename'] = 'foo'
|
|
resp.form['form-0-key'] = 'foo'
|
|
resp.form['form-0-value'] = 'bar'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 302
|
|
assert Query.objects.filter(resource=connector).count() == 1
|
|
query = Query.objects.latest('pk')
|
|
assert query.computed_properties == {
|
|
'foo': 'bar',
|
|
}
|
|
assert Job.objects.filter(method_name='update_queries', status='registered').count() == 1
|
|
|
|
resp = app.get('/manage/opengis/%s/query/%s/' % (connector.slug, query.pk))
|
|
assert resp.form['form-TOTAL_FORMS'].value == '2'
|
|
assert resp.form['form-0-key'].value == 'foo'
|
|
assert resp.form['form-0-value'].value == 'bar'
|
|
assert resp.form['form-1-key'].value == ''
|
|
assert resp.form['form-1-value'].value == ''
|
|
resp.form['form-0-value'] = 'bar-bis'
|
|
resp.form['form-1-key'] = 'blah'
|
|
resp.form['form-1-value'] = 'baz'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 302
|
|
query.refresh_from_db()
|
|
assert query.computed_properties == {
|
|
'foo': 'bar-bis',
|
|
'blah': 'baz',
|
|
}
|
|
assert Job.objects.filter(method_name='update_queries', status='registered').count() == 2
|
|
|
|
resp = app.get('/manage/opengis/%s/query/%s/' % (connector.slug, query.pk))
|
|
assert resp.form['form-TOTAL_FORMS'].value == '3'
|
|
assert resp.form['form-0-key'].value == 'foo'
|
|
assert resp.form['form-0-value'].value == 'bar-bis'
|
|
assert resp.form['form-1-key'].value == 'blah'
|
|
assert resp.form['form-1-value'].value == 'baz'
|
|
assert resp.form['form-2-key'].value == ''
|
|
assert resp.form['form-2-value'].value == ''
|
|
resp.form['form-0-key'] = 'foo'
|
|
resp.form['form-0-value'] = 'bar'
|
|
resp.form['form-1-key'] = ''
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 302
|
|
query.refresh_from_db()
|
|
assert query.computed_properties == {
|
|
'foo': 'bar',
|
|
}
|
|
|
|
|
|
def test_opengis_export_import(query):
|
|
assert OpenGIS.objects.count() == 1
|
|
assert Query.objects.count() == 1
|
|
serialization = {'resources': [query.resource.export_json()]}
|
|
OpenGIS.objects.all().delete()
|
|
assert OpenGIS.objects.count() == 0
|
|
assert Query.objects.count() == 0
|
|
import_site(serialization)
|
|
assert OpenGIS.objects.count() == 1
|
|
assert Query.objects.count() == 1
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_opengis_test_indexing_template_view(mocked_get, admin_user, app, connector, query):
|
|
url = '/manage/opengis/%s/query/%s/test-indexing-template/' % (connector.slug, query.pk)
|
|
mocked_get.side_effect = geoserver_geolocated_responses
|
|
query.update_cache()
|
|
|
|
app = login(app)
|
|
resp = app.get(url + '?template=')
|
|
assert resp.json['data'] == []
|
|
|
|
resp = app.get(url + '?template={{a}b}}')
|
|
assert resp.json['data'] == []
|
|
|
|
resp = app.get(url + '?template={{nom_commune}} {{nom_voie}}')
|
|
assert resp.json['data'] == [
|
|
'grenoble boulevard edouard rey',
|
|
'grenoble place victor hugo',
|
|
'grenoble boulevard edouard rey',
|
|
]
|
|
|
|
resp = app.get('/manage/opengis/%s/query/%s/' % (connector.slug, query.pk))
|
|
assert url in resp.text
|
|
assert 'Test template' in resp.text
|
|
|
|
|
|
def test_opengis_query_creation_validates_template(admin_user, app, connector):
|
|
app = login(app)
|
|
resp = app.get('/manage/opengis/%s/query/new/' % connector.slug)
|
|
resp.form['slug'] = 'foo'
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp.form['typename'] = 'foo'
|
|
resp.form['indexing_template'] = '{% if %}'
|
|
|
|
resp = resp.form.submit()
|
|
|
|
assert 'Unexpected end of expression in if tag' in str(resp.form.html)
|