#177 test fixes (mostly py3k):

* only use the strtobytes dict key hack if the given key didn't work
* clear child_reaper after cleaning it up so we can instantiate a new one
* assertEquals is deprecated, replace it by assertEqual
* import gobject via the compat class to make the tests run on py3k
* exec the subprocess test with the correct version of python: use 'python3' with py3k
* add child reaper test stub
* re-enable xor test

git-svn-id: https://xpra.org/svn/Xpra/trunk@9187 3bb7dfac-3a0b-4e04-842a-767bc560f471

src/tests/unit/codecs/cyxor_test_failing.py → src/tests/unit/codecs/cyxor_test.py

@@ -4,13 +4,19 @@
 # Xpra is released under the terms of the GNU GPL v2, or, at your option, any
 # later version. See the file COPYING for details.
 
+import time
 import unittest
 
-from xpra.codecs.xor.cyxor import xor_str
+from xpra.os_util import strtobytes
+try:
+    from xpra.codecs.xor.cyxor import xor_str       #@UnresolvedImport
+except:
+    xor_str = None
 import binascii
 def h(v):
     return binascii.hexlify(v)
 
+
 class TestHMAC(unittest.TestCase):
 
     def fail_xor(self, in1, in2):
@@ -19,19 +25,19 @@ def fail_xor(self, in1, in2):
         except:
             return
         raise Exception("xor_str did not fail on %s / %s", h(in1), h(in2))
-    
+
     def check_xor(self, in1, in2, expected):
         out = xor_str(in1, in2)
         #print("xor_str(%s, %s)=%s" % (h(in1), h(in2), h(out)))
         assert out==expected
-    
+
     def test_xor_str(self):
-        zeroes  = chr(0)*16
-        ones    = chr(1)*16
-        ff      = chr(255)*16
-        fe      = chr(254)*16
-        empty   = ""
-        lstr    = "\0x80"*64
+        zeroes  = strtobytes(chr(0)*16)
+        ones    = strtobytes(chr(1)*16)
+        ff      = strtobytes(chr(255)*16)
+        fe      = strtobytes(chr(254)*16)
+        empty   = b""
+        lstr    = b"\0x80"*64
         self.check_xor(zeroes, zeroes, zeroes)
         self.check_xor(ones, ones, zeroes)
         self.check_xor(ff, ones, fe)
@@ -41,14 +47,13 @@ def test_xor_str(self):
         self.fail_xor(empty, zeroes)
         self.fail_xor(lstr, ff)
         self.fail_xor(bool, int)
-    
+
 
     def test_large_xor_speed(self):
-        import time
         start = time.time()
         size = 1*1024*1024       #1MB
-        zeroes  = chr(0)*size
-        ones    = chr(1)*size
+        zeroes  = strtobytes(chr(0)*size)
+        ones    = strtobytes(chr(1)*size)
         count = 10
         for _ in range(count):
             self.check_xor(zeroes, ones, ones)

src/tests/unit/net/HMAC_test.py

@@ -10,10 +10,12 @@
 class TestHMAC(unittest.TestCase):
 
     def test_hardcoded(self):
-        password = "71051d81d27745b59c1c56c6e9046c19697e452453e04aa5abbd52c8edc8c232"
-        salt = "99ea464f-7117-4e38-95b3-d3aa80e7b806"
+        password = b"71051d81d27745b59c1c56c6e9046c19697e452453e04aa5abbd52c8edc8c232"
+        salt = b"99ea464f-7117-4e38-95b3-d3aa80e7b806"
         hmac_hash = hmac.HMAC(password, salt)
-        assert hmac_hash.hexdigest()=="dc26a074c9378b1b5735a27563320a26"
+        hd = hmac_hash.hexdigest()
+        ed = "dc26a074c9378b1b5735a27563320a26"
+        assert hd == ed, "expected digest %s but got %s" % (ed, hd)
 
 def main():
     unittest.main()

src/tests/unit/net/subprocess_wrapper_test.py

@@ -6,8 +6,10 @@
 
 import unittest
 
-import gobject
+from xpra.gtk_common.gobject_compat import import_gobject, import_glib
+gobject = import_gobject()
 gobject.threads_init()
+glib = import_glib()
 
 from xpra.log import enable_debug_for
 enable_debug_for("all")
@@ -89,16 +91,18 @@ def end(*args):
         def timeout_error():
             self.timeout = True
             mainloop.quit()
-        gobject.timeout_add(500, timeout_error)
-        gobject.idle_add(lp.send, "foo", "hello foo")
-        gobject.idle_add(lp.send, "bar", "hello bar")
-        gobject.idle_add(lp.send, "end")
+        glib.timeout_add(500, timeout_error)
+        sent_str = b"hello foo"
+        glib.idle_add(lp.send, "foo", sent_str)
+        glib.idle_add(lp.send, "bar", b"hello bar")
+        glib.idle_add(lp.send, "end")
         lp.stop = mainloop.quit
         #run!
         lp.start()
         mainloop.run()
         assert len(readback)==1, "expected 1 record in loopback but got %s" % len(readback)
-        assert readback[0][0] == "hello foo"
+        rss = readback[0][0]
+        assert rss== sent_str, "expected message string '%s' but got '%s'" % (sent_str, rss)
         assert self.timeout is False, "the test did not exit cleanly (not received the 'end' packet?)"
 
     def test_loopback_callee(self):
@@ -117,17 +121,19 @@ def test_signal_function(*args):
         def timeout_error():
             self.timeout = True
             lc.stop()
-        gobject.timeout_add(500, timeout_error)
-        gobject.idle_add(callee.emit, "test-signal", "hello foo")
+        glib.timeout_add(500, timeout_error)
+        signal_string = b"hello foo"
+        glib.idle_add(callee.emit, "test-signal", signal_string)
         #hook up a stop function call which ends this test cleanly
         def stop(*args):
             lc.stop()
         callee.stop = stop
-        gobject.idle_add(lc.send, "stop")
+        glib.idle_add(lc.send, "stop")
         #run!
         lc.start()
         assert len(readback)==1, "expected 1 record in loopback but got %s" % len(readback)
-        assert readback[0][0] == "hello foo"
+        rss = readback[0][0]
+        assert rss== signal_string, "expected signal string '%s' but got '%s'" % (signal_string, rss)
         assert self.timeout is False, "the test did not exit cleanly (not received the 'end' packet?)"
 
 def main():

src/tests/unit/os_util_test.py

@@ -23,7 +23,7 @@ def check(self, str_value):
         mvb = memoryview_to_bytes(mv)
         mvs = bytestostr(mvb)
         assert mvs==str_value
-        
+
 
     def test_strs(self):
         for l in (1, 16, 255):

src/tests/unit/run.py

@@ -23,8 +23,11 @@ def run_file(p):
         #ie: "unit.version_util_test"
         name = p[len(root)+1:-3].replace(os.path.sep, ".")
         print("running %s" % name)
+        cmd = [p]
+        if sys.version_info[0]>=3:
+            cmd.insert(0, "python%s" % sys.version_info[0])
         try:
-            proc = subprocess.Popen(p)
+            proc = subprocess.Popen(cmd)
         except:
             print("failed to execute %s" % p)
             sys.exit(1)

src/tests/unit/server/video_subregion_test.py

@@ -5,7 +5,8 @@
 # later version. See the file COPYING for details.
 
 import time
-import gobject
+from xpra.gtk_common.gobject_compat import import_gobject
+gobject = import_gobject()
 gobject.threads_init()
 
 import unittest

src/tests/unit/util_test.py

@@ -19,14 +19,14 @@ def _test_IntegerClass(self, IntegerClass):
         a = IntegerClass()
         a.increase()
         a.decrease()
-        self.assertEquals(a, 0)
+        self.assertEqual(int(a), 0)
         a.increase(10)
-        self.assertEquals(a.get(), 10)
+        self.assertEqual(int(a.get()), 10)
         a.decrease()        #9
-        self.assertLess(a, 10)
+        self.assertLess(int(a), 10)
         a.decrease(19)
-        self.assertGreaterEqual(a, -10)
-        self.assertEquals(IntegerClass(int(str(a))), -10)
+        self.assertGreaterEqual(int(a), -10)
+        self.assertEqual(int(IntegerClass(int(str(a)))), -10)
 
     def test_AtomicInteger(self):
         self._test_IntegerClass(AtomicInteger)
@@ -48,34 +48,34 @@ def test_typedict(self):
     def _test_values_type(self, d, getter, value_types, values_allowed=[]):
         for k in d.keys():
             v = getter(k)
-            self.assertIsNotNone(v)
+            self.assertIsNotNone(v, "expected value for %s key '%s'" % (type(k), k))
             self.assertIn(type(v), value_types)
             if values_allowed:
-                self.assertIn(v, values_allowed)
+                self.assertIn(v, values_allowed, "unexpected value for %s" % k)
 
     def test_strget(self):
         d = typedict({b"bytekey"    : b"bytevalue",
                       u"unicodekey" : u"unicodevalue"})
         self._test_values_type(d, d.strget, [str, unicode])
 
     def test_intget(self):
-        d = typedict({b"bytekey"    : b"1",
+        d = typedict({b"bytekey"    : "1",
                       u"unicodekey" : 2,
                       996           : 3.14})
         self._test_values_type(d, d.intget, [int], [1, 2, 3])
 
     def test_boolget(self):
-        d = typedict({"empty-string-is-false"       : "",
-                      "False boolean stays as it is": False,
-                      "zero is False"               : 0})
+        d = typedict({b"empty-string-is-false"          : b"",
+                      b"False boolean stays as it is"   : False,
+                      b"zero is False"                  : 0})
         self._test_values_type(d, d.boolget, [bool], [False])
-        d = typedict({"non-empty-string-is-true"    : "hello",
-                      "True boolean stays as it is" : True,
-                      "non-zero number is True"     : -1})
+        d = typedict({b"non-empty-string-is-true"       : "hello",
+                      b"True boolean stays as it is"    : True,
+                      b"non-zero number is True"        : -1})
         self._test_values_type(d, d.boolget, [bool], [True])
 
     def test_dictget(self):
-        d = typedict({"nested" : {}})
+        d = typedict({b"nested" : {}})
         self._test_values_type(d, d.dictget, [dict])
 
     #def intpair(self, k, default_value=None):
@@ -97,32 +97,32 @@ def test_updict(self):
               1         : 2}
         d = {}
         updict(d, "d1", d1)
-        self.assertEquals(d.get("d1.foo"), "bar")
+        self.assertEqual(d.get("d1.foo"), "bar")
         self.assertIsNone(d.get("d2"))
         updict(d, "d2", d2)
-        self.assertEquals(d.get("d2.1"), 2)
+        self.assertEqual(d.get("d2.1"), 2)
         #TODO: test suffix stuff
 
     def test_pver(self):
-        self.assertEquals(pver(""), "")
-        self.assertEquals(pver("any string"), "any string")
-        self.assertEquals(pver((1, 2, 3)), "1.2.3")
+        self.assertEqual(pver(""), "")
+        self.assertEqual(pver("any string"), "any string")
+        self.assertEqual(pver((1, 2, 3)), "1.2.3")
 
     def test_std(self):
-        self.assertEquals(std(""), "")
-        self.assertEquals(std("abcd"), "abcd")
-        self.assertEquals(std("r1"), "r1")
+        self.assertEqual(std(""), "")
+        self.assertEqual(std("abcd"), "abcd")
+        self.assertEqual(std("r1"), "r1")
         for invalid in ("*", "\n", "\r"):
-            self.assertEquals(std(invalid), "")
-            self.assertEquals(std("a"+invalid+"b"), "ab")
+            self.assertEqual(std(invalid), "")
+            self.assertEqual(std("a"+invalid+"b"), "ab")
 
     def test_alnum(self):
-        self.assertEquals(alnum("!\"$%^&*()_+{}:@~\\<>?/*-"), "")
-        self.assertEquals(alnum("aBcD123"), "aBcD123")
+        self.assertEqual(alnum("!\"$%^&*()_+{}:@~\\<>?/*-"), "")
+        self.assertEqual(alnum("aBcD123"), "aBcD123")
 
     def test_nonl(self):
-        self.assertEquals(nonl("\n\r"), "\\n\\r")
-        self.assertEquals(nonl("A\nB\rC"), "A\\nB\\rC")
+        self.assertEqual(nonl("\n\r"), "\\n\\r")
+        self.assertEqual(nonl("A\nB\rC"), "A\\nB\\rC")
 
     def test_xor(self):
         self.assertEqual(xor("A", "a"), xor("B", "b"))

src/xpra/child_reaper.py

@@ -41,6 +41,7 @@ def reaper_cleanup():
         return
     singleton.reap()
     singleton.poll()
+    singleton = None
 
 
 class ProcInfo(object):

src/xpra/util.py

@@ -163,10 +163,10 @@ class typedict(dict):
     log = Logger("util")
 
     def capsget(self, key, default=None):
-        k = key
-        if type(key)==str:
-            k = strtobytes(key)
-        v = self.get(k, default)
+        v = self.get(key)
+        #py3k and bytes as keys...
+        if v is None and type(key)==str:
+            v = self.get(strtobytes(key), default)
         if sys.version >= '3' and type(v)==bytes:
             v = bytestostr(v)
         return v