|
| 1 | +################################################################## |
| 2 | +# Copyright 2018 Open Source Geospatial Foundation and others # |
| 3 | +# licensed under MIT, Please consult LICENSE.txt for details # |
| 4 | +################################################################## |
| 5 | + |
| 6 | +from basic import TestBase |
| 7 | +from pywps import Service |
| 8 | +from pywps import get_ElementMakerForVersion |
| 9 | +from pywps.tests import client_for, assert_response_accepted, assert_response_success |
| 10 | + |
| 11 | +from pywps import configuration |
| 12 | +from processes import Greeter |
| 13 | +from urllib.parse import urlparse |
| 14 | + |
| 15 | +from time import sleep |
| 16 | + |
| 17 | +VERSION = "1.0.0" |
| 18 | + |
| 19 | +WPS, OWS = get_ElementMakerForVersion(VERSION) |
| 20 | + |
| 21 | + |
| 22 | +class ExecuteTest(TestBase): |
| 23 | + |
| 24 | + def setUp(self) -> None: |
| 25 | + super().setUp() |
| 26 | + # Use fake async processing for this test |
| 27 | + configuration.CONFIG.set('processing', 'mode', 'noasyncprocessing') |
| 28 | + |
| 29 | + def test_wps_status(self): |
| 30 | + client = client_for(Service(processes=[Greeter()])) |
| 31 | + request_doc = WPS.Execute( |
| 32 | + OWS.Identifier('greeter'), |
| 33 | + WPS.DataInputs( |
| 34 | + WPS.Input( |
| 35 | + OWS.Identifier('name'), |
| 36 | + WPS.Data( |
| 37 | + WPS.LiteralData( |
| 38 | + "SomeName" |
| 39 | + ) |
| 40 | + ) |
| 41 | + ) |
| 42 | + ), |
| 43 | + WPS.ResponseForm( |
| 44 | + WPS.ResponseDocument(storeExecuteResponse='true', status='true') |
| 45 | + ), |
| 46 | + version="1.0.0" |
| 47 | + ) |
| 48 | + resp = client.post_xml(doc=request_doc) |
| 49 | + assert_response_accepted(resp) |
| 50 | + |
| 51 | + url = resp.xml.xpath("//@statusLocation")[0] |
| 52 | + |
| 53 | + # Parse url because we do not have real server |
| 54 | + url = urlparse(url) |
| 55 | + resp = client.get(f'{url.path}?{url.query}') |
| 56 | + assert_response_success(resp) |
| 57 | + |
0 commit comments