Merge pull request #8 from jaykwon/master

Add SQLite checker

preflyt/checkers/sqlite.py

@@ -0,0 +1,33 @@
+"""Checks for SQLite data"""
+
+import os
+
+from preflyt.base import BaseChecker
+
+
+class SqliteChecker(BaseChecker):
+    """Verify that a SQLite3 DB exists and is readable."""
+
+    checker_name = "sqlite"
+
+    def __init__(self, path='db.sqlite3'):
+        """Initialize the checker
+
+        :param path: The path to the SQLite3 database.
+
+        """
+        super().__init__()
+        self._path = path
+
+    def check(self):
+        if not os.path.exists(self._path):
+            return False, "SQLite3 DB at path {} does not exist".format(self._path)
+
+        with open(self._path, 'rb') as infile:
+            header = infile.read(100)
+
+        passed = header[:15] == b'SQLite format 3'
+        if passed:
+            return True, "The DB {} is present and valid.".format(self._path)
+        return False, "The DB {} does not appear to be a \
+                       valid SQLite3 file.".format(self._path)

tests/test_sqlite.py

@@ -0,0 +1,73 @@
+"""Test the checkers"""
+import sqlite3
+import os
+from functools import wraps
+import tempfile
+
+from nose.tools import ok_, eq_
+
+from preflyt.checkers.sqlite import SqliteChecker
+
+
+def bootstrap_db(path):
+    def wrapper(func):
+        def create_table(path):
+            conn = sqlite3.connect(path)
+            cursor = conn.cursor()
+            cursor.execute('CREATE TABLE wet (stuntz text)')
+            conn.commit()
+            conn.close()
+
+        def remove_table(path):
+            os.remove(path)
+
+        @wraps(func)
+        def wrapped(*args, **kwargs):
+            create_table(path)
+            result = func(*args, **kwargs)
+            remove_table(path)
+            return result
+        return wrapped
+    return wrapper
+
+def test_init():
+    sqlite = SqliteChecker()
+    eq_(SqliteChecker.checker_name, "sqlite")
+    eq_(sqlite._path, "db.sqlite3")
+
+def test_init_path():
+    sqlite = SqliteChecker(path="test.db")
+    eq_(sqlite._path, "test.db")
+
+@bootstrap_db('db.sqlite3')
+def test_exists():
+    path = 'db.sqlite3'
+    sqlite = SqliteChecker(path=path)
+    eq_(sqlite._path, path)
+
+def test_not_exists():
+    path = 'db.sqliteeeee'
+    sqlite = SqliteChecker(path=path)
+    result, message = sqlite.check()
+    ok_(not result)
+    ok_("does not exist" in message, message)
+
+@bootstrap_db('db.sqlite3')
+def test_valid():
+    path = 'db.sqlite3'
+    sqlite = SqliteChecker(path=path)
+    result, message = sqlite.check()
+    ok_(result)
+    ok_("present and valid" in message, message)
+
+def test_invalid():
+    filehandle, filepath = tempfile.mkstemp()
+    try:
+        os.write(filehandle, b'Not valid')
+        os.close(filehandle)
+        sqlite = SqliteChecker(path=filepath)
+        result, message = sqlite.check()
+    finally:
+        os.remove(filepath)
+    ok_(not result)
+    ok_("does not appear" in message, message)