tests.py
1.18 KB
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")