1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
##################################################################
# 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)
|
