yandex-tank/Tests/MonitoringTest.py

from Tank.MonCollector.collector import MonitoringCollector, \
    MonitoringDataListener, SSHWrapper
from Tank.Plugins.ConsoleOnline import Screen
from Tank.Plugins.Monitoring import MonitoringPlugin, MonitoringWidget
from Tests.ConsoleOnlinePluginTest import FakeConsoleMarkup
from Tests.TankTests import TankTestCase
import logging
import time
import tempfile

class  MonitoringCollectorTestCase(TankTestCase):
    data = None
    
    def test_collector(self):
        mon = MonitoringCollector()
        mon.config="config/mon1.conf"
        mon.ssh_wrapper_class=SSHEmulator
        listener = TestMonListener()
        mon.add_listener(listener)
        mon.prepare()
        mon.start()
        mon.poll()

    def test_plugin_disabled(self):
        core = self.get_core()
        core.artifacts_base_dir=tempfile.mkdtemp()
        mon = MonitoringPlugin(core)
        core.set_option(mon.SECTION, 'config', 'none')
        mon.configure()
        mon.prepare_test()
        mon.start_test()
        self.assertEquals(-1, mon.is_test_finished())
        self.assertEquals(None, mon.monitoring)
        time.sleep(1)
        self.assertEquals(-1, mon.is_test_finished())
        mon.end_test(0)
        mon.post_process(0)

    def test_plugin_default(self):
        core = self.get_core()
        core.artifacts_base_dir=tempfile.mkdtemp()
        core.load_configs(['config/monitoring.conf'])
        core.load_plugins()
        core.plugins_configure()
        core.plugins_prepare_test()
        mon = MonitoringPlugin(core)
        mon.configure()
        mon.monitoring.ssh_wrapper_class=SSHEmulator
        mon.prepare_test()
        mon.start_test()
        self.assertEquals(-1, mon.is_test_finished())
        self.assertNotEquals(None, mon.monitoring)
        time.sleep(1)
        self.assertEquals(-1, mon.is_test_finished())
        mon.end_test(0)
        mon.post_process(0)

    def test_plugin_config(self):
        core = self.get_core()
        core.artifacts_base_dir=tempfile.mkdtemp()
        core.load_configs(['config/monitoring.conf'])
        core.load_plugins()
        core.plugins_configure()
        core.plugins_prepare_test()
        mon = MonitoringPlugin(core)
        mon.monitoring.ssh_wrapper_class=SSHEmulator
        core.set_option(mon.SECTION, 'config', "config/mon1.conf")
        mon.configure()
        mon.prepare_test()
        mon.start_test()
        self.assertEquals(-1, mon.is_test_finished())
        self.assertNotEquals(None, mon.monitoring)
        time.sleep(1)
        self.assertEquals(-1, mon.is_test_finished())
        mon.end_test(0)
        mon.post_process(0)
    
    def test_widget(self):
        core = self.get_core()
        core.artifacts_base_dir=tempfile.mkdtemp()
        owner = MonitoringPlugin(core)
        owner.monitoring=1
        widget = MonitoringWidget(owner)
        screen = Screen(50, FakeConsoleMarkup())
        res = widget.render(screen)
        self.assertEquals("Monitoring is <w>online<rst>:", res)

        widget.monitoring_data("start;127.0.0.1;1347631472;Memory_total;Memory_used;Memory_free;Memory_shared;Memory_buff;Memory_cached;Net_recv;Net_send;Disk_read;Disk_write;System_csw;System_int;CPU_user;CPU_nice;CPU_system;CPU_idle;CPU_iowait;CPU_irq;CPU_softirq;System_numproc;System_numthreads")
        res = widget.render(screen)
        
        widget.monitoring_data("127.0.0.1;1347631473;1507.65625;576.9609375;8055;1518;0;143360;34.9775784753;16.1434977578;0.0")
        res = widget.render(screen)
        self.assertNotEquals("Monitoring is <g>online<rst>:", res)
        
        widget.monitoring_data("127.0.0.1;1347631473;1506.65625;574.9609375;8055;1518;0;143360;34.9775784753;16.1434977578;0.0")
        res = widget.render(screen)

class TestMonListener(MonitoringDataListener):
    def __init__(self):
        self.data = []
    
    def monitoring_data(self, data_string):
        logging.debug("MON DATA: %s", data_string)
        self.data.append(data_string)
        
class SSHEmulator(SSHWrapper):

    def __init__(self):
        SSHWrapper.__init__(self)
    
    def get_scp_pipe(self, cmd):
        self.scp_pipe = PipeEmul('data/ssh_out.txt', 'data/ssh_err.txt')
        return self.scp_pipe
    
    def get_ssh_pipe(self, cmd):
        self.ssh_pipe = PipeEmul('data/ssh_out.txt', 'data/ssh_err.txt')
        return self.ssh_pipe
    
class PipeEmul:
    def __init__(self, out, err):
        self.stderr=open(err, 'rU')
        self.stdout=open(out, 'rU')
        self.returncode=0
        self.pid=0
        
    def wait(self):
        pass
   
    def readline(self):
        return self.stdout.readline()