from django.test import TestCase import threading import configparser import os import common from common.agent import Agent from common.sender import Sender class AgentTests(TestCase): def setUp(self): self.config_file = os.path.join(os.path.abspath(common.__path__[0]),"pyros_agent_config.ini") self.config = configparser.ConfigParser() self.config.read(self.config_file) def test_all_agents_created(self): ''' Tests if all the agents are alive, and send them stop message ''' return threads = threading.enumerate() thread_names = [thread.name for thread in threads if thread.name != "MainThread"] self.assertEqual(len(thread_names), 7, "There should be 7 threads (7 agents)") for section in list(self.config)[1:]: self.assertIn(section, thread_names, "Missing thread : %s" %(section)) for thread in [thread for thread in threads if thread.name != "MainThread"]: Sender.send_to(thread.name, Agent.MSG_SHUTDOWN) thread.join() self.assertEqual(threading.active_count(), 1, "Only the MainThread should still be alive")