HistoryCommon

Complexity

Total Complexity

13

Size/Duplication

Total Lines	88
Duplicated Lines	27.27 %

11 Methods

Rating	Name	Duplication	Size	Complexity
A	create_var()	0	8	3
A	stop_server_and_client()	0	3	1
A	test_history_read_one()	0	4	1
A	start_server_and_client()	0	7	1
A	test_history_read_5_in_timeframe_start_greater_than_end()	8	8	1
A	test_history_read_2_with_end()	0	7	1
A	test_history_read_all2()	0	5	1
A	test_history_read_last_3()	0	5	1
A	test_history_read_all()	8	8	1
A	test_history_read_none()	0	6	1
A	test_history_read_5_in_timeframe()	8	8	1

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

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

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

    def stop_server_and_client(self):
        self.clt.disconnect()
        self.srv.stop()

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

    def test_history_read_one(self):
        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):
        # FIXME not sure this once is supported by spec
        res = self.var.read_raw_history(None, None, 0)
        self.assertEqual(len(res), 20)
        self.assertEqual(res[0].Value.Value, self.values[-1])  # self.values was 0
        self.assertEqual(res[-1].Value.Value, self.values[0])  # self.values was -1

    # 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.values was -1
        self.assertEqual(res[0].Value.Value, self.values[-1])  # self.values was -3

    # 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.values was -1
        self.assertEqual(res[0].Value.Value, self.values[-1])  # self.values was 0

    # 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])  # self.values was -1
    
    def test_history_read_all(self):

This code seems to be duplicated in your project.

        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):

This code seems to be duplicated in your project.

        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):

This code seems to be duplicated in your project.

        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])



class TestHistory(unittest.TestCase, HistoryCommon):

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

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


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

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