<commit_before>from Tank.MonCollector.agent.agent import CpuStat, Custom, Disk, NetTcp, NetTxRx
import os
import tempfile
import time
import unittest
import base64


if __name__ == '__main__':
    unittest.main()

class  Mem_TestCase(unittest.TestCase):
    def setUp(self):
        self.foo = CpuStat()

    def test_get(self):
        print self.foo.check()
        self.assertEquals(len(self.foo.columns()), len(self.foo.check()))
        time.sleep(1)
        self.assertNotEquals(['0', '0'], self.foo.check())

class  Custom_TestCase(unittest.TestCase):
    def setUp(self):
        pass
    #def tearDown(self):
    #    self.foo.dispose()
    #    self.foo = None

    def test_custom_(self):
        custom_config = {'tail': [], 'call': ['ZGlmZkV4:aWZjb25maWcgLXMgZXRoMCB8IGF3ayAnJDE9PSJldGgwIiB7cHJpbnQgJDR9Jw==:1']}
        self.foo = Custom(**custom_config)

        #self.assertEqual(x, y, "Msg");
        x = self.foo.check()
        print x
        self.assertEquals(["0"], x)
        time.sleep(1)
        y = self.foo.check()
        print y
        assert x != y;
        time.sleep(0.5)
        print self.foo.check()
        
    def test_custom_nodiff(self):
        tail_fd, tailfile = tempfile.mkstemp()
        tail = ["%s:%s:%s" % (base64.b64encode('lbl'), base64.b64encode(tailfile), 0)]
        call = ["%s:%s:%s" % (base64.b64encode('lbl2'), base64.b64encode("date +%s"), 0)]
        self.foo = Custom(call, tail)

        x = self.foo.check()
        print "second test", x
        self.assertNotEquals(["0.0"], x)
        self.assertEquals('0', x[0])
        time.sleep(1)
        
        tailval = str(time.time())
        os.write(tail_fd, "%s\n" % tailval)
        y = self.foo.check()
        self.assertNotEquals(x[1], y[1])
        self.assertEquals(tailval, y[0])
        
        time.sleep(2)
        tailval = str(time.time())
        os.write(tail_fd, "%s\n" % tailval)
        z = self.foo.check()
        self.assertEquals(tailval, z[0])
        self.assertNotEquals(y[1], z[1])
        
    def test_custom_fail(self):
        tail = ["%s:%s:%s" % (base64.b64encode('lbl'), base64.b64encode("notexistent"), 0)]
        call = ["%s:%s:%s" % (base64.b64encode('lbl2'), base64.b64encode("notexistent"), 0)]
        self.foo = Custom(call, tail)

        x = self.foo.check()
        self.assertEquals("0", x[0])
        self.assertEquals("0", x[1])
        
    def test_custom_fail2(self):
        custom_config = {'tail': [], 'call': ['TnVtUGhyYXNlcw==:Y2F0IC92YXIvdG1wL3N0YXQx:0']}
        self.foo = Custom(**custom_config)

        x = self.foo.check()


class  Disk_TestCase(unittest.TestCase):
    def setUp(self):
        self.foo = Disk()

    def test_get(self):
        print self.foo.check()
        self.assertEquals(2, len(self.foo.check()))
        self.assertNotEquals(['', ''], self.foo.check())
        fd = tempfile.mkstemp()[0]
        os.write(fd, ' ' * 5000)
        time.sleep(1)
        self.assertNotEquals(['', ''], self.foo.check())

    def test_cols(self):
        res = self.foo.columns()
        self.assertEquals(['Disk_read', 'Disk_write'], res)



class  Net_tcp_TestCase(unittest.TestCase):
    def setUp(self):
        self.foo = NetTcp()

    def test_net_tcp_(self):
        print self.foo.check()
        self.assertEquals(4, len(self.foo.check()))
        self.assertNotEquals(['0', '0', '0', '0'], self.foo.check())<commit_msg>fix unit test<commit_after>from Tank.MonCollector.agent.agent import CpuStat, Custom, Disk, NetTcp, NetTxRx
import os
import tempfile
import time
import unittest
import base64


if __name__ == '__main__':
    unittest.main()

class  Mem_TestCase(unittest.TestCase):
    def setUp(self):
        self.foo = CpuStat()

    def test_get(self):
        print self.foo.check()
        self.assertEquals(len(self.foo.columns()), len(self.foo.check()))
        time.sleep(1)
        self.assertNotEquals(['0', '0'], self.foo.check())

class  Custom_TestCase(unittest.TestCase):
    def setUp(self):
        pass
    #def tearDown(self):
    #    self.foo.dispose()
    #    self.foo = None

    def test_custom_(self):
        custom_config = {'tail': [], 'call': ['ZGlmZkV4:aWZjb25maWcgLXMgZXRoMCB8IGF3ayAnJDE9PSJldGgwIiB7cHJpbnQgJDR9Jw==:1']}
        self.foo = Custom(**custom_config)

        #self.assertEqual(x, y, "Msg");
        x = self.foo.check()
        print x
        self.assertEquals(["0"], x)
        time.sleep(1)
        y = self.foo.check()
        print y
        assert x != y;
        time.sleep(0.5)
        print self.foo.check()
        
    def test_custom_nodiff(self):
        tail_fd, tailfile = tempfile.mkstemp()
        tail = ["%s:%s:%s" % (base64.b64encode('lbl'), base64.b64encode(tailfile), 0)]
        call = ["%s:%s:%s" % (base64.b64encode('lbl2'), base64.b64encode("date +%s"), 0)]
        self.foo = Custom(call, tail)

        x = self.foo.check()
        print "second test", x
        self.assertNotEquals(["0.0"], x)
        self.assertEquals('0', x[0])
        time.sleep(1)
        
        tailval = str(time.time())
        os.write(tail_fd, "%s\n" % tailval)
        y = self.foo.check()
        self.assertNotEquals(x[1], y[1])
        self.assertEquals(tailval, y[0])
        
        time.sleep(2)
        tailval = str(time.time())
        os.write(tail_fd, "%s\n" % tailval)
        z = self.foo.check()
        self.assertEquals(tailval, z[0])
        self.assertNotEquals(y[1], z[1])
        
    def test_custom_fail(self):
        tail = ["%s:%s:%s" % (base64.b64encode('lbl'), base64.b64encode("notexistent"), 0)]
        call = ["%s:%s:%s" % (base64.b64encode('lbl2'), base64.b64encode("notexistent"), 0)]
        self.foo = Custom(call, tail)

        x = self.foo.check()
        self.assertEquals("0", x[0])
        self.assertEquals("0", x[1])
        
    def test_custom_fail2(self):
        custom_config = {'tail': [], 'call': ['TnVtUGhyYXNlcw==:Y2F0IC92YXIvdG1wL3N0YXQx:0']}
        self.foo = Custom(**custom_config)

        x = self.foo.check()


class  Disk_TestCase(unittest.TestCase):
    def setUp(self):
        self.foo = Disk()

    def test_get(self):
        print self.foo.check()
        self.assertEquals(2, len(self.foo.check()))
        # self.assertNotEquals(['', ''], self.foo.check()) disabled since fails on TravisCI
        fd = tempfile.mkstemp()[0]
        os.write(fd, ' ' * 5000)
        time.sleep(1)
        # self.assertNotEquals(['', ''], self.foo.check()) disabled since fails on TravisCI

    def test_cols(self):
        res = self.foo.columns()
        self.assertEquals(['Disk_read', 'Disk_write'], res)



class  Net_tcp_TestCase(unittest.TestCase):
    def setUp(self):
        self.foo = NetTcp()

    def test_net_tcp_(self):
        print self.foo.check()
        self.assertEquals(4, len(self.foo.check()))
        self.assertNotEquals(['0', '0', '0', '0'], self.foo.check())
{'fraction_non_alphanumeric': 0.1675377468060395, 'fraction_numerical': 0.02700348432055749, 'mean_word_length': 2.5130035696073434, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2090799', 'n_tokens_mistral': 2521, 'n_tokens_neox': 2420, 'n_words': 421}