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
|
#
# Copyright 2009- ECMWF.
#
# This software is licensed under the terms of the Apache Licence version 2.0
# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
# In applying this licence, ECMWF does not waive the privileges and immunities
# granted to it by virtue of its status as an intergovernmental organisation
# nor does it submit to any jurisdiction.
#
# This test ensures that sorting ecflow objects is correct.
import unittest
from ecflow import *
class Test_sort(unittest.TestCase):
def get_object_list(self, obj, name_list):
if obj == 'Meter':
return [Meter(name, 0, 100) for name in name_list]
elif obj == 'Label':
return [Label(name, 'value') for name in name_list]
elif obj == 'Variable':
return [Variable(name, 'value') for name in name_list]
elif obj == 'Limit':
return [Limit(name, 10) for name in name_list]
else:
return [eval(obj)(name) for name in name_list]
def test_sort_objects(self):
names = ['zz', 'ww', 'aa']
object_list = ['Task', 'Family', 'Suite', 'Event', 'Meter', 'Label', 'Variable', 'Limit']
for obj in object_list:
self.assertEqual(sorted(self.get_object_list(obj, names)), self.get_object_list(obj, sorted(names)), " sort of " + obj + " failed")
class Test_sort_api(unittest.TestCase):
def test_sort_attributes_api(self):
nodes = [Defs(), Suite('s'), Family('f'), Task('t')]
sort_attr = ['event', 'meter', 'label', 'variable', 'all']
for attr in sort_attr:
for node in nodes:
node.sort_attributes(attr)
node.sort_attributes(attr, True)
node.sort_attributes(attr, True, ['/path', '/path2'])
sort_attr = [AttrType.event, AttrType.meter, AttrType.label, AttrType.variable, AttrType.all]
for attr in sort_attr:
for node in nodes:
node.sort_attributes(attr)
node.sort_attributes(attr, True)
node.sort_attributes(attr, True, ['/path', '/path2'])
# see ECFLOW-1642
class Test_ECFLOW_1642(unittest.TestCase):
def get_events(self):
return [Event(name) for name in
"foo buzz bar google zulu mason".split()]
def sorted_events(self):
return [Event(name) for name in
"bar buzz foo google mason zulu".split()]
def expected_def(self):
defs = Defs(Suite("mysuite",
Family("fam1", self.get_events()),
Family("fam2", self.sorted_events(),
Task("task1", self.sorted_events()),
),
Task("task2", self.sorted_events()),
self.sorted_events()))
return defs
def defs_to_sort(self):
defs = Defs(Suite("mysuite",
Family("fam1", self.get_events()),
Family("fam2", self.get_events(),
Task("task1", self.get_events()),
),
Task("task2", self.get_events()),
self.get_events()))
return defs
def test_no_sort(self):
defs = self.defs_to_sort()
defs.sort_attributes('event', True, ['/mysuite/fam1'])
self.assertEqual(defs, self.expected_def(), "Expected\n" + str(self.expected_def()) + "but found\n" + str(defs))
def test_no_sort2(self):
defs = self.defs_to_sort()
defs.sort_attributes(AttrType.event, True, ['/mysuite/fam1'])
self.assertEqual(defs, self.expected_def(), "Expected\n" + str(self.expected_def()) + "but found\n" + str(defs))
class Test_defs_sort(unittest.TestCase):
def test_defs_sort_attributes(self):
defs = Defs()
defs.add_variable("ZFRED", "/tmp/")
defs.add_variable("YECF_URL_CMD", "${BROWSER:=firefox} -new-tab %ECF_URL_BASE%/%ECF_URL%")
defs.add_variable("XECF_URL_BASE", "http://www.ecmwf.int")
defs.add_variable("AECF_URL", "publications/manuals/sms")
assert len(list(defs.user_variables)) == 4, "Expected *user* 4 variable"
# sort
expected = ['AECF_URL', 'XECF_URL_BASE', 'YECF_URL_CMD', 'ZFRED']
actual = []
defs.sort_attributes("variable");
for v in defs.user_variables: actual.append(v.name())
assert actual == expected, "Attributes not sorted, expected:" + str(expected) + " but found:" + str(actual)
expected = ['AECF_URL', 'XECF_URL_BASE', 'YECF_URL_CMD', 'ZFRED', 'ZZ']
actual = []
defs.add_variable("ZZ", "x")
defs.sort_attributes(AttrType.variable);
for v in defs.user_variables: actual.append(v.name())
assert actual == expected, "Attributes not sorted, expected:" + str(expected) + " but found:" + str(actual)
expected = ['AA', 'AECF_URL', 'XECF_URL_BASE', 'YECF_URL_CMD', 'ZFRED', 'ZZ']
actual = []
defs.add_variable("AA", "x")
defs.sort_attributes("all");
for v in defs.user_variables: actual.append(v.name())
assert actual == expected, "Attributes not sorted, expected:" + str(expected) + " but found:" + str(actual)
expected = ['AA', 'AECF_URL', 'BB', 'XECF_URL_BASE', 'YECF_URL_CMD', 'ZFRED', 'ZZ']
actual = []
defs.add_variable("BB", "x")
defs.sort_attributes(AttrType.all);
for v in defs.user_variables: actual.append(v.name())
assert actual == expected, "Attributes not sorted, expected:" + str(expected) + " but found:" + str(actual)
if __name__ == "__main__":
unittest.main()
print("All Tests pass")
|
