diff --git a/src/run_tests.py b/src/run_tests.py index 997b010..ef8e31e 100644 --- a/src/run_tests.py +++ b/src/run_tests.py @@ -1,12 +1,14 @@ import unittest from tests.test_csv import TestDAQJobStoreCSV +from tests.test_n1081b import TestDAQJobN1081B def run_tests(): test_suite = unittest.TestSuite() loader = unittest.TestLoader() test_suite.addTests(loader.loadTestsFromTestCase(TestDAQJobStoreCSV)) + test_suite.addTests(loader.loadTestsFromTestCase(TestDAQJobN1081B)) return test_suite diff --git a/src/tests/test_n1081b.py b/src/tests/test_n1081b.py new file mode 100644 index 0000000..8ce94b9 --- /dev/null +++ b/src/tests/test_n1081b.py @@ -0,0 +1,110 @@ +import unittest +from unittest.mock import MagicMock, patch + +from N1081B import N1081B + +from daq.caen.n1081b import DAQJobN1081B, DAQJobN1081BConfig +from daq.store.models import DAQJobMessageStore + + +class TestDAQJobN1081B(unittest.TestCase): + def setUp(self): + self.config = DAQJobN1081BConfig( + daq_job_type="n1081b", + host="localhost", + port="1234", + password="password", + sections_to_store=["SEC_A", "SEC_B"], + store_config=MagicMock(), + ) + self.daq_job = DAQJobN1081B(self.config) + + @patch.object(N1081B, "connect", return_value=True) + @patch.object(N1081B, "login", return_value=True) + def test_connect_to_device_success(self, mock_login, mock_connect): + self.daq_job._connect_to_device() + mock_connect.assert_called_once() + mock_login.assert_called_once_with("password") + + @patch.object(N1081B, "connect", return_value=False) + def test_connect_to_device_failure(self, mock_connect): + with self.assertRaises(Exception) as context: + self.daq_job._connect_to_device() + self.assertTrue("Connection failed" in str(context.exception)) + + @patch.object(N1081B, "login", return_value=False) + @patch.object(N1081B, "connect", return_value=True) + def test_login_failure(self, mock_connect, mock_login): + with self.assertRaises(Exception) as context: + self.daq_job._connect_to_device() + self.assertTrue("Login failed" in str(context.exception)) + + @patch.object( + N1081B, + "get_function_results", + return_value={"data": {"counters": [{"lemo": 1, "value": 100}]}}, + ) + def test_poll_sections(self, mock_get_function_results): + self.daq_job._send_store_message = MagicMock() + self.daq_job._poll_sections() + self.daq_job._send_store_message.assert_called() + self.assertEqual(self.daq_job._send_store_message.call_count, 2) + + @patch.object(N1081B, "get_function_results", return_value={"data": {}}) + def test_poll_sections_no_counters(self, mock_get_function_results): + self.daq_job._send_store_message = MagicMock() + self.daq_job._poll_sections() + self.daq_job._send_store_message.assert_not_called() + + @patch("time.sleep", return_value=None, side_effect=StopIteration) + @patch.object(DAQJobN1081B, "_poll_sections") + @patch.object(DAQJobN1081B, "_connect_to_device") + @patch.object(DAQJobN1081B, "_is_connected", side_effect=[False, True]) + def test_start( + self, mock_is_connected, mock_connect_to_device, mock_poll_sections, mock_sleep + ): + with self.assertRaises(StopIteration): + self.daq_job.start() + mock_connect_to_device.assert_called_once() + mock_poll_sections.assert_called_once() + + @patch.object(N1081B, "get_function_results", return_value=None) + def test_poll_sections_no_results(self, mock_get_function_results): + self.daq_job._send_store_message = MagicMock() + self.daq_job._poll_sections() + self.daq_job._send_store_message.assert_not_called() + + @patch.object( + N1081B, + "get_function_results", + return_value={"data": {"counters": [{"lemo": 1, "value": 100}]}}, + ) + def test_send_store_message(self, mock_get_function_results): + self.daq_job.message_out = MagicMock() + self.daq_job._send_store_message( + {"counters": [{"lemo": 1, "value": 100}]}, "SEC_A" + ) + self.daq_job.message_out.put.assert_called_once() + message = self.daq_job.message_out.put.call_args[0][0] + self.assertIsInstance(message, DAQJobMessageStore) + self.assertEqual(message.prefix, "SEC_A") + self.assertIn("timestamp", message.keys) + self.assertIn("lemo_1", message.keys) + self.assertIn(100, message.data) + + def test_invalid_section_in_config(self): + invalid_config = DAQJobN1081BConfig( + daq_job_type="", + host="localhost", + port="1234", + password="password", + sections_to_store=["INVALID_SECTION"], + store_config=MagicMock(), + ) + with self.assertRaises(Exception) as context: + DAQJobN1081B(invalid_config) + self.assertTrue("Invalid section: INVALID_SECTION" in str(context.exception)) + + +if __name__ == "__main__": + unittest.main()