ppf: Add insert_test_swarms method in test suite.

Inserts test entries in the `swarms` table. Select, insert and update
methods are checked against number of entries in the table.

diff --git a/ppf/db/tests/test_access_sqlite.py b/ppf/db/tests/test_access_sqlite.py
index fbb3f15..1a22ae1 100644 (file)
--- a/ppf/db/tests/test_access_sqlite.py
+++ b/ppf/db/tests/test_access_sqlite.py
@@ -3,7 +3,7 @@ import os
 import sqlite3
 import access
 
-sql_script = """
+sql_create_script = """
 drop table if exists status_messages;
 drop table if exists verbose_messages;
 drop table if exists client_sessions;
@@ -80,8 +80,15 @@ insert into btclients(name, language, dht, streaming)
                values('Aria', 'C', 1, 0);
 insert into btclients(name, language, dht, streaming)
                values('Mainline', 'Python', 1, 0);
+"""
+
+sql_init_script = """
+-- insert swarms in `swarms` table
 
--- .genfkey --exec
+insert into swarms(torrent, filesize, purpose, source)
+        values('DarkKnight', '123000', 'experiment', 'TVTorrents');
+insert into swarms(torrent, filesize, purpose, source)
+        values('Fedora', '1024', 'experiment', 'local');
 """
 
 class AccessTest(unittest.TestCase):
@@ -94,12 +101,20 @@ class AccessTest(unittest.TestCase):
     dba = None
     conn = None
     cursor = None
+    expected_swarms_count = 0
+    expected_btclients_count = 0
 
     def setUp(self):
         """ Create database file and instantiate objects. """
         conn = sqlite3.connect(self.database)
         cursor = conn.cursor()
-        cursor.executescript(sql_script)
+
+        cursor.executescript(sql_create_script)
+        cursor.executescript(sql_init_script)
+        # Number of table entries in the init script
+        self.expected_swarms_count = 2
+        self.expected_btclients_count = 6
+
         cursor.close()
         conn.close()
 
@@ -122,6 +137,16 @@ class AccessTest(unittest.TestCase):
 
         return num_rows
 
+    def insert_test_swarms(self):
+        """ Insert test entries through insert_swarms_row method. """
+        self.dba.insert_swarms_row(['Pulp Fiction', '102400',
+            'streaming', 'isohunt'])
+        self.dba.insert_swarms_row(['Karaoke', '50122',
+            'streaming', 'local'])
+
+        # Increment number of entries in the swarm table.
+        self.expected_swarms_count = self.expected_swarms_count + 2
+
     def test_connect(self):
         """ If connection is successful, no exception is raised. """
         try:
@@ -152,36 +177,52 @@ class AccessTest(unittest.TestCase):
             self.assertEqual(True, True)
 
     def test_select_swarms(self):
-        # TODO: An entry should be added in the startup script.
+        """
+        Select initial swarm entries. Check against expected_swarms_count.
+        """
         self.dba.select_swarms(False)
         num_rows = self.get_cursor_num_rows()
 
-        self.assertEqual(num_rows, 0)
+        self.assertEqual(num_rows, self.expected_swarms_count)
 
     def test_insert_swarms_row(self):
-        self.dba.insert_swarms_row(["fedora.torrent", 10000, "btex", "isohunt"])
+        """
+        Use insert_test_swarms to add new entries in swarms table.
+        Test number of rows selected against expected_swarms_count.
+        """
+        self.insert_test_swarms()
 
         self.dba.select_swarms(False)
         num_rows = self.get_cursor_num_rows()
 
-        self.assertEqual(num_rows, 1)
+        self.assertEqual(num_rows, self.expected_swarms_count)
 
     def test_insert_swarms(self):
-        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        """
+        Use insert_test_swarms to add new entries in swarms table.
+        Test number of rows selected against expected_swarms_count.
+        """
+        self.insert_test_swarms()
 
         self.dba.select_swarms(False)
         num_rows = self.get_cursor_num_rows()
 
-        self.assertEqual(num_rows, 1)
+        self.assertEqual(num_rows, self.expected_swarms_count)
 
     def test_delete_swarms(self):
-        self.dba.insert_swarms("fedora.torrent", 10000, "btex", "isohunt")
+        """
+        Use insert_test_swarms to add new entries in swarms table.
+        Delete all swarm entries and check whether number of rows is 0.
+        """
+        self.insert_test_swarms()
         self.dba.delete_swarms()
+        # No rows should longer be present in swarms table.
+        self.expected_swarms_count = 0
 
         self.dba.select_swarms(False)
         num_rows = self.get_cursor_num_rows()
 
-        self.assertEqual(num_rows, 0)
+        self.assertEqual(num_rows, self.expected_swarms_count)
 
     def test_select_btclients(self):
         self.dba.select_btclients(False)