[WIP] Added test cases

Tribler/Test/Community/Tunnel/processes/test_line_util.py

@@ -0,0 +1,114 @@
+import struct
+import unittest
+
+from Tribler.community.tunnel.processes.line_util import (fix_split,
+                                                          pack_data,
+                                                          unpack_data,
+                                                          unpack_complex)
+
+BINARY_STRING_ALL_CHARS = "".join([chr(i) for i in range(256)])
+
+
+class TestLineUtil(unittest.TestCase):
+
+    def test_fix_split_correct_single(self):
+        args = [BINARY_STRING_ALL_CHARS]
+        out = fix_split(1, "", args)
+
+        self.assertEqual(out, args)
+
+    def test_fix_split_correct_double(self):
+        args = ["test", BINARY_STRING_ALL_CHARS]
+        out = fix_split(2, "", args)
+
+        self.assertEqual(out, args)
+
+    def test_fix_split_correct_many(self):
+        args = [BINARY_STRING_ALL_CHARS] * 20
+        out = fix_split(20, "", args)
+
+        self.assertEqual(out, args)
+
+    def test_fix_split_broken_single(self):
+        delim = chr(128)
+        args = ["test"] + BINARY_STRING_ALL_CHARS.split(delim)
+        out = fix_split(2, delim, args)
+
+        self.assertEqual(out, ["test", BINARY_STRING_ALL_CHARS])
+
+    def test_fix_split_broken_double(self):
+        delim = chr(128)
+        args = (["test"]
+                + BINARY_STRING_ALL_CHARS.split(delim)
+                + BINARY_STRING_ALL_CHARS.split(delim))
+        out = fix_split(2, delim, args)
+
+        self.assertEqual(out, ["test", BINARY_STRING_ALL_CHARS
+                                       + delim
+                                       + BINARY_STRING_ALL_CHARS])
+
+    def test_pack_data_empty(self):
+        out = pack_data("")
+
+        self.assertEqual(len(out), 9)
+
+        l = struct.unpack("Q", out[:8])[0]
+
+        self.assertEqual(l, 1)
+        self.assertEqual(out[-1], "\n")
+
+    def test_pack_data_full(self):
+        out = pack_data(BINARY_STRING_ALL_CHARS)
+
+        self.assertEqual(len(out), len(BINARY_STRING_ALL_CHARS) + 9)
+
+        l = struct.unpack("Q", out[:8])[0]
+
+        self.assertEqual(l, len(BINARY_STRING_ALL_CHARS) + 1)
+        self.assertEqual(out[8:-1], BINARY_STRING_ALL_CHARS)
+        self.assertEqual(out[-1], "\n")
+
+    def test_unpack_data_incomplete(self):
+        data = "0000000"
+        l, out = unpack_data(data)
+
+        self.assertGreater(l, len(data))
+        self.assertEqual(out, data)
+
+    def test_unpack_data_empty(self):
+        data = pack_data("")
+        l, out = unpack_data(data)
+
+        self.assertEqual(out, "")
+        self.assertEqual(l, len(out) + 9)
+        self.assertEqual(l, len(data))
+
+    def test_unpack_data_full(self):
+        data = pack_data(BINARY_STRING_ALL_CHARS)
+        l, out = unpack_data(data)
+
+        self.assertEqual(out, BINARY_STRING_ALL_CHARS)
+        self.assertEqual(l, len(out) + 9)
+        self.assertEqual(l, len(data))
+
+    def test_unpack_complex_incomplete(self):
+        data = pack_data(BINARY_STRING_ALL_CHARS)[:-2]
+        keep, share = unpack_complex(data)
+
+        self.assertEqual(keep, data)
+        self.assertEqual(share, None)
+
+    def test_unpack_complex_complete(self):
+        data = pack_data(BINARY_STRING_ALL_CHARS)
+        keep, share = unpack_complex(data)
+
+        self.assertEqual(keep, "")
+        self.assertEqual(share, BINARY_STRING_ALL_CHARS)
+
+    def test_unpack_complex_overflow(self):
+        remainder = "test"
+        data = pack_data(BINARY_STRING_ALL_CHARS)
+        keep, share = unpack_complex(data + remainder)
+
+        self.assertEqual(keep, remainder)
+        self.assertEqual(share, BINARY_STRING_ALL_CHARS)

Tribler/Test/Community/Tunnel/remotes/test_remote_object.py

@@ -0,0 +1,172 @@
+import unittest
+
+from Tribler.community.tunnel.remotes.remote_object import RemoteObject, shared
+
+BINARY_STRING_ALL_CHARS = "".join([chr(i) for i in range(256)])
+
+
+class MockShared(RemoteObject):
+
+    @shared
+    def shared_normal(self):
+        pass
+
+    @shared()
+    def shared_normal_parentheses(self):
+        pass
+
+    @shared(False)
+    def shared_normal_explicit(self):
+        pass
+
+    @shared(True)
+    def shared_id(self):
+        pass
+
+
+class TestRemoteObject(unittest.TestCase):
+
+    def test_dirty_startup(self):
+        mock = MockShared()
+
+        self.assertFalse(mock.__is_dirty__())
+
+    def test_dirty_changed_normal(self):
+        mock = MockShared()
+        mock.shared_normal = "test"
+
+        self.assertTrue(mock.__is_dirty__())
+
+    def test_dirty_changed_normal_parentheses(self):
+        mock = MockShared()
+        mock.shared_normal_parentheses = "test"
+
+        self.assertTrue(mock.__is_dirty__())
+
+    def test_dirty_changed_normal_explicit(self):
+        mock = MockShared()
+        mock.shared_normal_explicit = "test"
+
+        self.assertTrue(mock.__is_dirty__())
+
+    def test_dirty_changed_shared_id(self):
+        mock = MockShared()
+        mock.shared_id = "test"
+
+        self.assertTrue(mock.__is_dirty__())
+
+    def test_serialize_with_bool(self):
+        mock = MockShared()
+        mock.shared_normal = True
+
+        s = MockShared.__serialize__(mock)
+        _, out = MockShared.__unserialize__(s)
+
+        self.assertEqual(out.shared_normal, True)
+
+    def test_serialize_with_int(self):
+        mock = MockShared()
+        mock.shared_normal = 3
+
+        s = MockShared.__serialize__(mock)
+        _, out = MockShared.__unserialize__(s)
+
+        self.assertEqual(out.shared_normal, 3)
+
+    def test_serialize_with_float(self):
+        mock = MockShared()
+        mock.shared_normal = 3.14
+
+        s = MockShared.__serialize__(mock)
+        _, out = MockShared.__unserialize__(s)
+
+        self.assertEqual(out.shared_normal, 3.14)
+
+    def test_serialize_with_str(self):
+        mock = MockShared()
+        mock.shared_normal = BINARY_STRING_ALL_CHARS
+
+        s = MockShared.__serialize__(mock)
+        _, out = MockShared.__unserialize__(s)
+
+        self.assertEqual(out.shared_normal, BINARY_STRING_ALL_CHARS)
+
+    def test_serialize_with_list(self):
+        a = [True, 3, 3.14, BINARY_STRING_ALL_CHARS]
+
+        mock = MockShared()
+        mock.shared_normal = a
+
+        s = MockShared.__serialize__(mock)
+        _, out = MockShared.__unserialize__(s)
+
+        self.assertListEqual(out.shared_normal, a)
+
+    def test_serialize_with_complex_nested(self):
+        mock = MockShared()
+        mock.shared_normal = [[chr(255),],]
+
+        s = MockShared.__serialize__(mock)
+
+        self.assertEqual(s, "")
+
+    def test_serialize_with_pointer(self):
+        mock = MockShared()
+        mock.shared_normal = lambda _: None
+
+        with self.assertRaises(TypeError):
+            MockShared.__serialize__(mock)
+
+    def test_serialize_only_update_size(self):
+        mock = MockShared()
+        mock.shared_normal = "test"
+
+        dirty = MockShared.__serialize__(mock)
+        clean = MockShared.__serialize__(mock)
+
+        self.assertLess(len(clean), len(dirty))
+
+    def test_serialize_only_update(self):
+        mock = MockShared()
+        mock.shared_normal = "test"
+
+        dirty = MockShared.__serialize__(mock, True)
+        full = MockShared.__serialize__(mock, False)
+
+        _, out_dirty = MockShared.__unserialize__(dirty)
+        _, out_full = MockShared.__unserialize__(full)
+
+        self.assertEqual(out_dirty.shared_normal, "test")
+        self.assertEqual(out_dirty.shared_normal, out_full.shared_normal)
+
+    def test_unserialize_unknown(self):
+        known = { "test2": MockShared() }
+        mock = MockShared()
+        mock.shared_id = "test"
+
+        s = MockShared.__serialize__(mock)
+        out_id, out = MockShared.__unserialize__(s, known)
+
+        self.assertEqual(out_id, "test")
+        self.assertEqual(out_id, mock.shared_id)
+        self.assertNotEqual(out, known["test2"])
+
+    def test_unserialize_known(self):
+        known = {"test": MockShared()}
+        mock = MockShared()
+        mock.shared_id = "test"
+
+        s = MockShared.__serialize__(mock)
+        out_id, out = MockShared.__unserialize__(s, known)
+
+        self.assertEqual(out_id, "test")
+        self.assertEqual(out_id, mock.shared_id)
+        self.assertEqual(out, known["test"])
+
+    def test_extract_class_name(self):
+        mock = MockShared()
+
+        s = MockShared.__serialize__(mock)
+        out = RemoteObject.__extract_class_name__(s)
+
+        self.assertEqual(out, "MockShared")

Tribler/Test/Community/Tunnel/remotes/test_sync_dict.py

@@ -0,0 +1,96 @@
+import unittest
+
+from Tribler.community.tunnel.remotes.remote_object import RemoteObject, shared
+from Tribler.community.tunnel.remotes.sync_dict import SyncDict
+
+
+class MockShared(RemoteObject):
+
+    @shared
+    def field1(self):
+        pass
+
+    @shared
+    def field2(self):
+        pass
+
+    @shared(True)
+    def field_id(self):
+        pass
+
+
+class TestSyncDict(unittest.TestCase):
+
+    def setUp(self):
+        self.called = False
+
+    def test_is_same_type(self):
+        sync_dict = SyncDict(MockShared)
+
+        self.assertTrue(sync_dict.is_same_type(MockShared.__name__))
+
+    def test_is_not_same_type(self):
+        sync_dict = SyncDict(MockShared)
+
+        self.assertFalse(sync_dict.is_same_type(SyncDict.__name__))
+
+    def test_callback_when_dirty(self):
+        def callback(_):
+            self.called = True
+
+        sync_dict = SyncDict(MockShared, callback=callback)
+        mock = MockShared()
+        mock.field_id = "test"
+        sync_dict[mock.field_id] = mock
+
+        sync_dict.synchronize()
+
+        self.assertTrue(self.called)
+
+    def test_no_callback_when_not_dirty(self):
+        def callback(_):
+            self.called = True
+
+        sync_dict = SyncDict(MockShared, callback=callback)
+
+        sync_dict.synchronize()
+
+        self.assertFalse(self.called)
+
+    def test_on_synchronize(self):
+        sync_dict2 = SyncDict(MockShared)
+        sync_dict = SyncDict(MockShared,
+                             callback=sync_dict2.on_synchronize)
+        mock = MockShared()
+        mock.field_id = "test"
+        sync_dict[mock.field_id] = mock
+
+        self.assertNotIn(mock.field_id, sync_dict2)
+
+        sync_dict.synchronize()
+
+        self.assertIn(mock.field_id, sync_dict2)
+
+    def test_synchronize_only_update(self):
+        sync_dict2 = SyncDict(MockShared)
+        sync_dict = SyncDict(MockShared,
+                             callback=sync_dict2.on_synchronize)
+        mock = MockShared()
+        mock.field_id = "test"
+        mock.field1 = "a"
+        mock.field2 = "b"
+        sync_dict[mock.field_id] = mock
+
+        sync_dict.synchronize(False)
+
+        self.assertEqual(sync_dict2[mock.field_id].field1, "a")
+        self.assertEqual(sync_dict2[mock.field_id].field2, "b")
+
+        sync_dict2[mock.field_id].field1 = "x"
+        sync_dict2[mock.field_id].field2 = "d"
+        mock.field1 = "c"
+
+        sync_dict.synchronize(True)
+
+        self.assertEqual(sync_dict2[mock.field_id].field1, "c")
+        self.assertEqual(sync_dict2[mock.field_id].field2, "d")

Tribler/community/tunnel/processes/line_util.py

@@ -65,7 +65,7 @@ def unpack_complex(line):
         return line, data
     elif target < len(line):
         # keep past target-8, share up to target-8
-        return data[target-8:], data[:target-8]
+        return line[target:], data[:target-9]
     return line, None