librenms/tests/tests.py

import threading
import unittest
from os import path

import sys
from time import sleep

try:
    import redis
except ImportError:
    print("Redis tests won't be run")
    pass

sys.path.append(path.dirname(path.dirname(path.abspath(__file__))))
import LibreNMS


class TestLocks(unittest.TestCase):
    def setUp(self):
        pass

    @staticmethod
    def lock_thread(manager, lock_name, expiration, unlock_sleep=0):
        manager.lock(lock_name, 'lock_thread', expiration)

        if unlock_sleep:
            sleep(unlock_sleep)
            manager.unlock(lock_name, 'lock_thread')

    def test_threading_lock(self):
        lm = LibreNMS.ThreadingLock()

        thread = threading.Thread(target=self.lock_thread, args=(lm, 'first.lock', 2, 1))
        thread.daemon = True
        thread.start()

        sleep(0.05)
        self.assertFalse(lm.lock('first.lock', 'main_thread', 0), "Acquired lock when it is held by thread")
        self.assertFalse(lm.unlock('first.lock', 'main_thread'), "Unlocked lock main doesn't own")

        sleep(1.1)
        self.assertTrue(lm.lock('first.lock', 'main_thread', 1),
                        "Could not acquire lock previously held by thread")
        self.assertFalse(lm.lock('first.lock', 'main_thread', 1, False), "Was able to re-lock a lock main owns")
        self.assertTrue(lm.lock('first.lock', 'main_thread', 1, True), "Could not re-lock a lock main owns")
        self.assertTrue(lm.check_lock('first.lock'))
        self.assertTrue(lm.unlock('first.lock', 'main_thread'), "Could not unlock lock main holds")
        self.assertFalse(lm.unlock('first.lock', 'main_thread'), "Unlocked an unlocked lock?")
        self.assertFalse(lm.check_lock('first.lock'))

    def test_redis_lock(self):
        if 'redis' not in sys.modules:
            self.assertTrue(True, 'Skipped Redis tests')
        else:
            rc = redis.Redis()
            rc.delete('lock:redis.lock')  # make sure no previous data exists

            lm = LibreNMS.RedisLock(namespace='lock')
            thread = threading.Thread(target=self.lock_thread, args=(lm, 'redis.lock', 2, 1))
            thread.daemon = True
            thread.start()

            sleep(0.05)
            self.assertFalse(lm.lock('redis.lock', 'main_thread', 1), "Acquired lock when it is held by thread")
            self.assertFalse(lm.unlock('redis.lock', 'main_thread'), "Unlocked lock main doesn't own")

            sleep(1.1)
            self.assertTrue(lm.lock('redis.lock', 'main_thread', 1),
                            "Could not acquire lock previously held by thread")
            self.assertFalse(lm.lock('redis.lock', 'main_thread', 1), "Relocked an existing lock")
            self.assertTrue(lm.lock('redis.lock', 'main_thread', 1, True), "Could not re-lock a lock main owns")
            self.assertTrue(lm.unlock('redis.lock', 'main_thread'), "Could not unlock lock main holds")
            self.assertFalse(lm.unlock('redis.lock', 'main_thread'), "Unlocked an unlocked lock?")

    def queue_thread(self, manager, expect, wait=True):
        self.assertEqual(expect, manager.get(wait), 'Got unexpected data in thread')

    def test_redis_queue(self):
        if 'redis' not in sys.modules:
            self.assertTrue(True, 'Skipped Redis tests')
        else:
            rc = redis.Redis()
            rc.delete('queue:testing')  # make sure no previous data exists
            qm = LibreNMS.RedisUniqueQueue('testing', namespace='queue')

            thread = threading.Thread(target=self.queue_thread, args=(qm, None, False))
            thread.daemon = True
            thread.start()

            thread = threading.Thread(target=self.queue_thread, args=(qm, '2'))
            thread.daemon = True
            thread.start()
            qm.put(2)

            qm.put(3)
            qm.put(4)
            sleep(0.05)
            self.assertEqual(2, qm.qsize())
            self.assertEqual('3', qm.get())
            self.assertEqual('4', qm.get(), "Did not get second item in queue")
            self.assertEqual(None, qm.get_nowait(), "Did not get None when queue should be empty")
            self.assertTrue(qm.empty(), "Queue should be empty")


class TestTimer(unittest.TestCase):
    def setUp(self):
        self.counter = 0

    def count(self):
        self.counter += 1

    def test_recurring_timer(self):
        self.assertEqual(0, self.counter)
        timer = LibreNMS.RecurringTimer(0.5, self.count)
        timer.start()
        self.assertEqual(0, self.counter)
        sleep(0.5)
        self.assertEqual(1, self.counter)
        self.assertEqual(1, self.counter)
        sleep(0.5)
        self.assertEqual(2, self.counter)
        timer.stop()
        self.assertTrue(timer._event.is_set())
        sleep(0.5)
        self.assertEqual(2, self.counter)
        timer.start()
        sleep(0.5)
        self.assertEqual(3, self.counter)
        timer.stop()

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