diff --git a/bin/weedb/NOTES.md b/bin/weedb/NOTES.md index fd44623a..f08ee038 100644 --- a/bin/weedb/NOTES.md +++ b/bin/weedb/NOTES.md @@ -1,16 +1,29 @@ This table shows how the various MySQLdb and sqlite exceptions are mapped to a weedb exception. -| weedb class | Sqlite class | MySQLdb class | MySQLdb error number | Description | -|:--------------|:---------------------|:-------------------|:--------------------:|:--------------------------------| -| | *N/A* | `OperationalError` | 2002 | Server down | -| | *N/A* | `OperationalError` | 2005 | Unknown host | -| | *N/A* | `OperationalError` | 1045 | Bad or non-existent password | -| | *N/A* | `OperationalError` | 1008 | Drop non-existent database | -| | `OperationalError` | `OperationalError` | 1044 | No permission | -| | *N/A* | `OperationalError` | 1049 | Open non-existent database | -| | *N/A* | `ProgrammingError` | 1007 | Database already exists | -| | `OperationalError` | `OperationalError` | 1050 | Table already exists | -| | `OperationalError` | `ProgrammingError` | 1146 | SELECT non-existing table | -| | `OperationalError` | `OperationalError` | 1054 | SELECT non-existent column | -| | *N/A* | `ProgrammingError` | 1146 | SELECT on non-existent database | -| | `IntegrityError` | `IntegrityError` | 1062 | Duplicate key | \ No newline at end of file +#weewx Version 3.6 or earlier: + +| weedb class | Sqlite class | MySQLdb class | MySQLdb error number | Description | +|--------------------|--------------------|--------------------|:--------------------:|---------------------------------| +| `CannotConnect` | *N/A* | `OperationalError` | 2002 | Server down | +| `OperationalError` | *N/A* | `OperationalError` | 2005 | Unknown host | +| `OperationalError` | *N/A* | `OperationalError` | 1045 | Bad or non-existent password | +| `NoDatabase` | *N/A* | `OperationalError` | 1008 | Drop non-existent database | +| `NoDatabase` | `OperationalError` | `OperationalError` | 1044 | No permission | +| `OperationalError` | *N/A* | `OperationalError` | 1049 | Open non-existent database | +| `DatabaseExists` | *N/A* | `ProgrammingError` | 1007 | Database already exists | +| `OperationalError` | `OperationalError` | `OperationalError` | 1050 | Table already exists | +| `ProgrammingError` | `OperationalError` | `ProgrammingError` | 1146 | SELECT non-existing table | +| `OperationalError` | `OperationalError` | `OperationalError` | 1054 | SELECT non-existing column | +| `ProgrammingError` | *N/A* | `ProgrammingError` | 1146 | SELECT on non-existing database | +| `IntegrityError` | `IntegrityError` | `IntegrityError` | 1062 | Duplicate key | + +Exception hierarchy + +~~~ +StandardError -> weedb.DatabaseError -> weedb.IntegrityError + weedb.ProgrammingError + weedb.OperationalError + weedb.DatabaseExists + weedb.CannotConnect + weedb.NoDatabase +~~~ \ No newline at end of file diff --git a/bin/weedb/test/check_mysql.py b/bin/weedb/test/check_mysql.py index 5fc0cb45..138dae03 100644 --- a/bin/weedb/test/check_mysql.py +++ b/bin/weedb/test/check_mysql.py @@ -6,6 +6,7 @@ # It uses two MySQL users, weewx1 and weewx2. The companion # script "setup_mysql" will set them up with the necessary permissions. # +from __future__ import with_statement import unittest import MySQLdb @@ -59,10 +60,12 @@ class TestMySQL(unittest.TestCase): self.assertEqual(e.exception[0], 1008) def test_drop_nopermission(self): - with Cursor(user='weewx2', passwd='weewx2') as cursor: - with self.assertRaises(OperationalError) as e: - cursor.execute("DROP DATABASE test_weewx1") - self.assertEqual(e.exception[0], 1044) + with Cursor(user='weewx1', passwd='weewx1') as cursor1: + cursor1.execute("CREATE DATABASE test_weewx1") + with Cursor(user='weewx2', passwd='weewx2') as cursor2: + with self.assertRaises(OperationalError) as e: + cursor2.execute("DROP DATABASE test_weewx1") + self.assertEqual(e.exception[0], 1044) def test_create_nopermission(self): with Cursor(user='weewx2', passwd='weewx2') as cursor: diff --git a/bin/weedb/check_sqlite.py b/bin/weedb/test/check_sqlite.py similarity index 98% rename from bin/weedb/check_sqlite.py rename to bin/weedb/test/check_sqlite.py index 4f758dad..e6059b0e 100644 --- a/bin/weedb/check_sqlite.py +++ b/bin/weedb/test/check_sqlite.py @@ -6,6 +6,7 @@ # It uses two MySQL users, weewx1 and weewx2. The companion # script "setup_mysql" will set them up with the necessary permissions. # +from __future__ import with_statement import unittest import sys import os diff --git a/bin/weedb/test/test_errors.py b/bin/weedb/test/test_errors.py new file mode 100644 index 00000000..94147357 --- /dev/null +++ b/bin/weedb/test/test_errors.py @@ -0,0 +1,189 @@ +"""Test the weedb exception hierarchy""" +from __future__ import with_statement +import unittest + +import MySQLdb + +import weedb + +# +# For these tests to work, the database for sqdb1 must in a place where you have write permissions, +# and the database for sqdb2 must be in a place where you do NOT have write permissions +sqdb1_dict = {'database_name': '/var/tmp/sqdb1.sdb', 'driver':'weedb.sqlite', 'timeout': '2'} +sqdb2_dict = {'database_name': '/usr/local/sqdb2.sdb', 'driver':'weedb.sqlite', 'timeout': '2'} +mysql1_dict = {'database_name': 'test_weewx1', 'user':'weewx1', 'password':'weewx1', 'driver':'weedb.mysql'} +mysql2_dict = {'database_name': 'test_weewx1', 'user':'weewx2', 'password':'weewx2', 'driver':'weedb.mysql'} + + +# Double check that we have the necessary permissions (or lack thereof): +try: + fd = open(sqdb1_dict['database_name'], 'w') + fd.close() +except: + print >>sys.stderr, "For tests to work properly, you must have permission to write to '%s'." % sqdb1_dict['database_name'] + print >>sys.stderr, "Change the permissions and try again." +try: + fd = open(sqdb2_dict['database_name'], 'w') + fd.close() +except IOError: + pass +else: + print >>sys.stderr, "For tests to work properly, you must NOT have permission to write to '%s'." % sqdb2_dict['database_name'] + print >>sys.stderr, "Change the permissions and try again." + +class Cursor(object): + """Class to be used to wrap a cursor in a 'with' clause.""" + def __init__(self, db_dict): + self.connection = weedb.connect(db_dict) + self.cursor = self.connection.cursor() + + def __enter__(self): + return self.cursor + + def __exit__(self, etyp, einst, etb): # @UnusedVariable + self.cursor.close() + self.connection.close() + +class Common(unittest.TestCase): + + def setUp(self): + try: + weedb.drop(mysql1_dict) + except weedb.NoDatabase: + pass + try: + weedb.drop(mysql2_dict) + except weedb.NoDatabase: + pass + try: + weedb.drop(sqdb1_dict) + except weedb.NoDatabase: + pass + try: + weedb.drop(sqdb2_dict) + except weedb.NoDatabase: + pass + + def test_bad_host(self): + mysql_dict = dict(mysql1_dict) + mysql_dict['host'] = 'foohost' + # TODO: SHould be CannotConnect, which inherits from OperationalError + with self.assertRaises(weedb.OperationalError): + weedb.connect(mysql_dict) + + def test_bad_password(self): + mysql_dict = dict(mysql1_dict) + mysql_dict['password'] = 'badpw' + #TODO: Should be PasswordError, which inherits from OperationalError + with self.assertRaises(weedb.OperationalError): + weedb.connect(mysql_dict) + + def test_drop_nonexistent_database(self): + with self.assertRaises(weedb.NoDatabase): + weedb.drop(mysql1_dict) + with self.assertRaises(weedb.NoDatabase): + weedb.drop(sqdb1_dict) + + def test_drop_nopermission(self): + weedb.create(mysql1_dict) + #TODO: Should be NoPermission + with self.assertRaises(weedb.NoDatabase): + weedb.drop(mysql2_dict) + with self.assertRaises(weedb.NoDatabase): + weedb.drop(sqdb2_dict) + + def test_create_nopermission(self): + with self.assertRaises(weedb.OperationalError): + weedb.create(mysql2_dict) + # TODO: The following test fails. It should return weedb.OperationalError + import sqlite3 + with self.assertRaises(sqlite3.OperationalError): + weedb.create(sqdb2_dict) + + def test_double_db_create(self): + weedb.create(mysql1_dict) + with self.assertRaises(weedb.DatabaseExists): + weedb.create(mysql1_dict) + weedb.create(sqdb1_dict) + with self.assertRaises(weedb.DatabaseExists): + weedb.create(sqdb1_dict) + + def test_open_nonexistent_database(self): + with self.assertRaises(weedb.OperationalError): + connect=weedb.connect(mysql1_dict) + with self.assertRaises(weedb.OperationalError): + connect=weedb.connect(sqdb1_dict) + + def test_select_nonexistent_database(self): + mysql_dict = dict(mysql1_dict) + mysql_dict.pop('database_name') + connect = weedb.connect(mysql_dict) + cursor = connect.cursor() + with self.assertRaises(weedb.ProgrammingError): + cursor.execute("SELECT foo from test_weewx1.bar") + cursor.close() + connect.close() + + # There's no analogous operation with sqlite. You + # must create the database in order to open it. + + def test_select_nonexistent_table(self): + def test(db_dict): + weedb.create(db_dict) + connect = weedb.connect(db_dict) + cursor = connect.cursor() + cursor.execute("CREATE TABLE bar (col1 int, col2 int)") + with self.assertRaises(weedb.ProgrammingError) as e: + cursor.execute("SELECT foo from fubar") + cursor.close() + connect.close() + + test(mysql1_dict) + test(sqdb1_dict) + + def test_double_table_create(self): + def test(db_dict): + weedb.create(db_dict) + connect = weedb.connect(db_dict) + cursor = connect.cursor() + cursor.execute("CREATE TABLE bar (col1 int, col2 int)") + with self.assertRaises(weedb.OperationalError) as e: + cursor.execute("CREATE TABLE bar (col1 int, col2 int)") + cursor.close() + connect.close() + + test(mysql1_dict) + test(sqdb1_dict) + + def test_select_nonexistent_column(self): + def test(db_dict): + weedb.create(db_dict) + connect = weedb.connect(db_dict) + cursor = connect.cursor() + cursor.execute("CREATE TABLE bar (col1 int, col2 int)") + with self.assertRaises(weedb.OperationalError) as e: + cursor.execute("SELECT foo from bar") + cursor.close() + connect.close() + + test(mysql1_dict) + test(sqdb1_dict) + + def test_duplicate_key(self): + def test(db_dict): + weedb.create(db_dict) + connect = weedb.connect(db_dict) + cursor = connect.cursor() + cursor.execute("CREATE TABLE test1 ( dateTime INTEGER NOT NULL UNIQUE PRIMARY KEY, col1 int, col2 int)") + cursor.execute("INSERT INTO test1 (dateTime, col1, col2) VALUES (1, 10, 20)") + with self.assertRaises(weedb.IntegrityError) as e: + cursor.execute("INSERT INTO test1 (dateTime, col1, col2) VALUES (1, 30, 40)") + cursor.close() + connect.close() + + test(mysql1_dict) + test(sqdb1_dict) + + +if __name__ == '__main__': + unittest.main()