Skip to content
Snippets Groups Projects

Resolve "Initial Publish uptime of CS instance to IS"

Files
5
@@ -9,11 +9,11 @@ if __name__ == '__main__':
SCRIPT_DIR = abspath(join(dirname(__file__), '..'))
sys.path.append(SCRIPT_DIR)
from cs.Threads.InformationServiceThread import InformationServiceThread
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 TestInformationServiceThread(unittest.TestCase):
class TestInfoServiceThread(unittest.TestCase):
def setUp(self):
## we mock the event so that the thread won't be left hanging
@@ -23,22 +23,28 @@ class TestInformationServiceThread(unittest.TestCase):
## we mock the configuration
self.mock_cfg = mock.create_autospec(ConfigHolder)
self.mock_cfg.configure_mock(partition="initial", LogDir="", LogLevel="")
self.mock_cfg.configure_mock(partition="initial", LogDir="", LogLevel="", FileName="Conf_test.cfg")
class TestInfoServiceExit(TestInformationServiceThread):
## 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):
## we're not interested in any logging for this test
with mock.patch("cs.Threads.InformationServiceThread.enable_file_logging"):
# I run the thread
info_thread = InformationServiceThread(self.mock_cfg, self.mock_event)
info_thread.start()
# I run the thread
self.info_thread.start()
# and shut it down
info_thread.info_service_exit()
info_thread.join(3.0)
self.info_thread.info_service_exit()
self.info_thread.join(3.0)
# and I see that the thread has been killed
self.assertFalse(info_thread.is_alive(), msg="InformationService Thread didn't die")
self.assertFalse(self.info_thread.is_alive(), msg="InfoService Thread didn't die")
class TestGetProcessUptime(TestInfoServiceThread):
def test_uptime_should_be_positive(self):
self.assertGreaterEqual(self.info_thread.get_process_uptime_in_seconds(), 0)
if __name__ == "__main__":
unittest.main()
Loading