forked from Open-EO/openeo-grassgis-driver
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_process_graph_generation.py
113 lines (81 loc) · 4.29 KB
/
test_process_graph_generation.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
# -*- coding: utf-8 -*-
import unittest
from openeo_grass_gis_driver.test_base import TestBase
from openeo_grass_gis_driver.actinia_processing.base import ProcessNode, ProcessGraph
from openeo_grass_gis_driver.utils.process_graph_examples_v04 import OPENEO_EXAMPLE_1, \
FILTER_BBOX, NDVI_STRDS, USE_CASE_1, ZONAL_STATISTICS, DATERANGE
__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 GraphValidationTestCase(TestBase):
def test_graph_creation_openeo_example(self):
pg = ProcessGraph(OPENEO_EXAMPLE_1)
self.assertEqual(2, len(pg.root_nodes))
self.assertEqual(9, len(pg.node_dict))
# print(pg.node_dict["export1"])
# print(pg.node_dict["export2"])
# print(pg.node_dict["filter1"])
# print(pg.node_dict["filter2"])
# print(pg.node_dict["filter3"])
# print(pg.node_dict["getcol1"])
# print(pg.node_dict["mergec1"])
# print(pg.node_dict["reduce1"])
# print(pg.node_dict["reduce2"])
def test_graph_creation_graph_filter_bbox(self):
pg = ProcessGraph(FILTER_BBOX)
# print(pg.node_dict["get_data_1"])
# print(pg.node_dict["filter_bbox_1"])
self.assertEqual(1, len(pg.root_nodes))
self.assertEqual(2, len(pg.node_dict))
self.assertIsNone(pg.node_dict["filter_bbox_1"].child)
self.assertEqual(pg.node_dict["filter_bbox_1"], pg.node_dict["get_data_1"].child)
self.assertEqual(1, len(pg.node_dict["filter_bbox_1"].parents))
self.assertTrue(pg.node_dict["get_data_1"] in pg.node_dict["filter_bbox_1"].parents)
def test_graph_creation_graph_zonal_statistics(self):
pg = ProcessGraph(ZONAL_STATISTICS)
# print(pg.node_dict["zonal_statistics_1"])
# print(pg.node_dict["get_b08_data"])
self.assertEqual(1, len(pg.root_nodes))
self.assertEqual(2, len(pg.node_dict))
self.assertIsNone(pg.node_dict["zonal_statistics_1"].child)
self.assertEqual(pg.node_dict["zonal_statistics_1"], pg.node_dict["get_b08_data"].child)
self.assertEqual(1, len(pg.node_dict["zonal_statistics_1"].parents))
self.assertTrue(pg.node_dict["get_b08_data"] in pg.node_dict["zonal_statistics_1"].parents)
def test_graph_creation_graph_daterange(self):
pg = ProcessGraph(DATERANGE)
# print(pg.node_dict["filter_daterange_1"])
# print(pg.node_dict["get_strds_data"])
self.assertEqual(1, len(pg.root_nodes))
self.assertEqual(2, len(pg.node_dict))
self.assertIsNone(pg.node_dict["filter_daterange_1"].child)
self.assertEqual(pg.node_dict["filter_daterange_1"], pg.node_dict["get_strds_data"].child)
self.assertEqual(1, len(pg.node_dict["filter_daterange_1"].parents))
self.assertTrue(pg.node_dict["get_strds_data"] in pg.node_dict["filter_daterange_1"].parents)
def test_graph_creation_graph_ndvi_strds(self):
pg = ProcessGraph(NDVI_STRDS)
# print(pg.node_dict["get_nir_data"])
# print(pg.node_dict["get_red_data"])
# print(pg.node_dict["ndvi_1"])
self.assertEqual(1, len(pg.root_nodes))
self.assertEqual(3, len(pg.node_dict))
self.assertIsNone(pg.node_dict["ndvi_1"].child)
self.assertEqual(2, len(pg.node_dict["ndvi_1"].parents))
self.assertTrue(pg.node_dict["get_nir_data"] in pg.node_dict["ndvi_1"].parents)
self.assertTrue(pg.node_dict["get_red_data"] in pg.node_dict["ndvi_1"].parents)
self.assertEqual(pg.node_dict["ndvi_1"], pg.node_dict["get_nir_data"].child)
self.assertEqual(pg.node_dict["ndvi_1"], pg.node_dict["get_red_data"].child)
def test_graph_creation_graph_use_case_1(self):
pg = ProcessGraph(USE_CASE_1)
# print(pg.node_dict["get_nir_data"])
# print(pg.node_dict["get_red_data"])
# print(pg.node_dict["filter_bbox_red"])
# print(pg.node_dict["filter_bbox_nir"])
# print(pg.node_dict["ndvi_1"])
# print(pg.node_dict["filter_daterange_ndvi"])
# print(pg.node_dict["reduce_time_1"])
self.assertEqual(1, len(pg.root_nodes))
self.assertEqual(7, len(pg.node_dict))
if __name__ == "__main__":
unittest.main()