4027556941

Build Type

push

github

Committed by GitHub

Commit Message

fixed tox.ini parameter passenv (#670)

Run Details

5172 of 6354 relevant lines covered (81.4%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.09

/tests/test_assync.py

##################################################################
# Copyright 2018 Open Source Geospatial Foundation and others    #
# licensed under MIT, Please consult LICENSE.txt for details     #
##################################################################

import unittest
import pytest
import time
from pywps import Service, configuration
from pywps import get_ElementMakerForVersion
from pywps.tests import client_for, assert_response_accepted, assert_response_success
from .processes import Sleep
from owslib.wps import WPSExecution
from pathlib import Path

VERSION = "1.0.0"

WPS, OWS = get_ElementMakerForVersion(VERSION)


class ExecuteTest(unittest.TestCase):
    def setUp(self) -> None:
        # Running processes using the MultiProcessing scheduler and a file-based database
        configuration.CONFIG.set('processing', 'mode', 'distributed')
        configuration.CONFIG.set("logging", "database", "sqlite:////tmp/test-pywps-logs.sqlite3")

    def tearDown(self) -> None:
        configuration.load_configuration()

    def test_async(self):
        client = client_for(Service(processes=[Sleep()]))
        wps = WPSExecution()

        # Build an asynchronous request (requires specifying outputs and setting the mode).
        doc = wps.buildRequest('sleep',
                               inputs=[('seconds', '.01')],
                               output=[('finished', None, None)],
                               mode='async')

        resp = client.post_xml(doc=doc)
        wps.parseResponse(resp.xml)
        assert_response_accepted(resp)

        # The process should not have finished by now. If it does, it's running in sync mode.
        with pytest.raises(AssertionError):
            assert_response_success(resp)

        # Parse response to extract the status file path
        url = resp.xml.xpath("//@statusLocation")[0]

        # OWSlib only reads from URLs, not local files. So we need to read the response manually.
        p = Path(url[6:])

        # Poll the process until it completes
        total_time = 0
        sleep_time = .01
        while wps.status not in ["ProcessSucceeded", "ProcessFailed"]:
            resp = p.read_bytes()
            if resp:
                wps.checkStatus(response=resp, sleepSecs=0.01)
            else:
                time.sleep(sleep_time)
                total_time += sleep_time
            if total_time > 1:
                raise TimeoutError

        assert wps.status == 'ProcessSucceeded'


def load_tests(loader=None, tests=None, pattern=None):
    if not loader:
        loader = unittest.TestLoader()
    suite_list = [
        loader.loadTestsFromTestCase(ExecuteTest),
    ]
    return unittest.TestSuite(suite_list)

1	##################################################################
2	# Copyright 2018 Open Source Geospatial Foundation and others #
3	# licensed under MIT, Please consult LICENSE.txt for details #
4	##################################################################
5
6	import unittest	1✔
7	import pytest	1✔
8	import time	1✔
9	from pywps import Service, configuration	1✔
10	from pywps import get_ElementMakerForVersion	1✔
11	from pywps.tests import client_for, assert_response_accepted, assert_response_success	1✔
12	from .processes import Sleep	1✔
13	from owslib.wps import WPSExecution	1✔
14	from pathlib import Path	1✔
15
16	VERSION = "1.0.0"	1✔
17
18	WPS, OWS = get_ElementMakerForVersion(VERSION)	1✔
19
20
21	class ExecuteTest(unittest.TestCase):	1✔
22	def setUp(self) -> None:	1✔
23	# Running processes using the MultiProcessing scheduler and a file-based database
24	configuration.CONFIG.set('processing', 'mode', 'distributed')	1✔
25	configuration.CONFIG.set("logging", "database", "sqlite:////tmp/test-pywps-logs.sqlite3")	1✔
26
27	def tearDown(self) -> None:	1✔
28	configuration.load_configuration()	1✔
29
30	def test_async(self):	1✔
31	client = client_for(Service(processes=[Sleep()]))	1✔
32	wps = WPSExecution()	1✔
33
34	# Build an asynchronous request (requires specifying outputs and setting the mode).
35	doc = wps.buildRequest('sleep',	1✔
36	inputs=[('seconds', '.01')],
37	output=[('finished', None, None)],
38	mode='async')
39
40	resp = client.post_xml(doc=doc)	1✔
41	wps.parseResponse(resp.xml)	1✔
42	assert_response_accepted(resp)	1✔
43
44	# The process should not have finished by now. If it does, it's running in sync mode.
45	with pytest.raises(AssertionError):	1✔
46	assert_response_success(resp)	1✔
47
48	# Parse response to extract the status file path
49	url = resp.xml.xpath("//@statusLocation")[0]	1✔
50
51	# OWSlib only reads from URLs, not local files. So we need to read the response manually.
52	p = Path(url[6:])	1✔
53
54	# Poll the process until it completes
55	total_time = 0	1✔
56	sleep_time = .01	1✔
57	while wps.status not in ["ProcessSucceeded", "ProcessFailed"]:	1✔
58	resp = p.read_bytes()	1✔
59	if resp:	1✔
60	wps.checkStatus(response=resp, sleepSecs=0.01)	1✔
61	else:
62	time.sleep(sleep_time)	×
63	total_time += sleep_time	×
64	if total_time > 1:	1✔
65	raise TimeoutError	×
66
67	assert wps.status == 'ProcessSucceeded'	1✔
68
69
70	def load_tests(loader=None, tests=None, pattern=None):	1✔
71	if not loader:	×
72	loader = unittest.TestLoader()	×
73	suite_list = [	×
74	loader.loadTestsFromTestCase(ExecuteTest),
75	]
76	return unittest.TestSuite(suite_list)	×

geopython / pywps / 4027556941

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous