Complete basic tests for access.py.

diff --git a/ppf/db/tests/test_access_sqlite.py b/ppf/db/tests/test_access_sqlite.py
index bc0d59c..e326db3 100644 (file)
--- a/ppf/db/tests/test_access_sqlite.py
+++ b/ppf/db/tests/test_access_sqlite.py
@@ -117,8 +117,6 @@ class AccessTest(unittest.TestCase):
 
         self.assertNotEqual(conn, None)
 
-        dba.disconnect()
-
     def test_disconnect(self):
         dba = access.DatabaseAccess()
         conn = dba.connect(self.database)
@@ -133,10 +131,8 @@ class AccessTest(unittest.TestCase):
 
         self.assertNotEqual(cursor, None)
 
-        dba.disconnect()
-
     def test_select_swarms(self):
-        # Needs fixing. An entry should be added in the startup script.
+        # TODO: An entry should be added in the startup script.
         dba = access.SQLiteDatabaseAccess()
         dba.connect(self.database)
         cursor = dba.select_swarms(False)
@@ -185,61 +181,225 @@ class AccessTest(unittest.TestCase):
         self.assertEqual(num_rows, 0)
 
     def test_select_btclients(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        cursor = dba.select_btclients(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 6)
 
     def test_select_btclients_by_name(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        cursor = dba.select_btclients_by_name("Tribler", False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 1)
 
     def test_select_btclients_id_by_name(self):
-        self.assertEqual(True, True)
+        # TODO: unify return values for select_* methods in access.py
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        id = dba.select_btclients_id_by_name("Tribler")
+
+        self.assertEqual(id, 1)
 
     def test_insert_btclients_row(self):
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
         self.assertEqual(True, True)
 
     def test_insert_btclients(self):
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
         self.assertEqual(True, True)
 
     def test_select_client_sessions_by_id(self):
-        self.assertEqual(True, True)
+        # TODO: An entry should be added in the startup script.
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        cursor = dba.select_client_sessions_by_id(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
     def test_select_client_sessions_by_swarm(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        cursor = dba.select_client_sessions_by_swarm(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
     def test_insert_client_sessions_row(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        cursor = dba.select_client_sessions_by_id(False)
 
-    def test_insert_client_sessions(self):
-        self.assertEqual(True, True)
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 1)
 
     def test_delete_client_sessions_by_id(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.delete_client_sessions_by_id(1)
+        cursor = dba.select_client_sessions_by_id(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
     def test_delete_client_sessions_by_swarm(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.delete_client_sessions_by_swarm(1)
+        cursor = dba.select_client_sessions_by_id(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
     def test_select_status_messages(self):
-        self.assertEqual(True, True)
+        # TODO: An entry should be added in the startup script.
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        cursor = dba.select_status_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
     def test_insert_status_messages_row(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.insert_status_messages_row(['1', '2455128.10', '222', '0', '213', '56', '200', '300', '121.324'])
+
+        cursor = dba.select_status_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 1)
 
     def test_insert_status_messages(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.insert_status_messages('1', '2455128.10', '222', '0', '213', '56', '200', '300', '121.324')
+
+        cursor = dba.select_status_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 1)
 
     def test_delete_status_messages(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.insert_status_messages('1', '2455128.10', '222', '0', '213', '56', '200', '300', '121.324')
+        dba.delete_status_messages()
+
+        cursor = dba.select_status_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
     def test_select_verbose_messages(self):
-        self.assertEqual(True, True)
+        # TODO: An entry should be added in the startup script.
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        cursor = dba.select_verbose_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
     def test_insert_verbose_messages_row(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.insert_verbose_messages_row(['1', '2455128.121295811', '0', '127.0.0.1', '1345', '0', '3', '4', '13', '777'])
+
+        cursor = dba.select_verbose_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 1)
 
     def test_insert_verbose_messages(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.insert_verbose_messages('1', '2455128.121295811', '0', '127.0.0.1', '1345', '0', '3', '4', '13', '777')
+
+        cursor = dba.select_verbose_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 1)
+
 
     def test_delete_verbose_messages(self):
-        self.assertEqual(True, True)
+        dba = access.SQLiteDatabaseAccess()
+        dba.connect(self.database)
+        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        dba.insert_verbose_messages_row(['1', '2455128.121295811', '0', '127.0.0.1', '1345', '0', '3', '4', '13', '777'])
+        dba.delete_verbose_messages()
+
+        cursor = dba.select_verbose_messages(False)
+
+        num_rows = 0
+        for row in cursor:
+            num_rows = num_rows+1
+
+        self.assertEqual(num_rows, 0)
 
 #
 #        if len(sys.argv) != 2: