ppf/new: In test_config.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_config.py b/ppf/new/tests/test_config.py
index 40e49e5..f928768 100644 (file)
--- a/ppf/new/tests/test_config.py
+++ b/ppf/new/tests/test_config.py
@@ -9,6 +9,8 @@ import os
 import os.path
 import datetime
 import sys
+import MySQLdb
+import subprocess
 
 import config
 import storage
@@ -215,20 +217,87 @@ class AccessConfigTest(unittest.TestCase):
     # Class specific variables. Initialized in setUp, used throughout tests.
     access_file = "access.sample.ini"
     test_data = None
+    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"
     ac = None
 
     def setUp(self):
+        """Instantiate objects and create MySQL and SQLite databases."""
         self.test_data = {'mysql': {'user': 'root',
                                     'database': 'p2p_test',
                                     'password': 'p2p4th3m45535',
                                     'host': 'localhost'},
                           'sqlite': {'database': 'test.db'}}
 
+        # Create MySQL database. Create tables and indices.
+        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 self.ac as AccessConfig instance
         access_file_path = os.path.join(os.path.dirname(__file__),
                                         self.access_file)
         self.ac = config.AccessConfig()
         self.ac.load(access_file_path)
 
+    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_load(self):
         # Check for the same sections.
         # Use sort() method because self.test_data is unordered.