Skip to content
Snippets Groups Projects

Resolve "Initial Publish uptime of CS instance to IS"

2 files
+ 60
23
Compare changes
  • Side-by-side
  • Inline
Files
2
+ 59
0
import unittest
import mock
import threading
if __name__ == '__main__':
import sys
from os.path import dirname, abspath, join
## add CastorScript/Script to path so imports keeps working
SCRIPT_DIR = abspath(join(dirname(__file__), '..'))
sys.path.append(SCRIPT_DIR)
from cs.Threads.InfoServiceThread import InfoServiceThread
from cs.Tools.ConfigParser import ConfigHolder
# as a developer I want to publish information to the information service
class TestInfoServiceThread(unittest.TestCase):
def setUp(self):
## we mock the event so that the thread won't be left hanging
self.mock_event = mock.create_autospec(threading.Event())
kwargs = {"wait.return_value": True}
self.mock_event.configure_mock(**kwargs)
## we mock the configuration
self.mock_cfg = mock.create_autospec(ConfigHolder)
#self.mock_cfg.configure_mock(partition="initial", LogDir="", LogLevel="", FileName="Conf_test.cfg")
self.mock_cfg.configure_mock(partition="initial",
LogDir="",
LogLevel="",
FileName="Conf_test.cfg",
IS_partition="initial",
IS_server="RunParams",
IS_publication_period="5")
## we're not interested in any logging for this test
with mock.patch("cs.Threads.InfoServiceThread.enable_file_logging"):
self.info_thread = InfoServiceThread(self.mock_cfg, self.mock_event)
class TestInfoServiceExit(TestInfoServiceThread):
def test_it_should_stop_thread_execution(self):
# I run the thread
self.info_thread.start()
# and shut it down
self.info_thread.info_service_exit()
self.info_thread.join(3.0)
# and I see that the thread has been killed
self.assertFalse(self.info_thread.is_alive(), msg="InfoService Thread didn't die")
class TestGetProcessUptime(TestInfoServiceThread):
def test_it_should_return_a_positive_value(self):
self.assertGreaterEqual(self.info_thread.get_process_uptime_in_seconds(), 0)
if __name__ == "__main__":
unittest.main()
Loading