ppf/new: In test_top.py: create databases in setUp and discard them in tearDown.

This was necessary to avoid command line initialization before running tests.

diff --git a/ppf/new/tests/test_top.py b/ppf/new/tests/test_top.py
index 97091eb..0389fb3 100644 (file)
--- a/ppf/new/tests/test_top.py
+++ b/ppf/new/tests/test_top.py
@@ -23,6 +23,13 @@ class EnhanceSwarmDescriptionTest(unittest.TestCase):
     # Class specific variables. Initialized in setUp, used throughout tests.
     config_file = "config.sample.ini"
     access_file = "access.sample.ini"
+    mysql_create_script = "p2p-log-mysql.sql"
+    mysql_init_script = "p2p-init-test-mysql.sql"
+    sqlite_create_script = "p2p-log-sqlite.sql"
+    sqlite_init_script = "p2p-init-test.sql"
+    database = {"database": "p2p_test", "user": "root",
+                "password": "p2p4th3m45535"}
+    sqlite_db = "test.db"
     sd = None
     sw = None
     expected_swarm_id = 0
@@ -44,6 +51,59 @@ class EnhanceSwarmDescriptionTest(unittest.TestCase):
         return session_ids
 
     def setUp(self):
+        """Create MySQL and SQLite databases and instantiate objects."""
+        # Create MySQL database. Create tables and indices.
+        # TODO: Check exceptions.
+
+        sql_create_script_path = os.path.join(os.path.dirname(__file__),
+                                              self.mysql_create_script)
+
+        self.conn = MySQLdb.Connection(user=self.database['user'],
+                                       passwd=self.database['password'])
+        self.cursor = self.conn.cursor()
+
+        self.cursor.execute("CREATE DATABASE %s" %self.database['database'])
+        self.cursor.execute("USE %s" %self.database['database'])
+
+        f = open(sql_create_script_path, 'r')
+        p = subprocess.Popen(["mysql", self.database['database'],
+                              "--user=%s" %self.database['user'],
+                              "--password=%s" %self.database['password']],
+                             stdin=f)
+        ret = os.waitpid(p.pid, 0)[1]
+        f.close()
+
+        # Populate MySQL database.
+        # TODO: Check exceptions.
+        sql_init_script_path = os.path.join(os.path.dirname(__file__),
+                                            self.mysql_init_script)
+
+        f = open(sql_init_script_path, 'r')
+        p = subprocess.Popen(["mysql", self.database['database'],
+                              "--user=%s" %self.database['user'],
+                              "--password=%s" %self.database['password']],
+                             stdin=f)
+        ret = os.waitpid(p.pid, 0)[1]
+        f.close()
+
+        # Create SQLite database file. Create tables and indices.
+        # TODO: Check exceptions.
+        sql_create_script_path = os.path.join(os.path.dirname(__file__),
+                                              self.sqlite_create_script)
+        f = open(sql_create_script_path, 'r')
+        p = subprocess.Popen(["sqlite3", self.sqlite_db], stdin=f)
+        ret = os.waitpid(p.pid, 0)[1]
+        f.close()
+
+        # Populate database.
+        # TODO: Check exceptions.
+        sql_init_script_path = os.path.join(os.path.dirname(__file__),
+                                            self.sqlite_init_script)
+        f = open(sql_init_script_path, 'r')
+        p = subprocess.Popen(["sqlite3", self.sqlite_db], stdin=f)
+        ret = os.waitpid(p.pid, 0)[1]
+        f.close()
+
         # Initialize SwarmDescription and AccessConfig instances
         config_file_path = os.path.join(os.path.dirname(__file__),
                                         self.config_file)
@@ -57,6 +117,14 @@ class EnhanceSwarmDescriptionTest(unittest.TestCase):
         self.expected_session_ids = [3, 4, 5, 6]
         self.expected_swarm_id = 3
 
+    def tearDown(self):
+        """Close connection and remove SQLite / MySQL databases."""
+        os.remove(self.sqlite_db)
+
+        self.cursor.execute("DROP DATABASE %s" %self.database['database'])
+        self.cursor.close()
+        self.conn.close()
+
     def test_correct_swarm_id(self):
         top.enhance_swarm_description(self.sd, self.sw)
         swarm_id = self.get_swarm_id()