HistoryCommon.create_var() - Code Metrics - Inspection of "SQL History storage" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#158)

by Olivier

created 2016-04-18 08:51 UTC

HistoryCommon.create_var() A

↳ Parent: HistoryCommon

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	3

Metric	Value
cc	3
dl	0
loc	9
ccs	8
cts	8
cp	1
crap	3
rs	9.6666

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

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(cls.var, period=None, count=10)
        for i in cls.values:
            cls.var.set_value(i)
        time.sleep(1)

    # no start and no end is not defined by spec, return reverse order
    def test_history_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_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_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_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_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_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_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_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_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_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_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_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()

    @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	1		from opcua.server.history_sql import HistorySQLite
9
10	1		from tests_common import CommonTests, add_server_methods
11
12	1		port_num1 = 48530
13	1		port_num2 = 48530
14
15
16
17	1		class HistoryCommon(object):
18	1		srv = Server
19	1		clt = Client
20
21	1		@classmethod
22			def start_server_and_client(cls):
23	1		cls.srv = Server()
24	1		cls.srv.set_endpoint('opc.tcp://localhost:%d' % port_num1)
25	1		cls.srv.start()
26
27	1		cls.clt = Client('opc.tcp://localhost:%d' % port_num1)
28	1		cls.clt.connect()
29
30	1		@classmethod
31			def stop_server_and_client(cls):
32	1		cls.clt.disconnect()
33	1		cls.srv.stop()
34
35	1		@classmethod
36			def create_var(cls):
37	1		o = cls.srv.get_objects_node()
38	1		cls.values = [i for i in range(20)]
39	1		cls.var = o.add_variable(3, "history_var", 0)
40	1		cls.srv.iserver.enable_history(cls.var, period=None, count=10)
41	1		for i in cls.values:
42	1		cls.var.set_value(i)
43	1		time.sleep(1)
44
45			# no start and no end is not defined by spec, return reverse order
46	1		def test_history_read_one(self):
47			# Spec says that at least two parameters should be provided, so
48			# this one is out of spec
49	1		res = self.var.read_raw_history(None, None, 1)
50	1		self.assertEqual(len(res), 1)
51	1		self.assertEqual(res[0].Value.Value, self.values[-1])
52
53			# no start and no end is not defined by spec, return reverse order
54	1		def test_history_read_none(self):
55	1		res = self.var.read_raw_history(None, None, 0)
56	1		self.assertEqual(len(res), 20)
57	1		self.assertEqual(res[0].Value.Value, self.values[-1])
58	1		self.assertEqual(res[-1].Value.Value, self.values[0])
59
60			# no start and no end is not defined by spec, return reverse order
61	1		def test_history_read_last_3(self):
62	1		res = self.var.read_raw_history(None, None, 3)
63	1		self.assertEqual(len(res), 3)
64	1		self.assertEqual(res[-1].Value.Value, self.values[-3])
65	1		self.assertEqual(res[0].Value.Value, self.values[-1])
66
67			# no start and no end is not defined by spec, return reverse order
68	1		def test_history_read_all2(self):
69	1		res = self.var.read_raw_history(None, None, 9999)
70	1		self.assertEqual(len(res), 20)
71	1		self.assertEqual(res[-1].Value.Value, self.values[0])
72	1		self.assertEqual(res[0].Value.Value, self.values[-1])
73
74			# only has end time, should return reverse order
75	1		def test_history_read_2_with_end(self):
76	1		now = datetime.utcnow()
77	1		old = now - timedelta(days=6)
78
79	1		res = self.var.read_raw_history(None, now, 2)
80	1		self.assertEqual(len(res), 2)
81	1		self.assertEqual(res[-1].Value.Value, self.values[-2])
82
83			# both start and endtime, return from start to end
84	1	View Code Duplication	def test_history_read_all(self):
			0 ignored issues – show Duplication introduced 2016-04-14 07:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
85	1		now = datetime.utcnow()
86	1		old = now - timedelta(days=6)
87
88	1		res = self.var.read_raw_history(old, now, 0)
89	1		self.assertEqual(len(res), 20)
90	1		self.assertEqual(res[-1].Value.Value, self.values[-1])
91	1		self.assertEqual(res[0].Value.Value, self.values[0])
92
93	1	View Code Duplication	def test_history_read_5_in_timeframe(self):
			0 ignored issues – show Duplication introduced 2016-04-14 07:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
94	1		now = datetime.utcnow()
95	1		old = now - timedelta(days=6)
96
97	1		res = self.var.read_raw_history(old, now, 5)
98	1		self.assertEqual(len(res), 5)
99	1		self.assertEqual(res[-1].Value.Value, self.values[4])
100	1		self.assertEqual(res[0].Value.Value, self.values[0])
101
102			# start time greater than end time, should return reverse order
103	1	View Code Duplication	def test_history_read_5_in_timeframe_start_greater_than_end(self):
			0 ignored issues – show Duplication introduced 2016-04-17 07:34 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
104	1		now = datetime.utcnow()
105	1		old = now - timedelta(days=6)
106
107	1		res = self.var.read_raw_history(now, old, 5)
108	1		self.assertEqual(len(res), 5)
109	1		self.assertEqual(res[-1].Value.Value, self.values[-5])
110	1		self.assertEqual(res[0].Value.Value, self.values[-1])
111
112			# only start return original order
113	1	View Code Duplication	def test_history_read_6_with_start(self):
			0 ignored issues – show Duplication introduced 2016-04-18 07:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
114	1		now = datetime.utcnow()
115	1		old = now - timedelta(days=6)
116	1		res = self.var.read_raw_history(old, None, 6)
117	1		self.assertEqual(len(res), 6)
118	1		self.assertEqual(res[-1].Value.Value, self.values[5])
119	1		self.assertEqual(res[0].Value.Value, self.values[0])
120
121			# only start return original order
122	1	View Code Duplication	def test_history_read_all_with_start(self):
			0 ignored issues – show Duplication introduced 2016-04-18 07:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
123	1		now = datetime.utcnow()
124	1		old = now - timedelta(days=6)
125	1		res = self.var.read_raw_history(old, None, 0)
126	1		self.assertEqual(len(res), 20)
127	1		self.assertEqual(res[-1].Value.Value, self.values[-1])
128	1		self.assertEqual(res[0].Value.Value, self.values[0])
129
130			# only end return reversed order
131	1		def test_history_read_all_with_end(self):
132	1		end = datetime.utcnow() + timedelta(days=6)
133	1		res = self.var.read_raw_history(None, end, 0)
134	1		self.assertEqual(len(res), 20)
135	1		self.assertEqual(res[-1].Value.Value, self.values[0])
136	1		self.assertEqual(res[0].Value.Value, self.values[-1])
137
138			# only end return reversed order
139	1		def test_history_read_3_with_end(self):
140	1		end = datetime.utcnow() + timedelta(days=6)
141	1		res = self.var.read_raw_history(None, end, 3)
142	1		self.assertEqual(len(res), 3)
143	1		self.assertEqual(res[2].Value.Value, self.values[-3])
144	1		self.assertEqual(res[0].Value.Value, self.values[-1])
145
146
147
148
149	1		class TestHistory(unittest.TestCase, HistoryCommon):
150
151	1		@classmethod
152			def setUpClass(cls):
153	1		cls.start_server_and_client()
154	1		cls.create_var()
155
156	1		@classmethod
157			def tearDownClass(cls):
158	1		cls.stop_server_and_client()
159
160
161	1		class TestHistorySQL(unittest.TestCase, HistoryCommon):
162	1		@classmethod
163			def setUpClass(cls):
164	1		cls.start_server_and_client()
165	1		cls.srv.iserver.history_manager.set_storage(HistorySQLite(":memory:"))
166	1		cls.create_var()
167
168	1		@classmethod
169			def tearDownClass(cls):
170	1		cls.stop_server_and_client()
171
172
173
174
175
176
177
178
179
180
181

FreeOpcUa / python-opcua

Pull Request — master (#158)

HistoryCommon.create_var() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like