From 1b2ed8e94b312b62898d3be1c062c3428fb32247 Mon Sep 17 00:00:00 2001 From: Razvan Deaconescu Date: Fri, 19 Aug 2011 09:50:30 +0300 Subject: [PATCH] ppf: Add database connection/initialization in setUp test method. --- ppf/db/tests/test_access_sqlite.py | 244 +++++++++++++---------------- 1 file changed, 106 insertions(+), 138 deletions(-) diff --git a/ppf/db/tests/test_access_sqlite.py b/ppf/db/tests/test_access_sqlite.py index e326db3..bb397b4 100644 --- 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) -- 2.20.1