Completed
Push — master ( bac760...d0d0cb )
by Olivier
01:26
created

TestHistorySQL.setUpClass()   A

Complexity

Conditions 1

Size

Total Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 4
CRAP Score 1
Metric Value
cc 1
dl 0
loc 5
ccs 4
cts 4
cp 1
crap 1
rs 9.4285
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 by
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 by
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 by
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 by
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 by
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