File 0000-Adding-missing-unit-test-files.patch of Package rhevm-guest-agent

File 0000-Adding-missing-unit-test-files.patch of Package rhevm-guest-agent (Revision 1738f94cbe1a2a662936421cd5f3c43f)

Currently displaying revision 1738f94cbe1a2a662936421cd5f3c43f, show latest

From 7bd2232b7827eb886cb18b5ea62cf2a4f31a540e Mon Sep 17 00:00:00 2001
From: Vinzenz Feenstra <vfeenstr@redhat.com>
Date: Thu, 15 May 2014 14:49:57 +0200
Subject: [PATCH] Adding missing unit test files

Signed-off-by: Vinzenz Feenstra <vfeenstr@redhat.com>
---
 tests/encoding_test.py     |  45 ++++++++++
 tests/guest_agent_test.py  |  93 ++++++++++++++++++++
 tests/message_validator.py | 210 +++++++++++++++++++++++++++++++++++++++++++++
 tests/test_port.py         |  26 ++++++
 tests/testrunner.py        |  68 +++++++++++++++
 5 files changed, 442 insertions(+)
 create mode 100644 tests/encoding_test.py
 create mode 100644 tests/guest_agent_test.py
 create mode 100644 tests/message_validator.py
 create mode 100644 tests/test_port.py
 create mode 100644 tests/testrunner.py

diff --git a/tests/encoding_test.py b/tests/encoding_test.py
new file mode 100644
index 0000000..770bcb6
--- /dev/null
+++ b/tests/encoding_test.py
@@ -0,0 +1,45 @@
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+# vim:fenc=utf-8
+
+from testrunner import GuestAgentTestCase as TestCaseBase
+from VirtIoChannel import _filter_object
+
+
+class EncodingTest(TestCaseBase):
+
+    def testNonUnicodeKeyInput(self):
+        non_unicode_key = {'non-unicode-key': u'unicode value'}
+        self.assertEquals({u'non-unicode-key': u'unicode value'},
+                          _filter_object(non_unicode_key))
+
+    def testNonUnicodeValueInput(self):
+        non_unicode_value = {u'unicode-key': 'non-unicode value'}
+        self.assertEquals({u'unicode-key': u'non-unicode value'},
+                          _filter_object(non_unicode_value))
+
+    def testWindowsFailureOnValidValue(self):
+        VALID = u'\u0F65'
+        self.assertEquals(VALID, _filter_object(VALID))
+
+    def testNullChar(self):
+        non_unicode_value = {u'unicode-key': '\x00'}
+        self.assertEquals({u'unicode-key': u'\ufffd'},
+                          _filter_object(non_unicode_value))
+
+    def testIllegalUnicodeInput(self):
+        ILLEGAL_DATA = {u'foo': u'\x00data\x00test\uffff\ufffe\udc79\ud800'}
+        EXPECTED = {u'foo': u'\ufffddata\ufffdtest\ufffd\ufffd\ufffd\ufffd'}
+        self.assertEqual(EXPECTED, _filter_object(ILLEGAL_DATA))
+
+    def testIllegalUnicodeCharacters(self):
+        INVALID = (u'\u0000', u'\ufffe', u'\uffff', u'\ud800', u'\udc79',
+                   u'\U00000000', '\x00', '\x01', '\x02', '\x03', '\x04',
+                   '\x05')
+        for invchar in INVALID:
+            self.assertEqual(u'\ufffd', _filter_object(invchar))
+
+    def testLegalUnicodeCharacters(self):
+        LEGAL = (u'\u2122', u'Hello World')
+        for legalchar in LEGAL:
+            self.assertEqual(legalchar, _filter_object(legalchar))
diff --git a/tests/guest_agent_test.py b/tests/guest_agent_test.py
new file mode 100644
index 0000000..b3939ad
--- /dev/null
+++ b/tests/guest_agent_test.py
@@ -0,0 +1,93 @@
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+# vim:fenc=utf-8
+
+
+from ConfigParser import ConfigParser
+import platform
+
+from message_validator import MessageValidator
+from testrunner import GuestAgentTestCase
+
+import test_port
+
+
+def _linux_setup_test(conf):
+    port_name = 'linux-functional-test-port'
+    conf.set('general', 'applications_list',
+             'kernel ovirt-guest-agent xorg-x11-drv-qxl '
+             'linux-image xserver-xorg-video-qxl')
+    conf.set('general', 'ignored_fs',
+             'rootfs tmpfs autofs cgroup selinuxfs udev mqueue '
+             'nfds proc sysfs devtmpfs hugetlbfs rpc_pipefs devpts '
+             'securityfs debugfs binfmt_misc fuse.gvfsd-fuse '
+             'fuse.gvfs-fuse-daemon fusectl usbfs')
+    from GuestAgentLinux2 import LinuxVdsAgent
+    return port_name, LinuxVdsAgent
+
+
+def _win32_setup_test(conf):
+    port_name = "windows-functional-test-port"
+    from GuestAgentWin32 import WinVdsAgent
+    return port_name, WinVdsAgent
+
+
+class FunctionalTest(GuestAgentTestCase):
+    def setUp(self):
+        self._config = ConfigParser()
+        self._config.add_section('general')
+        self._config.add_section('virtio')
+
+        agent_class = None
+        if platform.system() in ['Windows', 'Microsoft']:
+            self._vport_name, agent_class = _win32_setup_test(self._config)
+        else:
+            self._vport_name, agent_class = _linux_setup_test(self._config)
+
+        self._validator = MessageValidator(self._vport_name)
+        self._vport = self._validator.port()
+        test_port.add_test_port(self._vport_name, self._vport)
+
+        self._config.set('general', 'heart_beat_rate', '5')
+        self._config.set('general', 'report_user_rate', '10')
+        self._config.set('general', 'report_application_rate', '120')
+        self._config.set('general', 'report_disk_usage', '300')
+        self._config.set('virtio', 'device', self._vport_name)
+
+        self.vdsAgent = agent_class(self._config)
+
+    def testSendInfo(self):
+        self._validator.verifySendInfo(self.vdsAgent)
+
+    def testSendAppList(self):
+        self._validator.verifySendAppList(self.vdsAgent)
+
+    def testSendDisksUsages(self):
+        self._validator.verifySendDisksUsages(self.vdsAgent)
+
+    def testSendMemoryStats(self):
+        self._validator.verifySendMemoryStats(self.vdsAgent)
+
+    def testSendFQDN(self):
+        self._validator.verifySendFQDN(self.vdsAgent)
+
+    def testSendUserInfo(self):
+        self._validator.verifySendUserInfo(self.vdsAgent)
+
+    def testSessionLogon(self):
+        self._validator.verifySessionLogon(self.vdsAgent)
+
+    def testSessionLogoff(self):
+        self._validator.verifySessionLogon(self.vdsAgent)
+
+    def testSessionLock(self):
+        self._validator.verifySessionLock(self.vdsAgent)
+
+    def testSessionUnlock(self):
+        self._validator.verifySessionUnlock(self.vdsAgent)
+
+    def testSessionStartup(self):
+        self._validator.verifySessionStartup(self.vdsAgent)
+
+    def testSessionShutdown(self):
+        self._validator.verifySessionShutdown(self.vdsAgent)
diff --git a/tests/message_validator.py b/tests/message_validator.py
new file mode 100644
index 0000000..95c379d
--- /dev/null
+++ b/tests/message_validator.py
@@ -0,0 +1,210 @@
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+# vim:fenc=utf-8
+
+import test_port
+import json
+import logging
+
+
+class TestPortWriteBuffer(test_port.TestPort):
+    def __init__(self, vport_name, *args, **kwargs):
+        test_port.TestPort.__init__(self, vport_name, *args, **kwargs)
+        self._buffer = ''
+
+    def write(self, buffer):
+        self._buffer = self._buffer + buffer
+        return len(buffer)
+
+    def read(self, size):
+        return ''
+
+    def clear(self):
+        self._buffer = ''
+
+
+def _ensure_messages(*messages):
+    def wrapped(f):
+        def fun(self, *args, **kwargs):
+            result = f(self, *args, **kwargs)
+            names = []
+            parsed = self._get_messages()
+            assert(len(parsed) == len(messages))
+            for m in parsed:
+                assert('__name__' in m)
+                names.append(m['__name__'])
+                self._check(m)
+            for m in messages:
+                assert(m in names)
+            return result
+        return fun
+    return wrapped
+
+
+def _name_only(n):
+    def wrapped(o):
+        assert(len(o) == 1)
+        assert(o['__name__'] == n)
+    return wrapped
+
+
+def assert_string_param(o, n):
+    assert(n in o)
+    assert(isinstance(o[n], basestring))
+
+
+def assert_integral_param(o, n):
+    assert(n in o)
+    integral = isinstance(o[n], (int, long))
+    if not integral and isinstance(o[n], basestring):
+        var = long(o[n])
+        integral = True
+    assert(integral)
+
+
+def _name_and_one_str_param(msg_name, param_name):
+    def wrapped(o):
+        assert(o['__name__'] == msg_name)
+        assert_string_param(o, param_name)
+    return wrapped
+
+
+def assert_is_string_list(o):
+    assert(isinstance(o, list))
+    for s in o:
+        assert(isinstance(s, basestring))
+
+
+def _name_and_one_string_list_param(msg_name, param_name):
+    def wrapped(o):
+        assert(o['__name__'] == msg_name)
+        assert(param_name in o)
+        assert_is_string_list(o[param_name])
+    return wrapped
+
+
+def validate_network_interfaces(msg):
+    assert(msg['__name__'] == 'network-interfaces')
+    assert('interfaces' in msg)
+    assert(isinstance(msg['interfaces'], list))
+    for obj in msg['interfaces']:
+        assert_string_param(obj, 'hw')
+        assert_string_param(obj, 'name')
+        assert('inet' in obj)
+        assert_is_string_list(obj['inet'])
+        assert('inet6' in obj)
+        assert_is_string_list(obj['inet6'])
+
+
+def validate_disks_usage(msg):
+    for disk in msg['disks']:
+        assert_string_param(disk, 'fs')
+        assert_string_param(disk, 'path')
+        assert('total' in disk)
+        assert_integral_param(disk, 'total')
+        assert('used' in disk)
+        assert_integral_param(disk, 'used')
+
+
+def validate_memory_stats(msg):
+    assert('memory' in msg)
+    mem = msg['memory']
+    assert_integral_param(mem, 'majflt')
+    assert_integral_param(mem, 'mem_free')
+    assert_integral_param(mem, 'mem_total')
+    assert_integral_param(mem, 'mem_unused')
+    assert_integral_param(mem, 'pageflt')
+    assert_integral_param(mem, 'swap_in')
+    assert_integral_param(mem, 'swap_out')
+
+
+_MSG_VALIDATORS = {
+    'active-user': _name_and_one_str_param('active-user', 'name'),
+    'applications': _name_and_one_string_list_param('applications',
+                                                    'applications'),
+    'disks-usage': validate_disks_usage,
+    'fqdn': _name_and_one_str_param('fqdn', 'fqdn'),
+    'host-name': _name_and_one_str_param('host-name', 'name'),
+    'memory-stats': validate_memory_stats,
+    'network-interfaces': validate_network_interfaces,
+    'os-version': _name_and_one_str_param('os-version', 'version'),
+    'session-lock': _name_only('session-lock'),
+    'session-logoff': _name_only('session-logoff'),
+    'session-logon': _name_only('session-logon'),
+    'session-shutdown': _name_only('session-shutdown'),
+    'session-startup': _name_only('session-startup'),
+    'session-unlock': _name_only('session-unlock'),
+}
+
+
+def _check_fun(msg):
+    logging.debug("Message: %s", str(msg))
+    assert(msg['__name__'] in _MSG_VALIDATORS)
+    _MSG_VALIDATORS[msg['__name__']](msg)
+
+
+class MessageValidator(object):
+    def __init__(self, vport_name):
+        self._port = TestPortWriteBuffer(vport_name)
+
+    def port(self):
+        return self._port
+
+    def _get_messages(self):
+        result = []
+        for line in self._port._buffer.split('\n'):
+            line = line.strip()
+            if line:
+                result.append(json.loads(line))
+        return result
+
+    def _check(self, msg):
+        _check_fun(msg)
+
+    @_ensure_messages('host-name', 'os-version', 'network-interfaces')
+    def verifySendInfo(self, agent):
+        agent.sendInfo()
+
+    @_ensure_messages('applications')
+    def verifySendAppList(self, agent):
+        agent.sendAppList()
+
+    @_ensure_messages('disks-usage')
+    def verifySendDisksUsages(self, agent):
+        agent.sendDisksUsages()
+
+    @_ensure_messages('memory-stats')
+    def verifySendMemoryStats(self, agent):
+        agent.sendMemoryStats()
+
+    @_ensure_messages('active-user')
+    def verifySendUserInfo(self, agent):
+        agent.sendUserInfo()
+
+    @_ensure_messages('fqdn')
+    def verifySendFQDN(self, agent):
+        agent.sendFQDN()
+
+    @_ensure_messages('active-user', 'session-logon')
+    def verifySessionLogon(self, agent):
+        agent.sessionLogon()
+
+    @_ensure_messages('active-user', 'session-logoff')
+    def verifySessionLogoff(self, agent):
+        agent.sessionLogoff()
+
+    @_ensure_messages('session-lock')
+    def verifySessionLock(self, agent):
+        agent.sessionLock()
+
+    @_ensure_messages('session-unlock')
+    def verifySessionUnlock(self, agent):
+        agent.sessionUnlock()
+
+    @_ensure_messages('session-startup')
+    def verifySessionStartup(self, agent):
+        agent.sessionStartup()
+
+    @_ensure_messages('session-shutdown')
+    def verifySessionShutdown(self, agent):
+        agent.sessionShutdown()
diff --git a/tests/test_port.py b/tests/test_port.py
new file mode 100644
index 0000000..4cb9576
--- /dev/null
+++ b/tests/test_port.py
@@ -0,0 +1,26 @@
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+# vim:fenc=utf-8
+
+
+class TestPort(object):
+    def __init__(self, vport_name, *args, **kwargs):
+        self._vport_name = vport_name
+
+    def write(buffer):
+        return len(buffer)
+
+    def read(size):
+        return ""
+
+
+_registered_ports = {}
+
+
+def get_test_port(vport_name):
+    return _registered_ports.get(vport_name, TestPort(vport_name))
+
+
+def add_test_port(vport_name, port):
+    assert(isinstance(port, TestPort))
+    _registered_ports[vport_name] = port
diff --git a/tests/testrunner.py b/tests/testrunner.py
new file mode 100644
index 0000000..6c7f822
--- /dev/null
+++ b/tests/testrunner.py
@@ -0,0 +1,68 @@
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import logging
+import os
+import sys
+import unittest
+
+from nose import config
+from nose import core
+from nose import result
+
+from VirtIoChannel import VirtIoStream
+
+
+class GuestAgentTestCase(unittest.TestCase):
+    def __init__(self, *args, **kwargs):
+        unittest.TestCase.__init__(self, *args, **kwargs)
+        self.log = logging.getLogger(self.__class__.__name__)
+
+    def assertRaises(self, exceptions, callable, *args, **kwargs):
+            passed = False
+            try:
+                callable(*args, **kwargs)
+            except exceptions:
+                passed = True
+            self.assertTrue(passed)
+
+
+class GuestAgentTestRunner(core.TextTestRunner):
+    def __init__(self, *args, **kwargs):
+        core.TextTestRunner.__init__(self, *args, **kwargs)
+
+    def _makeResult(self):
+        return result.TextTestResult(self.stream,
+                                     self.descriptions,
+                                     self.verbosity,
+                                     self.config)
+
+    def run(self, test):
+        result_ = core.TextTestRunner.run(self, test)
+        return result_
+
+
+def run():
+    argv = sys.argv
+    stream = sys.stdout
+    verbosity = 3
+    testdir = os.path.dirname(os.path.abspath(__file__))
+
+    conf = config.Config(stream=stream,
+                         env=os.environ,
+                         verbosity=verbosity,
+                         workingDir=testdir,
+                         plugins=core.DefaultPluginManager())
+
+    runner = GuestAgentTestRunner(stream=conf.stream,
+                                  verbosity=conf.verbosity,
+                                  config=conf)
+
+    sys.exit(not core.run(config=conf, testRunner=runner, argv=argv))
+
+
+if __name__ == '__main__':
+    # We're ensuring VirtIoStream is monkey patched to unit test output mode
+    # which requires no VirtIO Channel to be present
+    VirtIoStream.is_test = True
+    run()
-- 
1.9.0