TestPlumdUtils

Complexity

Total Complexity

24

Size/Duplication

Total Lines	221
Duplicated Lines	0 %

Importance

Changes	2
Bugs	2	Features	2

18 Methods

Rating	Name	Duplication	Size	Complexity
A	test_interval_invalid1()	0	5	1
A	test_get_file_list_nonexistent()	0	6	1
A	test_interval_eventset()	0	14	2
A	test_get_file_map_normal()	0	14	1
A	test_get_file_map_list_short1()	0	12	1
A	test_get_file_map_list_empty()	0	6	1
A	test_get_hostname()	0	4	1
A	test_get_file_list_normal()	0	8	1
A	test_get_file_map_short1()	0	12	1
A	test_get_file_list_empty()	0	6	1
A	test_get_file_normal1()	0	6	1
A	test_get_file_map_short2()	0	12	1
A	test_interval_eventnotset()	0	18	3
A	test_get_file_map_empty()	0	6	1
A	test_interval_normal1()	0	16	3
A	test_get_file_map_list_nonexistent()	0	6	1
A	test_get_file_map_list_normal()	0	14	1
A	test_get_file_map_list_short2()	0	12	1

import sys
import math
import time
import socket
import os.path
import unittest
from threading import Event
from collections import deque

MAX_DIFF = 8192
FILE_PATH = os.path.realpath(__file__)
PATH = os.path.dirname(FILE_PATH)
DAT_PATH = os.path.join(PATH, "dat")
sys.path.insert(0, os.path.realpath(os.path.join(PATH, "../../../")))
import plumd.util


class TestPlumdUtils(unittest.TestCase):

    def test_get_hostname(self):
        """ensure hostname properly generated"""
        hostname = socket.gethostname().split(".")[0]
        self.assertEqual(plumd.util.get_hostname(), hostname)


    def test_get_file_map_normal(self):
        """ensure get_file_map returns expected mappings"""
        self.maxDiff = MAX_DIFF
        fname = os.path.join(DAT_PATH, "get_file_map/normal1.dat")
        check = plumd.util.get_file_map(fname, 2, 0)
        expected = {
            'vda': deque(['253','0','12985','19498','566600','80185','8755',
                          '25098','270730','344158','0','179702','424339']),
            'vda1': deque(['253','1','112','1231','2702','80','13','4','34',
                           '79','0','120','159']),
            'vda2': deque(['253','2','12864','18251','563698','80078','8742',
                           '25094','270696','344079','0','179608','424153'])
        }
        self.assertEqual(check, expected)


    def test_get_file_map_short1(self):
        """ensure get_file_map properly handles invalid/short input"""
        self.maxDiff = MAX_DIFF
        fname = os.path.join(DAT_PATH, "get_file_map/short1.dat")
        check = plumd.util.get_file_map(fname, 2, 0)
        expected = {
            'vda': deque(['253','0','12985','19498','566600','80185','8755',
                          '25098','270730','344158','0','179702','424339']),
            'vda1': deque(['253','1','112','1231','2702','80','13','4','34',
                           '79','0','120','159'])
        }
        self.assertEqual(check, expected)


    def test_get_file_map_short2(self):
        """ensure get_file_map properly handles invalid/short input"""
        self.maxDiff = MAX_DIFF
        fname = os.path.join(DAT_PATH, "get_file_map/short2.dat")
        check = plumd.util.get_file_map(fname, 2, 0)
        expected = {
            'vda': deque(['253','0','12985','19498','566600','80185','8755',
                          '25098','270730','344158','0','179702','424339']),
            'vda1': deque(['253','1','112','1231','2702','80','13','4','34',
                           '79','0','120','159'])
        }
        self.assertEqual(check, expected)


    def test_get_file_map_empty(self):
        """ensure get_file_map properly handles invalid/short input"""
        fname = os.path.join(DAT_PATH, "get_file_map/empty.dat")
        check = plumd.util.get_file_map(fname, 2, 0)
        expected = {}
        self.assertEqual(check, expected)


    def test_get_file_map_list_normal(self):
        """ensure get_file_map returns expected mappings"""
        self.maxDiff = MAX_DIFF
        fname = os.path.join(DAT_PATH, "get_file_map/normal1.dat")
        check = plumd.util.get_file_map_list(fname, 2, 0)
        expected = {
            'vda': ['253','0','12985','19498','566600','80185','8755',
                          '25098','270730','344158','0','179702','424339'],
            'vda1': ['253','1','112','1231','2702','80','13','4','34',
                           '79','0','120','159'],
            'vda2': ['253','2','12864','18251','563698','80078','8742',
                           '25094','270696','344079','0','179608','424153']
        }
        self.assertEqual(check, expected)


    def test_get_file_map_list_short1(self):
        """ensure get_file_map properly handles invalid/short input"""
        self.maxDiff = MAX_DIFF
        fname = os.path.join(DAT_PATH, "get_file_map/short1.dat")
        check = plumd.util.get_file_map_list(fname, 2, 0)
        expected = {
            'vda': ['253','0','12985','19498','566600','80185','8755',
                          '25098','270730','344158','0','179702','424339'],
            'vda1': ['253','1','112','1231','2702','80','13','4','34',
                           '79','0','120','159']
        }
        self.assertEqual(check, expected)


    def test_get_file_map_list_short2(self):
        """ensure get_file_map properly handles invalid/short input"""
        self.maxDiff = MAX_DIFF
        fname = os.path.join(DAT_PATH, "get_file_map/short2.dat")
        check = plumd.util.get_file_map_list(fname, 2, 0)
        expected = {
            'vda': ['253','0','12985','19498','566600','80185','8755',
                          '25098','270730','344158','0','179702','424339'],
            'vda1': ['253','1','112','1231','2702','80','13','4','34',
                           '79','0','120','159']
        }
        self.assertEqual(check, expected)


    def test_get_file_map_list_empty(self):
        """ensure get_file_map properly handles invalid/short input"""
        fname = os.path.join(DAT_PATH, "get_file_map/empty.dat")
        check = plumd.util.get_file_map_list(fname, 2, 0)
        expected = {}
        self.assertEqual(check, expected)


    def test_get_file_map_list_nonexistent(self):
        """ensure get_file_map properly handles invalid/short input"""
        fname = os.path.join(DAT_PATH, "get_file_map/nonexistantfile")
        check = plumd.util.get_file_map_list(fname, 2, 0)
        expected = {}
        self.assertEqual(check, expected)


    def test_get_file_list_normal(self):
        """ensure get_file_list properly handles normal inpurt"""
        fname = os.path.join(DAT_PATH, "get_file_list/normal1.dat")
        check = plumd.util.get_file_list(fname)
        expected = deque(['rootfs / rootfs rw 0 0\r',
                          'sysfs /sys sysfs rw,seclabel,nosuid,nodev,noexec,relatime 0 0\r',
                          'proc /proc proc rw,nosuid,nodev,noexec,relatime 0 0'])
        self.assertEqual(check, expected)


    def test_get_file_list_empty(self):
        """ensure get_file_list properly handles empty inpurt"""
        fname = os.path.join(DAT_PATH, "get_file_list/empty.dat")
        check = plumd.util.get_file_list(fname)
        expected = deque([''])
        self.assertEqual(check, expected)


    def test_get_file_list_nonexistent(self):
        """ensure get_file_list properly handles empty inpurt"""
        fname = os.path.join(DAT_PATH, "get_file_list/nonexistantfile")
        check = plumd.util.get_file_list(fname)
        expected = deque([''])
        self.assertEqual(check, expected)


    def test_get_file_normal1(self):
        """ensure get_file properly handles normal inpurt"""
        fname = os.path.join(DAT_PATH, "get_file/normal1.dat")
        check = plumd.util.get_file(fname)
        expected = "get_file Normal1"
        self.assertEqual(check, expected)


    def test_get_file_normal1(self):
        """ensure get_file properly handles empty inpurt"""
        fname = os.path.join(DAT_PATH, "get_file/nonexistantfile")
        check = plumd.util.get_file(fname)
        expected = ""
        self.assertEqual(check, expected)


    def test_interval_normal1(self):
        """ensure interval class functioning normally"""
        interval = 0.01
        start = time.time()
        with plumd.util.Interval(interval):
            pass
        end = time.time()
        # should be roughly 0.1 seconds
        took = end - start
        if took > interval:
            diff = took - interval
        else:
            diff = interval - took
        # expect a small difference
        msg = "expected interval of {0} s, difference was: {1}"
        self.assertTrue(diff < interval/100, msg=msg.format(interval, diff))


    def test_interval_invalid1(self):
        """ensure interval class handling invalid interval"""
        interval = -0.1
        start = time.time()
        self.assertRaises(ValueError, plumd.util.Interval, interval)


    def test_interval_eventnotset(self):
        """ensure interval class handling events properly"""
        interval = 0.01
        evt = Event()
        start = time.time()
        with plumd.util.Interval(interval, evt):
            pass
        end = time.time()
        took = end - start
        # should be roughly 0.1 seconds
        if took > interval:
            diff = took - interval
        else:
            diff = interval - took
        expected = interval/10
        # expect a very very small difference - note: heavily loaded systems may trigger failure
        msg = "expected interval of {0} s, difference was: {1}"
        self.assertTrue(diff < expected, msg=msg.format(diff, diff))


    def test_interval_eventset(self):
        """ensure interval class handling events properly"""
        interval = 0.01
        evt = Event()
        start = time.time()
        with plumd.util.Interval(interval, evt):
            evt.set()
        end = time.time()
        diff = end - start
        # in this case the loop is expected to complete early
        # if we get here the test passes if it didn't take longer
        # than interval
        # running on heavily loaded systems could fail this though..
        self.assertTrue(interval > diff)

if __name__ == '__main__':
    unittest.main()