From d35bfa7c8ec7b7c657ac2d4e48243a8bdd8f9ea2 Mon Sep 17 00:00:00 2001 From: Razvan Deaconescu Date: Thu, 18 Aug 2011 20:45:36 +0300 Subject: [PATCH] Complete basic tests for access.py. --- ppf/db/tests/test_access_sqlite.py | 206 +++++++++++++++++++++++++---- 1 file changed, 183 insertions(+), 23 deletions(-) diff --git a/ppf/db/tests/test_access_sqlite.py b/ppf/db/tests/test_access_sqlite.py index bc0d59c..e326db3 100644 --- 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: -- 2.20.1