ppf: Add database connection/initialization in setUp test method.

diff --git a/ppf/db/tests/test_access_sqlite.py b/ppf/db/tests/test_access_sqlite.py
index e326db3..bb397b4 100644 (file)
--- a/ppf/db/tests/test_access_sqlite.py
+++ b/ppf/db/tests/test_access_sqlite.py
@@ -86,317 +86,285 @@ insert into btclients(name, language, dht, streaming)
 
 class AccessTest(unittest.TestCase):
     """
-    Test case for Access class.
+    Test suite for SQLiteDatabaseAccess class.
     """
 
+    # Class specific variables. Initialized in setUp, used throughout tests.
     database = "test.db"
+    dba = None
+    conn = None
+    cursor = None
 
     def setUp(self):
+        """ Create database file and instantiate objects. """
         conn = sqlite3.connect(self.database)
         cursor = conn.cursor()
         cursor.executescript(sql_script)
         cursor.close()
         conn.close()
 
+        self.dba = access.SQLiteDatabaseAccess()
+        self.conn = self.dba.connect(self.database)
+        self.cursor = self.dba.get_cursor()
+
     def tearDown(self):
+        """ Close connection and remove database file. """
+        self.dba.disconnect()
         os.remove(self.database)
 
-    def test_run(self):
-        conn = sqlite3.connect(self.database)
-        cursor = conn.cursor()
-        cursor.execute("""select * from btclients""");
-        for row in cursor:
-            #print row
-            pass
-        cursor.close()
-        conn.close()
-
     def test_connect(self):
-        dba = access.SQLiteDatabaseAccess()
-        conn = dba.connect(self.database)
-
-        self.assertNotEqual(conn, None)
+        """ If connection is successful, no exception is raised. """
+        try:
+            self.cursor.execute("SELECT id FROM swarms LIMIT 1;")
+            # If no exception is raised, all is OK.
+            self.assertEqual(True, True)
+        except sqlite3.Error, e:
+            # Exception is raised. Fail.
+            self.assertEqual(True, False)
 
     def test_disconnect(self):
-        dba = access.DatabaseAccess()
-        conn = dba.connect(self.database)
-        dba.disconnect()
-
-        self.assertEqual(conn, None)
-
-    def test_get_cursor(self):
+        """
+        Create separate connection to test disconnect() method.
+        After disconnect, run an SQL statement. If disconnect was successful
+        an exception should be raised.
+        """
         dba = access.SQLiteDatabaseAccess()
         conn = dba.connect(self.database)
         cursor = dba.get_cursor()
+        dba.disconnect()
 
-        self.assertNotEqual(cursor, None)
+        try:
+            cursor.execute("SELECT id FROM swarms LIMIT 1;")
+            # Exception should be raised. If not, fail.
+            self.assertEqual(True, False)
+        except sqlite3.Error, e:
+            # Exception is raised; disconnect() was successful.
+            self.assertEqual(True, True)
 
     def test_select_swarms(self):
         # TODO: An entry should be added in the startup script.
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        cursor = dba.select_swarms(False)
+        self.dba.select_swarms(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_insert_swarms_row(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        dba.insert_swarms_row(["fedora.torrent", 10000, "btex", "isohunt"])
-        cursor = dba.select_swarms(False)
+        self.dba.insert_swarms_row(["fedora.torrent", 10000, "btex", "isohunt"])
+        self.dba.select_swarms(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
     def test_insert_swarms(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
-        cursor = dba.select_swarms(False)
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.select_swarms(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
     def test_delete_swarms(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
-        dba.delete_swarms()
-        cursor = dba.select_swarms(False)
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.delete_swarms()
+        self.dba.select_swarms(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_select_btclients(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        cursor = dba.select_btclients(False)
+        self.dba.select_btclients(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 6)
 
     def test_select_btclients_by_name(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        cursor = dba.select_btclients_by_name("Tribler", False)
+        self.dba.select_btclients_by_name("Tribler", False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
     def test_select_btclients_id_by_name(self):
         # 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")
+        id = self.dba.select_btclients_id_by_name("Tribler")
 
         self.assertEqual(id, 1)
 
     def test_insert_btclients_row(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
+        # TODO
         self.assertEqual(True, True)
 
     def test_insert_btclients(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
+        # TODO
         self.assertEqual(True, True)
 
     def test_select_client_sessions_by_id(self):
         # 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)
+        self.dba.select_client_sessions_by_id(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_select_client_sessions_by_swarm(self):
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        cursor = dba.select_client_sessions_by_swarm(False)
+        self.dba.select_client_sessions_by_swarm(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_insert_client_sessions_row(self):
-        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)
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.select_client_sessions_by_id(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
     def test_delete_client_sessions_by_id(self):
-        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)
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.delete_client_sessions_by_id(1)
+        self.dba.select_client_sessions_by_id(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_delete_client_sessions_by_swarm(self):
-        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)
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.delete_client_sessions_by_swarm(1)
+        self.dba.select_client_sessions_by_id(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_select_status_messages(self):
         # TODO: An entry should be added in the startup script.
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        cursor = dba.select_status_messages(False)
+        self.dba.select_status_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_insert_status_messages_row(self):
-        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'])
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.insert_status_messages_row(['1', '2455128.10', '222', '0', '213', '56', '200', '300', '121.324'])
 
-        cursor = dba.select_status_messages(False)
+        self.dba.select_status_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
     def test_insert_status_messages(self):
-        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')
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.insert_status_messages('1', '2455128.10', '222', '0', '213', '56', '200', '300', '121.324')
 
-        cursor = dba.select_status_messages(False)
+        self.dba.select_status_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
     def test_delete_status_messages(self):
-        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()
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.insert_status_messages('1', '2455128.10', '222', '0', '213', '56', '200', '300', '121.324')
+        self.dba.delete_status_messages()
 
-        cursor = dba.select_status_messages(False)
+        self.dba.select_status_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_select_verbose_messages(self):
         # TODO: An entry should be added in the startup script.
-        dba = access.SQLiteDatabaseAccess()
-        dba.connect(self.database)
-        cursor = dba.select_verbose_messages(False)
+        self.dba.select_verbose_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)
 
     def test_insert_verbose_messages_row(self):
-        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'])
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.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)
+        self.dba.select_verbose_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
     def test_insert_verbose_messages(self):
-        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')
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.insert_verbose_messages('1', '2455128.121295811', '0', '127.0.0.1', '1345', '0', '3', '4', '13', '777')
 
-        cursor = dba.select_verbose_messages(False)
+        self.dba.select_verbose_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 1)
 
 
     def test_delete_verbose_messages(self):
-        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()
+        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        self.dba.insert_client_sessions_row(['1', '2', 'Linux', '2.6.26', '512', '1500', '141.85.224.205', '50500', '512', '64', '2455125.02409722'])
+        self.dba.insert_verbose_messages_row(['1', '2455128.121295811', '0', '127.0.0.1', '1345', '0', '3', '4', '13', '777'])
+        self.dba.delete_verbose_messages()
 
-        cursor = dba.select_verbose_messages(False)
+        self.dba.select_verbose_messages(False)
 
         num_rows = 0
-        for row in cursor:
+        for row in self.cursor:
             num_rows = num_rows+1
 
         self.assertEqual(num_rows, 0)