6720056456

Committed 01 Nov 2023 01:06PM UTC coverage: 84.061% (+2.7%) from 81.332%

Build # 6720056456

Build Type

push

github

Committed by

web-flow

Commit Message

fix tests (#687)



* skip py37

Run Details

5374 of 6393 relevant lines covered (84.06%)

0.84 hits per line

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

94.68

/tests/test_processing.py

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

"""Unit tests for processing
"""

from basic import TestBase

import json
import uuid

from pywps import configuration
import pywps.processing
from pywps.processing.job import Job
from pywps.processing.basic import MultiProcessing
from pywps.app import WPSRequest
from pywps.response.execute import ExecuteResponse

from processes import Greeter, InOut, BBox


class GreeterProcessingTest(TestBase):
    """Processing test case with Greeter process"""

    def setUp(self):
        super().setUp()

        self.workdir = pywps.configuration.get_config_value('server', 'workdir')

        self.uuid = uuid.uuid1()
        self.dummy_process = Greeter()
        self.dummy_process._set_uuid(self.uuid)
        self.dummy_process.set_workdir(self.workdir)
        self.wps_request = WPSRequest()
        self.wps_response = ExecuteResponse(self.wps_request, self.uuid,
                                            process=self.dummy_process)
        self.job = Job(
            process=self.dummy_process,
            wps_request=self.wps_request,
            wps_response=self.wps_response)

    def test_default_mode(self):
        """Test pywps.formats.Format class
        """
        self.assertEqual(configuration.get_config_value('processing', 'mode'),
                         'default')
        process = pywps.processing.Process(
            process=self.dummy_process,
            wps_request=self.wps_request,
            wps_response=self.wps_response)
        # process.start()
        self.assertTrue(isinstance(process, MultiProcessing))

    def test_job_json(self):
        new_job = Job.from_json(json.loads(self.job.json))
        self.assertEqual(new_job.name, 'greeter')
        self.assertEqual(new_job.uuid, str(self.uuid))
        self.assertEqual(new_job.workdir, self.workdir)
        self.assertEqual(len(new_job.process.inputs), 1)

    def test_job_dump(self):
        new_job = Job.load(self.job.dump())
        self.assertEqual(new_job.name, 'greeter')
        self.assertEqual(new_job.uuid, str(self.uuid))
        self.assertEqual(new_job.workdir, self.workdir)
        self.assertEqual(len(new_job.process.inputs), 1)


class InOutProcessingTest(TestBase):
    """Processing test case with InOut process"""

    def setUp(self):
        super().setUp()

        self.workdir = pywps.configuration.get_config_value('server', 'workdir')

        self.uuid = uuid.uuid1()
        self.dummy_process = InOut()
        self.dummy_process._set_uuid(self.uuid)
        self.dummy_process.set_workdir(self.workdir)
        self.wps_request = WPSRequest()
        self.wps_response = ExecuteResponse(self.wps_request, self.uuid,
                                            process=self.dummy_process)
        self.job = Job(
            process=self.dummy_process,
            wps_request=self.wps_request,
            wps_response=self.wps_response)

    def test_job_json(self):
        new_job = Job.from_json(json.loads(self.job.json))
        self.assertEqual(new_job.name, 'inout')
        self.assertEqual(new_job.uuid, str(self.uuid))
        self.assertEqual(new_job.workdir, self.workdir)
        self.assertEqual(len(new_job.process.inputs), 3)
        self.assertEqual(new_job.json, self.job.json)  # idempotent test

    def test_job_dump(self):
        new_job = Job.load(self.job.dump())
        self.assertEqual(new_job.name, 'inout')
        self.assertEqual(new_job.uuid, str(self.uuid))
        self.assertEqual(new_job.workdir, self.workdir)
        self.assertEqual(len(new_job.process.inputs), 3)
        self.assertEqual(new_job.json, self.job.json)  # idempotent test


class BBoxProcessingTest(TestBase):
    """Processing test case with BBox input and output process"""

    def setUp(self):
        super().setUp()

        self.workdir = pywps.configuration.get_config_value('server', 'workdir')

        self.uuid = uuid.uuid1()
        self.dummy_process = BBox()
        self.dummy_process._set_uuid(self.uuid)
        self.dummy_process.set_workdir(self.workdir)
        self.wps_request = WPSRequest()
        self.wps_response = ExecuteResponse(self.wps_request, self.uuid,
                                            process=self.dummy_process)
        self.job = Job(
            process=self.dummy_process,
            wps_request=self.wps_request,
            wps_response=self.wps_response)

    def test_job_json(self):
        new_job = Job.from_json(json.loads(self.job.json))
        self.assertEqual(new_job.name, 'bbox_test')
        self.assertEqual(new_job.uuid, str(self.uuid))
        self.assertEqual(new_job.workdir, self.workdir)
        self.assertEqual(len(new_job.process.inputs), 1)
        self.assertEqual(new_job.json, self.job.json)  # idempotent test

    def test_job_dump(self):
        new_job = Job.load(self.job.dump())
        self.assertEqual(new_job.name, 'bbox_test')
        self.assertEqual(new_job.uuid, str(self.uuid))
        self.assertEqual(new_job.workdir, self.workdir)
        self.assertEqual(len(new_job.process.inputs), 1)
        self.assertEqual(new_job.json, self.job.json)  # idempotent test


def load_tests(loader=None, tests=None, pattern=None):
    """Load local tests
    """
    import unittest

    if not loader:
        loader = unittest.TestLoader()
    suite_list = [
        loader.loadTestsFromTestCase(GreeterProcessingTest),
        loader.loadTestsFromTestCase(InOutProcessingTest),
        loader.loadTestsFromTestCase(BBoxProcessingTest)
    ]
    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	"""Unit tests for processing	1✔
7	"""
8
9	from basic import TestBase	1✔
10
11	import json	1✔
12	import uuid	1✔
13
14	from pywps import configuration	1✔
15	import pywps.processing	1✔
16	from pywps.processing.job import Job	1✔
17	from pywps.processing.basic import MultiProcessing	1✔
18	from pywps.app import WPSRequest	1✔
19	from pywps.response.execute import ExecuteResponse	1✔
20
21	from processes import Greeter, InOut, BBox	1✔
22
23
24	class GreeterProcessingTest(TestBase):	1✔
25	"""Processing test case with Greeter process"""
26
27	def setUp(self):	1✔
28	super().setUp()	1✔
29
30	self.workdir = pywps.configuration.get_config_value('server', 'workdir')	1✔
31
32	self.uuid = uuid.uuid1()	1✔
33	self.dummy_process = Greeter()	1✔
34	self.dummy_process._set_uuid(self.uuid)	1✔
35	self.dummy_process.set_workdir(self.workdir)	1✔
36	self.wps_request = WPSRequest()	1✔
37	self.wps_response = ExecuteResponse(self.wps_request, self.uuid,	1✔
38	process=self.dummy_process)
39	self.job = Job(	1✔
40	process=self.dummy_process,
41	wps_request=self.wps_request,
42	wps_response=self.wps_response)
43
44	def test_default_mode(self):	1✔
45	"""Test pywps.formats.Format class
46	"""
47	self.assertEqual(configuration.get_config_value('processing', 'mode'),	1✔
48	'default')
49	process = pywps.processing.Process(	1✔
50	process=self.dummy_process,
51	wps_request=self.wps_request,
52	wps_response=self.wps_response)
53	# process.start()
54	self.assertTrue(isinstance(process, MultiProcessing))	1✔
55
56	def test_job_json(self):	1✔
57	new_job = Job.from_json(json.loads(self.job.json))	1✔
58	self.assertEqual(new_job.name, 'greeter')	1✔
59	self.assertEqual(new_job.uuid, str(self.uuid))	1✔
60	self.assertEqual(new_job.workdir, self.workdir)	1✔
61	self.assertEqual(len(new_job.process.inputs), 1)	1✔
62
63	def test_job_dump(self):	1✔
64	new_job = Job.load(self.job.dump())	1✔
65	self.assertEqual(new_job.name, 'greeter')	1✔
66	self.assertEqual(new_job.uuid, str(self.uuid))	1✔
67	self.assertEqual(new_job.workdir, self.workdir)	1✔
68	self.assertEqual(len(new_job.process.inputs), 1)	1✔
69
70
71	class InOutProcessingTest(TestBase):	1✔
72	"""Processing test case with InOut process"""
73
74	def setUp(self):	1✔
75	super().setUp()	1✔
76
77	self.workdir = pywps.configuration.get_config_value('server', 'workdir')	1✔
78
79	self.uuid = uuid.uuid1()	1✔
80	self.dummy_process = InOut()	1✔
81	self.dummy_process._set_uuid(self.uuid)	1✔
82	self.dummy_process.set_workdir(self.workdir)	1✔
83	self.wps_request = WPSRequest()	1✔
84	self.wps_response = ExecuteResponse(self.wps_request, self.uuid,	1✔
85	process=self.dummy_process)
86	self.job = Job(	1✔
87	process=self.dummy_process,
88	wps_request=self.wps_request,
89	wps_response=self.wps_response)
90
91	def test_job_json(self):	1✔
92	new_job = Job.from_json(json.loads(self.job.json))	1✔
93	self.assertEqual(new_job.name, 'inout')	1✔
94	self.assertEqual(new_job.uuid, str(self.uuid))	1✔
95	self.assertEqual(new_job.workdir, self.workdir)	1✔
96	self.assertEqual(len(new_job.process.inputs), 3)	1✔
97	self.assertEqual(new_job.json, self.job.json) # idempotent test	1✔
98
99	def test_job_dump(self):	1✔
100	new_job = Job.load(self.job.dump())	1✔
101	self.assertEqual(new_job.name, 'inout')	1✔
102	self.assertEqual(new_job.uuid, str(self.uuid))	1✔
103	self.assertEqual(new_job.workdir, self.workdir)	1✔
104	self.assertEqual(len(new_job.process.inputs), 3)	1✔
105	self.assertEqual(new_job.json, self.job.json) # idempotent test	1✔
106
107
108	class BBoxProcessingTest(TestBase):	1✔
109	"""Processing test case with BBox input and output process"""
110
111	def setUp(self):	1✔
112	super().setUp()	1✔
113
114	self.workdir = pywps.configuration.get_config_value('server', 'workdir')	1✔
115
116	self.uuid = uuid.uuid1()	1✔
117	self.dummy_process = BBox()	1✔
118	self.dummy_process._set_uuid(self.uuid)	1✔
119	self.dummy_process.set_workdir(self.workdir)	1✔
120	self.wps_request = WPSRequest()	1✔
121	self.wps_response = ExecuteResponse(self.wps_request, self.uuid,	1✔
122	process=self.dummy_process)
123	self.job = Job(	1✔
124	process=self.dummy_process,
125	wps_request=self.wps_request,
126	wps_response=self.wps_response)
127
128	def test_job_json(self):	1✔
129	new_job = Job.from_json(json.loads(self.job.json))	1✔
130	self.assertEqual(new_job.name, 'bbox_test')	1✔
131	self.assertEqual(new_job.uuid, str(self.uuid))	1✔
132	self.assertEqual(new_job.workdir, self.workdir)	1✔
133	self.assertEqual(len(new_job.process.inputs), 1)	1✔
134	self.assertEqual(new_job.json, self.job.json) # idempotent test	1✔
135
136	def test_job_dump(self):	1✔
137	new_job = Job.load(self.job.dump())	1✔
138	self.assertEqual(new_job.name, 'bbox_test')	1✔
139	self.assertEqual(new_job.uuid, str(self.uuid))	1✔
140	self.assertEqual(new_job.workdir, self.workdir)	1✔
141	self.assertEqual(len(new_job.process.inputs), 1)	1✔
142	self.assertEqual(new_job.json, self.job.json) # idempotent test	1✔
143
144
145	def load_tests(loader=None, tests=None, pattern=None):	1✔
146	"""Load local tests
147	"""
148	import unittest	×
149
150	if not loader:	×
151	loader = unittest.TestLoader()	×
152	suite_list = [	×
153	loader.loadTestsFromTestCase(GreeterProcessingTest),
154	loader.loadTestsFromTestCase(InOutProcessingTest),
155	loader.loadTestsFromTestCase(BBoxProcessingTest)
156	]
157	return unittest.TestSuite(suite_list)	×

geopython / pywps / 6720056456

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