From 1e4fdbd99ccdf84f19f6bfcbfe970bce043eebd7 Mon Sep 17 00:00:00 2001
From: Robert Latta <robert.latta@geant.org>
Date: Fri, 7 Dec 2018 14:33:19 +0000
Subject: [PATCH] Tests for alarmsdb
---
test/test_alarmsdb.py | 101 ++++++++++++++++++++++++++++++++++++++++++
1 file changed, 101 insertions(+)
create mode 100644 test/test_alarmsdb.py
diff --git a/test/test_alarmsdb.py b/test/test_alarmsdb.py
new file mode 100644
index 00000000..c993a762
--- /dev/null
+++ b/test/test_alarmsdb.py
@@ -0,0 +1,101 @@
+import inventory_provider.alarmsdb as alarmsdb
+
+
+def test_infinera_interface_status(mocker):
+ mocked_get_cursor = mocker.patch('inventory_provider.alarmsdb.cursor')
+ mocked_execute = mocked_get_cursor. \
+ return_value.__enter__.return_value.execute
+ mocked_fetchone = mocked_get_cursor.return_value.__enter__. \
+ return_value.fetchone
+ mocked_fetchone.return_value = ('Raised',)
+ assert alarmsdb.get_last_known_infinera_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "down"
+
+ mocked_fetchone.return_value = ("Clear",)
+ assert alarmsdb.get_last_known_infinera_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "up"
+
+ mocked_fetchone.return_value = ()
+ assert alarmsdb.get_last_known_infinera_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "unknown"
+
+ mocked_execute.assert_called_with(
+ "SELECT status FROM infinera_alarms WHERE"
+ " CONCAT(ne_name, '-', REPLACE(object_name, 'T', '')) = %s"
+ " ORDER BY ne_init_time DESC, ne_clear_time DESC LIMIT 1",
+ ("eq1-intfc1",))
+
+
+def test_coriant_interface_status(mocker):
+ mocked_get_cursor = mocker.patch('inventory_provider.alarmsdb.cursor')
+ mocked_execute = mocked_get_cursor. \
+ return_value.__enter__.return_value.execute
+ mocked_fetchone = mocked_get_cursor.return_value.__enter__. \
+ return_value.fetchone
+ mocked_fetchone.return_value = ('Raised',)
+ assert alarmsdb.get_last_known_coriant_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "down"
+
+ mocked_fetchone.return_value = ("Clear",)
+ assert alarmsdb.get_last_known_coriant_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "up"
+
+ mocked_fetchone.return_value = ()
+ assert alarmsdb.get_last_known_coriant_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "unknown"
+
+ mocked_execute.assert_called_with(
+ "SELECT status FROM coriant_alarms"
+ " WHERE ne_id_name = %s AND entity_string LIKE %s"
+ " ORDER BY last_event_time DESC LIMIT 1",
+ ("eq1", "intfc1-%"))
+
+
+def test_juniper_interface_status(mocker):
+ mocked_get_cursor = mocker.patch('inventory_provider.alarmsdb.cursor')
+ mocked_execute = mocked_get_cursor. \
+ return_value.__enter__.return_value.execute
+ mocked_fetchone = mocked_get_cursor.return_value.__enter__. \
+ return_value.fetchone
+ mocked_fetchone.return_value = (0,)
+ assert alarmsdb.get_last_known_juniper_link_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "down"
+
+ mocked_fetchone.return_value = (1,)
+ assert alarmsdb.get_last_known_juniper_link_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "up"
+
+ mocked_fetchone.return_value = ()
+ assert alarmsdb.get_last_known_juniper_link_interface_status(
+ None, 'eq1', 'intfc1'
+ ) == "unknown"
+
+ mocked_execute.assert_called_with(
+ "SELECT IF(link_admin_status = 'up'"
+ " AND link_oper_status = 'up', 1, 0)"
+ " AS up FROM juniper_alarms"
+ " WHERE equipment_name = %s AND link_interface_name = %s"
+ " ORDER BY alarm_id DESC LIMIT 1",
+ ("lo0.eq1", "intfc1"))
+
+
+def test_interface_status(mocker):
+ mocked_infinera = mocker.patch(
+ 'inventory_provider.alarmsdb.get_last_known_infinera_interface_status')
+ mocked_infinera.return_value = "unknown"
+ mocked_coriant = mocker.patch(
+ 'inventory_provider.alarmsdb.get_last_known_coriant_interface_status')
+ mocked_coriant.return_value = "unknown"
+ mocked_juniper = mocker.patch(
+ 'inventory_provider.alarmsdb.'
+ 'get_last_known_juniper_link_interface_status')
+ mocked_juniper.return_value = "unknown"
+ assert alarmsdb.get_last_known_interface_status(None, '', '') == "unknown"
--
GitLab