HistoryCommon.test_history_read_last_3() - Code Metrics - Inspection of "Merge pull request #180 from zerox1212/event_histo..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d538b5...98e417 )

by Olivier

created 2016-05-10 08:41 UTC

HistoryCommon.test_history_read_last_3() A

↳ Parent: HistoryCommon.create_srv_events()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	1

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	5
ccs	5
cts	5
cp	1
crap	1
rs	9.4285

import time
from datetime import datetime, timedelta
import unittest

from opcua import Client
from opcua import Server
from opcua import ua

from opcua.server.history_sql import HistorySQLite

from tests_common import CommonTests, add_server_methods

from opcua.common.subscription import get_event_properties_from_type_node as get_props

port_num1 = 48530
port_num2 = 48530



class HistoryCommon(object):
    srv = Server
    clt = Client

    @classmethod
    def start_server_and_client(cls):
        cls.srv = Server()
        cls.srv.set_endpoint('opc.tcp://localhost:%d' % port_num1)
        cls.srv.start()

        cls.clt = Client('opc.tcp://localhost:%d' % port_num1)
        cls.clt.connect()

    @classmethod
    def stop_server_and_client(cls):
        cls.clt.disconnect()
        cls.srv.stop()

    @classmethod
    def create_var(cls):
        o = cls.srv.get_objects_node()
        cls.values = [i for i in range(20)]
        cls.var = o.add_variable(3, "history_var", 0)
        cls.srv.iserver.enable_history_data_change(cls.var, period=None, count=10)
        for i in cls.values:
            cls.var.set_value(i)
        time.sleep(1)

    @classmethod
    def create_srv_events(cls):
        cls.ev_values = [i for i in range(20)]
        cls.srvevgen = cls.srv.get_event_generator()

        # for now we need to build our own filter explicitly because node class can't do it
        cls.evfilter = ua.EventFilter()
        ev_type_node = cls.srv.get_node(cls.srvevgen.event.EventType)
        for ev_property in get_props(ev_type_node):
            op = ua.SimpleAttributeOperand()
            op.TypeDefinitionId = ev_type_node.nodeid
            op.AttributeId = ua.AttributeIds.Value
            op.BrowsePath = [ev_property.get_browse_name()]
            cls.evfilter.SelectClauses.append(op)

        cls.revevfilter = ua.EventFilter()
        for ev_property in reversed(get_props(ev_type_node)):
            op = ua.SimpleAttributeOperand()
            op.TypeDefinitionId = ev_type_node.nodeid
            op.AttributeId = ua.AttributeIds.Value
            op.BrowsePath = [ev_property.get_browse_name()]
            cls.revevfilter.SelectClauses.append(op)

        cls.srv_node = cls.srv.get_node(ua.ObjectIds.Server)
        cls.srv.iserver.enable_history_event(cls.srv_node, period=None)

        for i in cls.ev_values:
            cls.srvevgen.event.Severity = cls.ev_values[i]
            cls.srvevgen.trigger(message="test message")
            time.sleep(.1)
        time.sleep(2)

    # no start and no end is not defined by spec, return reverse order
    def test_history_var_read_one(self):
        # Spec says that at least two parameters should be provided, so
        # this one is out of spec
        res = self.var.read_raw_history(None, None, 1)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0].Value.Value, self.values[-1])

    # no start and no end is not defined by spec, return reverse order
    def test_history_var_read_none(self):
        res = self.var.read_raw_history(None, None, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[0].Value.Value, self.values[-1])
        self.assertEqual(res[-1].Value.Value, self.values[0])

    # no start and no end is not defined by spec, return reverse order
    def test_history_var_read_last_3(self):
        res = self.var.read_raw_history(None, None, 3)
        self.assertEqual(len(res), 3)
        self.assertEqual(res[-1].Value.Value, self.values[-3])
        self.assertEqual(res[0].Value.Value, self.values[-1])

    # no start and no end is not defined by spec, return reverse order
    def test_history_var_read_all2(self):
        res = self.var.read_raw_history(None, None, 9999)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].Value.Value, self.values[0])
        self.assertEqual(res[0].Value.Value, self.values[-1])

    # only has end time, should return reverse order
    def test_history_var_read_2_with_end(self):
        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.var.read_raw_history(None, now, 2)
        self.assertEqual(len(res), 2)
        self.assertEqual(res[-1].Value.Value, self.values[-2])
    
    # both start and endtime, return from start to end
    def test_history_var_read_all(self):
        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.var.read_raw_history(old, now, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].Value.Value, self.values[-1])
        self.assertEqual(res[0].Value.Value, self.values[0])

    def test_history_var_read_5_in_timeframe(self):
        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.var.read_raw_history(old, now, 5)
        self.assertEqual(len(res), 5)
        self.assertEqual(res[-1].Value.Value, self.values[4])
        self.assertEqual(res[0].Value.Value, self.values[0])

    # start time greater than end time, should return reverse order
    def test_history_var_read_5_in_timeframe_start_greater_than_end(self):
        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.var.read_raw_history(now, old, 5)
        self.assertEqual(len(res), 5)
        self.assertEqual(res[-1].Value.Value, self.values[-5])
        self.assertEqual(res[0].Value.Value, self.values[-1])
    
    # only start return original order
    def test_history_var_read_6_with_start(self):
        now = datetime.utcnow()
        old = now - timedelta(days=6)
        res = self.var.read_raw_history(old, None, 6)
        self.assertEqual(len(res), 6)
        self.assertEqual(res[-1].Value.Value, self.values[5])
        self.assertEqual(res[0].Value.Value, self.values[0])

    # only start return original order
    def test_history_var_read_all_with_start(self):

        now = datetime.utcnow()
        old = now - timedelta(days=6)
        res = self.var.read_raw_history(old, None, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].Value.Value, self.values[-1])
        self.assertEqual(res[0].Value.Value, self.values[0])

    # only end return reversed order
    def test_history_var_read_all_with_end(self):

        end = datetime.utcnow() + timedelta(days=6)
        res = self.var.read_raw_history(None, end, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].Value.Value, self.values[0])
        self.assertEqual(res[0].Value.Value, self.values[-1])

    # only end return reversed order
    def test_history_var_read_3_with_end(self):
        end = datetime.utcnow() + timedelta(days=6)
        res = self.var.read_raw_history(None, end, 3)
        self.assertEqual(len(res), 3)
        self.assertEqual(res[2].Value.Value, self.values[-3])
        self.assertEqual(res[0].Value.Value, self.values[-1])


class TestHistory(unittest.TestCase, HistoryCommon):

    @classmethod
    def setUpClass(cls):
        cls.start_server_and_client()
        cls.create_var()

    @classmethod
    def tearDownClass(cls):
        cls.stop_server_and_client()


class TestHistorySQL(unittest.TestCase, HistoryCommon):
    @classmethod
    def setUpClass(cls):
        cls.start_server_and_client()
        cls.srv.iserver.history_manager.set_storage(HistorySQLite(":memory:"))
        cls.create_var()
        cls.create_srv_events()

    # ~~~~~~~~~~~~~~~~~~~~~~~ events ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    # only has end time, should return reverse order
    def test_history_ev_read_2_with_end(self):
        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.srv_node.read_event_history(self.evfilter, None, now, 2)
        self.assertEqual(len(res), 2)
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-2])

    # both start and end time, return from start to end
    def test_history_ev_read_all(self):

        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.srv_node.read_event_history(self.evfilter, old, now, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-1])
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])

    def test_history_ev_read_5_in_timeframe(self):

        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.srv_node.read_event_history(self.evfilter, old, now, 5)
        self.assertEqual(len(res), 5)
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[4])
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])

    # start time greater than end time, should return reverse order
    def test_history_ev_read_5_in_timeframe_start_greater_than_end(self):

        now = datetime.utcnow()
        old = now - timedelta(days=6)

        res = self.srv_node.read_event_history(self.evfilter, now, old, 5)
        self.assertEqual(len(res), 5)
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-5])
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[-1])

    # only start return original order
    def test_history_ev_read_6_with_start(self):

        now = datetime.utcnow()
        old = now - timedelta(days=6)
        res = self.srv_node.read_event_history(self.evfilter, old, None, 6)
        self.assertEqual(len(res), 6)
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[5])
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])

    # only start return original order
    def test_history_ev_read_all_with_start(self):

        now = datetime.utcnow()
        old = now - timedelta(days=6)
        res = self.srv_node.read_event_history(self.evfilter, old, None, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[-1])
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[0])

    # only end return reversed order
    def test_history_ev_read_all_with_end(self):

        end = datetime.utcnow() + timedelta(days=6)
        res = self.srv_node.read_event_history(self.evfilter, None, end, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].EventFields[8].Value, self.ev_values[0])
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[-1])

    # only end return reversed order
    def test_history_ev_read_3_with_end(self):

        end = datetime.utcnow() + timedelta(days=6)
        res = self.srv_node.read_event_history(self.evfilter, None, end, 3)
        self.assertEqual(len(res), 3)
        self.assertEqual(res[2].EventFields[8].Value, self.ev_values[-3])
        self.assertEqual(res[0].EventFields[8].Value, self.ev_values[-1])

    # reverse event filter select clauses and test that results match the filter order
    def test_history_ev_read_all_filter_order_reversed(self):

        now = datetime.utcnow()
        old = now - timedelta(days=6)
        res = self.srv_node.read_event_history(self.revevfilter, old, None, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[-1].EventFields[0].Value, self.ev_values[-1])
        self.assertEqual(res[0].EventFields[0].Value, self.ev_values[0])

    @classmethod
    def tearDownClass(cls):
        cls.stop_server_and_client()












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

FreeOpcUa / python-opcua

Push — master ( d538b5...98e417 )

HistoryCommon.test_history_read_last_3() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like