Completed
Pull Request — master (#185)
by Olivier
05:08
created

HistoryCommon   A

Complexity

Total Complexity 22

Size/Duplication

Total Lines 160
Duplicated Lines 8.13 %

Test Coverage

Coverage 100%

Importance

Changes 6
Bugs 2 Features 2
Metric Value
c 6
b 2
f 2
dl 13
loc 160
ccs 117
cts 117
cp 1
rs 10
wmc 22

How to fix   Duplicated Code   

Duplicated Code

Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.

Common duplication problems, and corresponding solutions are:

1
import time
2
from datetime import datetime, timedelta
3
import unittest
4
5
from opcua import Client
6
from opcua import Server
7
from opcua import ua
8
9
from opcua.server.history_sql import HistorySQLite
10
11
from tests_common import CommonTests, add_server_methods
12
13
from opcua.common.subscription import get_event_properties_from_type_node as get_props
14
15
port_num1 = 48530
16
port_num2 = 48530
17
18
19
20
class HistoryCommon(object):
21
    srv = Server
22
    clt = Client
23
24
    @classmethod
25
    def start_server_and_client(cls):
26
        cls.srv = Server()
27
        cls.srv.set_endpoint('opc.tcp://localhost:%d' % port_num1)
28
        cls.srv.start()
29
30
        cls.clt = Client('opc.tcp://localhost:%d' % port_num1)
31
        cls.clt.connect()
32
33
    @classmethod
34
    def stop_server_and_client(cls):
35
        cls.clt.disconnect()
36
        cls.srv.stop()
37
38
    @classmethod
39
    def create_var(cls):
40
        o = cls.srv.get_objects_node()
41
        cls.values = [i for i in range(20)]
42
        cls.var = o.add_variable(3, "history_var", 0)
43
        cls.srv.iserver.enable_history_data_change(cls.var, period=None, count=10)
44
        for i in cls.values:
45
            cls.var.set_value(i)
46
        time.sleep(1)
47
48
    @classmethod
49
    def create_srv_events(cls):
50
        cls.ev_values = [i for i in range(20)]
51
        cls.srvevgen = cls.srv.get_event_generator()
52
53
        # for now we need to build our own filter explicitly because node class can't do it
54
        cls.evfilter = ua.EventFilter()
55
        ev_type_node = cls.srv.get_node(cls.srvevgen.event.EventType)
56
        for ev_property in get_props(ev_type_node):
57
            op = ua.SimpleAttributeOperand()
58
            op.TypeDefinitionId = ev_type_node.nodeid
59
            op.AttributeId = ua.AttributeIds.Value
60
            op.BrowsePath = [ev_property.get_browse_name()]
61
            cls.evfilter.SelectClauses.append(op)
62
63
        cls.revevfilter = ua.EventFilter()
64
        for ev_property in reversed(get_props(ev_type_node)):
65
            op = ua.SimpleAttributeOperand()
66
            op.TypeDefinitionId = ev_type_node.nodeid
67
            op.AttributeId = ua.AttributeIds.Value
68
            op.BrowsePath = [ev_property.get_browse_name()]
69
            cls.revevfilter.SelectClauses.append(op)
70
71
        cls.srv_node = cls.srv.get_node(ua.ObjectIds.Server)
72
        cls.srv.iserver.enable_history_event(cls.srv_node, period=None)
73
74
        for i in cls.ev_values:
75
            cls.srvevgen.event.Severity = cls.ev_values[i]
76
            cls.srvevgen.trigger(message="test message")
77
            time.sleep(.1)
78
        time.sleep(2)
79
80
    # no start and no end is not defined by spec, return reverse order
81
    def test_history_var_read_one(self):
82
        # Spec says that at least two parameters should be provided, so
83
        # this one is out of spec
84
        res = self.var.read_raw_history(None, None, 1)
85
        self.assertEqual(len(res), 1)
86
        self.assertEqual(res[0].Value.Value, self.values[-1])
87
88
    # no start and no end is not defined by spec, return reverse order
89
    def test_history_var_read_none(self):
90
        res = self.var.read_raw_history(None, None, 0)
91
        self.assertEqual(len(res), 20)
92
        self.assertEqual(res[0].Value.Value, self.values[-1])
93
        self.assertEqual(res[-1].Value.Value, self.values[0])
94
95
    # no start and no end is not defined by spec, return reverse order
96
    def test_history_var_read_last_3(self):
97
        res = self.var.read_raw_history(None, None, 3)
98
        self.assertEqual(len(res), 3)
99
        self.assertEqual(res[-1].Value.Value, self.values[-3])
100
        self.assertEqual(res[0].Value.Value, self.values[-1])
101
102
    # no start and no end is not defined by spec, return reverse order
103
    def test_history_var_read_all2(self):
104
        res = self.var.read_raw_history(None, None, 9999)
105
        self.assertEqual(len(res), 20)
106
        self.assertEqual(res[-1].Value.Value, self.values[0])
107
        self.assertEqual(res[0].Value.Value, self.values[-1])
108
109
    # only has end time, should return reverse order
110
    def test_history_var_read_2_with_end(self):
111
        now = datetime.utcnow()
112
        old = now - timedelta(days=6)
113
114
        res = self.var.read_raw_history(None, now, 2)
115
        self.assertEqual(len(res), 2)
116
        self.assertEqual(res[-1].Value.Value, self.values[-2])
117
    
118
    # both start and endtime, return from start to end
119
    def test_history_var_read_all(self):
120
        now = datetime.utcnow()
121
        old = now - timedelta(days=6)
122
123
        res = self.var.read_raw_history(old, now, 0)
124
        self.assertEqual(len(res), 20)
125
        self.assertEqual(res[-1].Value.Value, self.values[-1])
126
        self.assertEqual(res[0].Value.Value, self.values[0])
127
128
    def test_history_var_read_5_in_timeframe(self):
129
        now = datetime.utcnow()
130
        old = now - timedelta(days=6)
131
132
        res = self.var.read_raw_history(old, now, 5)
133
        self.assertEqual(len(res), 5)
134
        self.assertEqual(res[-1].Value.Value, self.values[4])
135
        self.assertEqual(res[0].Value.Value, self.values[0])
136
137
    # start time greater than end time, should return reverse order
138
    def test_history_var_read_5_in_timeframe_start_greater_than_end(self):
139
        now = datetime.utcnow()
140
        old = now - timedelta(days=6)
141
142
        res = self.var.read_raw_history(now, old, 5)
143
        self.assertEqual(len(res), 5)
144
        self.assertEqual(res[-1].Value.Value, self.values[-5])
145
        self.assertEqual(res[0].Value.Value, self.values[-1])
146
    
147
    # only start return original order
148
    def test_history_var_read_6_with_start(self):
149
        now = datetime.utcnow()
150
        old = now - timedelta(days=6)
151
        res = self.var.read_raw_history(old, None, 6)
152
        self.assertEqual(len(res), 6)
153
        self.assertEqual(res[-1].Value.Value, self.values[5])
154
        self.assertEqual(res[0].Value.Value, self.values[0])
155
156
    # only start return original order
157 View Code Duplication
    def test_history_var_read_all_with_start(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
158
        now = datetime.utcnow()
159
        old = now - timedelta(days=6)
160
        res = self.var.read_raw_history(old, None, 0)
161
        self.assertEqual(len(res), 20)
162
        self.assertEqual(res[-1].Value.Value, self.values[-1])
163
        self.assertEqual(res[0].Value.Value, self.values[0])
164
165
    # only end return reversed order
166 View Code Duplication
    def test_history_var_read_all_with_end(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
167
        end = datetime.utcnow() + timedelta(days=6)
168
        res = self.var.read_raw_history(None, end, 0)
169
        self.assertEqual(len(res), 20)
170
        self.assertEqual(res[-1].Value.Value, self.values[0])
171
        self.assertEqual(res[0].Value.Value, self.values[-1])
172
173
    # only end return reversed order
174
    def test_history_var_read_3_with_end(self):
175
        end = datetime.utcnow() + timedelta(days=6)
176
        res = self.var.read_raw_history(None, end, 3)
177
        self.assertEqual(len(res), 3)
178
        self.assertEqual(res[2].Value.Value, self.values[-3])
179
        self.assertEqual(res[0].Value.Value, self.values[-1])
180
181
182
class TestHistory(unittest.TestCase, HistoryCommon):
183
184
    @classmethod
185
    def setUpClass(cls):
186
        cls.start_server_and_client()
187
        cls.create_var()
188
189
    @classmethod
190
    def tearDownClass(cls):
191
        cls.stop_server_and_client()
192
193
194
class TestHistorySQL(unittest.TestCase, HistoryCommon):
195
    @classmethod
196
    def setUpClass(cls):
197
        cls.start_server_and_client()
198
        cls.srv.iserver.history_manager.set_storage(HistorySQLite(":memory:"))
199
        cls.create_var()
200
        cls.create_srv_events()
201
202
    # ~~~~~~~~~~~~~~~~~~~~~~~ events ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
203
204
    # only has end time, should return reverse order
205
    def test_history_ev_read_2_with_end(self):
206
        now = datetime.utcnow()
207
        old = now - timedelta(days=6)
208
209
        res = self.srv_node.read_event_history(self.evfilter, None, now, 2)
210
        self.assertEqual(len(res), 2)
211
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-2])
212
213
    # both start and end time, return from start to end
214 View Code Duplication
    def test_history_ev_read_all(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
215
        now = datetime.utcnow()
216
        old = now - timedelta(days=6)
217
218
        res = self.srv_node.read_event_history(self.evfilter, old, now, 0)
219
        self.assertEqual(len(res), 20)
220
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-1])
221
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])
222
223 View Code Duplication
    def test_history_ev_read_5_in_timeframe(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
224
        now = datetime.utcnow()
225
        old = now - timedelta(days=6)
226
227
        res = self.srv_node.read_event_history(self.evfilter, old, now, 5)
228
        self.assertEqual(len(res), 5)
229
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[4])
230
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])
231
232
    # start time greater than end time, should return reverse order
233 View Code Duplication
    def test_history_ev_read_5_in_timeframe_start_greater_than_end(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
234
        now = datetime.utcnow()
235
        old = now - timedelta(days=6)
236
237
        res = self.srv_node.read_event_history(self.evfilter, now, old, 5)
238
        self.assertEqual(len(res), 5)
239
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-5])
240
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[-1])
241
242
    # only start return original order
243 View Code Duplication
    def test_history_ev_read_6_with_start(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
244
        now = datetime.utcnow()
245
        old = now - timedelta(days=6)
246
        res = self.srv_node.read_event_history(self.evfilter, old, None, 6)
247
        self.assertEqual(len(res), 6)
248
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[5])
249
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])
250
251
    # only start return original order
252 View Code Duplication
    def test_history_ev_read_all_with_start(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
253
        now = datetime.utcnow()
254
        old = now - timedelta(days=6)
255
        res = self.srv_node.read_event_history(self.evfilter, old, None, 0)
256
        self.assertEqual(len(res), 20)
257
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-1])
258
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])
259
260
    # only end return reversed order
261 View Code Duplication
    def test_history_ev_read_all_with_end(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
262
        end = datetime.utcnow() + timedelta(days=6)
263
        res = self.srv_node.read_event_history(self.evfilter, None, end, 0)
264
        self.assertEqual(len(res), 20)
265
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[0])
266
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[-1])
267
268
    # only end return reversed order
269 View Code Duplication
    def test_history_ev_read_3_with_end(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
270
        end = datetime.utcnow() + timedelta(days=6)
271
        res = self.srv_node.read_event_history(self.evfilter, None, end, 3)
272
        self.assertEqual(len(res), 3)
273
        self.assertEqual(res[2].EventFields[8].Value, self.ev_values[-3])
274
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[-1])
275
276
    # reverse event filter select clauses and test that results match the filter order
277 View Code Duplication
    def test_history_ev_read_all_filter_order_reversed(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
278
        now = datetime.utcnow()
279
        old = now - timedelta(days=6)
280
        res = self.srv_node.read_event_history(self.revevfilter, old, None, 0)
281
        self.assertEqual(len(res), 20)
282
        self.assertEqual(res[-1].EventFields[0].Value, self.ev_values[-1])
283
        self.assertEqual(res[0].EventFields[0].Value, self.ev_values[0])
284
285
    @classmethod
286
    def tearDownClass(cls):
287
        cls.stop_server_and_client()
288
289
290
291
292
293
294
295
296
297
298