forked from Open-EO/openeo-grassgis-driver
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_actinia_interface.py
134 lines (105 loc) · 4.57 KB
/
test_actinia_interface.py
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
# -*- coding: utf-8 -*-
import unittest
from openeo_grass_gis_driver.actinia_processing.actinia_interface import ActiniaInterface
from openeo_grass_gis_driver.test_base import TestBase
from pprint import pprint
import time
__license__ = "Apache License, Version 2.0"
__author__ = "Sören Gebbert"
__copyright__ = "Copyright 2018, Sören Gebbert, mundialis"
__maintainer__ = "Soeren Gebbert"
__email__ = "soerengebbert@googlemail.com"
class ActiniaInterfaceTestCase(TestBase):
def test_health_check(self):
iface = ActiniaInterface(self.gconf)
self.assertTrue(iface.check_health())
def test_list_raster(self):
iface = ActiniaInterface(self.gconf)
status, layers = iface.list_raster(location="nc_spm_08", mapset="PERMANENT")
pprint(layers)
self.assertEqual(status, 200)
self.assertEqual(len(layers), 41)
def test_list_vector(self):
iface = ActiniaInterface(self.gconf)
status, layers = iface.list_vector(location="nc_spm_08", mapset="PERMANENT")
pprint(layers)
self.assertEqual(status, 200)
self.assertEqual(len(layers), 46)
def test_list_strds(self):
iface = ActiniaInterface(self.gconf)
status, layers = iface.list_strds(location="latlong_wgs84", mapset="modis_ndvi_global")
pprint(layers)
self.assertEqual(status, 200)
self.assertEqual(len(layers), 1)
def test_strds_info(self):
iface = ActiniaInterface(self.gconf)
status, info = iface.layer_info(layer_name="latlong_wgs84.modis_ndvi_global.strds.ndvi_16_5600m")
pprint(info)
self.assertEqual(status, 200)
self.assertTrue("temporal_type" in info)
self.assertTrue("aggregation_type" in info)
self.assertTrue("creation_time" in info)
self.assertTrue("creator" in info)
self.assertTrue("granularity" in info)
self.assertTrue("modification_time" in info)
self.assertTrue("number_of_maps" in info)
def test_mapset_info(self):
iface = ActiniaInterface(self.gconf)
status, info = iface.mapset_info(location="latlong_wgs84", mapset="modis_ndvi_global")
pprint(info)
self.assertEqual(status, 200)
self.assertTrue("region" in info)
self.assertTrue("projection" in info)
def test_list_mapsets(self):
iface = ActiniaInterface(self.gconf)
status, mapsets = iface.list_mapsets(location="latlong_wgs84")
pprint(mapsets)
self.assertEqual(status, 200)
def test_layer_exists_1(self):
iface = ActiniaInterface(self.gconf)
status = iface.check_layer_exists(layer_name="latlong_wgs84.modis_ndvi_global.strds.ndvi_16_5600m")
self.assertTrue(status)
def test_layer_exists_2(self):
iface = ActiniaInterface(self.gconf)
status = iface.check_layer_exists(layer_name="latlong_wgs84.modis_ndvi_global.strds.ndvi_16_5600m")
self.assertTrue(status)
def test_layer_exists_2_error(self):
iface = ActiniaInterface(self.gconf)
status = iface.check_layer_exists(layer_name="latlong_wgs84.modis_ndvi_global.strds.ndvi_16_5600m_nope")
self.assertFalse(status)
def test_async_persistent_processing(self):
iface = ActiniaInterface(self.gconf)
process_chain = {
"version": "1",
"list": [
{"id": "g_region_1",
"module": "g.region",
"flags": "g"}]}
status, response = iface.async_persistent_processing(location="nc_spm_08",
mapset="new_user_mapset",
process_chain=process_chain)
resource_id = response["resource_id"]
print(status)
print(resource_id)
self.assertEqual(status, 200)
status, info = iface.resource_info(resource_id)
print(status)
print(info)
time.sleep(2)
status, info = iface.resource_info(resource_id)
print(status)
print(info)
self.assertEqual(info["status"], "finished")
def disfunc_test_mapset_creation_deletion(self):
config = self.gconf
iface = ActiniaInterface(config)
status, response = iface.create_mapset(location="nc_spm_08", mapset="new_mapset")
print(status)
self.assertEqual(status, 200)
print(response)
status, response = iface.delete_mapset(location="nc_spm_08", mapset="new_mapset")
print(status)
self.assertEqual(status, 200)
print(response)
if __name__ == "__main__":
unittest.main()