Improve PgDB class and add unit tests

mylib/pgsql.py

@@ -11,6 +11,45 @@ import psycopg2
 log = logging.getLogger(__name__)
 
 
+#
+# Exceptions
+#
+
+class PgDBException(Exception):
+    """ That is the base exception class for all the other exceptions provided by this module. """
+
+    def __init__(self, error, *args, **kwargs):
+        for arg, value in kwargs.items():
+            setattr(self, arg, value)
+        super().__init__(error.format(*args, **kwargs))
+
+
+class PgDBDuplicatedSQLParameter(PgDBException, KeyError):
+    """
+    Raised when trying to set a SQL query parameter
+    and an other parameter with the same name is already set
+    """
+
+    def __init__(self, parameter_name):
+        super().__init__(
+            "Duplicated SQL parameter '{parameter_name}'",
+            parameter_name=parameter_name
+        )
+
+
+class PgDBUnsupportedWHEREClauses(PgDBException, TypeError):
+    """
+    Raised when trying to execute query with unsupported
+    WHERE clauses provided
+    """
+
+    def __init__(self, where_clauses):
+        super().__init__(
+            "Unsupported WHERE clauses: {where_clauses}",
+            where_clauses=where_clauses
+        )
+
+
 class PgDB:
     """ PostgreSQL client """
 
@@ -62,13 +101,24 @@ class PgDB:
                 self.con.set_client_encoding(enc)
                 return True
             except Exception:
-                log.error('An error occured setting Postgresql database connection encoding to "%s"', enc, exc_info=1)
+                log.error(
+                    'An error occured setting Postgresql database connection encoding to "%s"',
+                    enc, exc_info=1
+                )
         return False
 
     def doSQL(self, sql, params=None):
-        """ Run SELECT SQL query and return result as dict """
+        """
+        Run SQL query and commit changes (rollback on error)
+
+        :param sql: The SQL query
+        :param params: The SQL query's parameters as dict (optional)
+
+        :return: True on success, False otherwise
+        :rtype: bool
+        """
         if self.just_try:
-            log.debug(u"Just-try mode : do not really execute SQL query '%s'", sql)
+            log.debug("Just-try mode : do not really execute SQL query '%s'", sql)
             return True
 
         cursor = self.con.cursor()
@@ -80,134 +130,281 @@ class PgDB:
             self.con.commit()
             return True
         except Exception:
-            log.error(u'Error during SQL request "%s"', sql.decode('utf-8', 'ignore'), exc_info=1)
+            log.error(
+                'Error during SQL request "%s" %s',
+                sql,
+                "with params = %s" % ', '.join([
+                    "%s = %s" % (key, value)
+                    for key, value in params.items()
+                ]) if params else "without params",
+                exc_info=True
+            )
             self.con.rollback()
             return False
 
-    def doSelect(self, sql, params):
+    def doSelect(self, sql, params=None):
-        """ Run SELECT SQL query and return result as dict """
+        """
+        Run SELECT SQL query and return list of selected rows as dict
+
+        :param sql: The SQL query
+        :param params: The SQL query's parameters as dict (optional)
+
+        :return: List of selected rows as dict on success, False otherwise
+        :rtype: list, bool
+        """
         cursor = self.con.cursor()
         try:
             cursor.execute(sql, params)
             results = cursor.fetchall()
             return results
         except Exception:
-            log.error(u'Error during SQL request "%s"', sql.decode('utf-8', 'ignore'), exc_info=1)
+            log.error(
-            return False
+                'Error during SQL request "%s" %s',
+                sql,
+                "with params = %s" % ', '.join([
+                    "%s = %s" % (key, value)
+                    for key, value in params.items()
+                ]) if params else "without params",
+                exc_info=True
+            )
+        return False
+
 
     #
     # SQL helpers
     #
-    def _quote_value(self, value):
+
+    @classmethod
+    def _combine_params(cls, params, to_add=None, **kwargs):
+        if to_add:
+            assert isinstance(to_add, dict), "to_add must be a dict or None"
+            params = cls._combine_params(params, **to_add)
+
+        for param, value in kwargs.items():
+            if param in params:
+                raise PgDBDuplicatedSQLParameter(param)
+            params[param] = value
+        return params
+
+    @classmethod
+    def _format_where_clauses(cls, where_clauses, params=None, where_op=None):
+        """
+        Format WHERE clauses
+
+        :param where_clauses: The WHERE clauses. Could be:
+            - a raw SQL WHERE clause as string
+            - a tuple of two elements: a raw WHERE clause and its parameters as dict
+            - a dict of WHERE clauses with field name as key and WHERE clause value as value
+            - a list of any of previous valid WHERE clauses
+        :param params: Dict of other already set SQL query parameters (optional)
+        :param where_op: SQL operator used to combine WHERE clauses together (optional, default: AND)
+
+        :return: A tuple of two elements: raw SQL WHERE combined clauses and parameters on success
+        :rtype: string, bool
+        """
+        if params is None:
+            params = dict()
+        if where_op is None:
+            where_op = 'AND'
+
+        if isinstance(where_clauses, str):
+            return (where_clauses, params)
+
+        if isinstance(where_clauses, tuple) and len(where_clauses) == 2 and isinstance(where_clauses[1], dict):
+            cls._combine_params(params, where_clauses[1])
+            return (where_clauses[0], params)
+
+        if isinstance(where_clauses, (list, tuple)):
+            sql_where_clauses = []
+            for where_clause in where_clauses:
+                sql2, params = cls._format_where_clauses(where_clause, params=params, where_op=where_op)
+                sql_where_clauses.append(sql2)
+            return (
+                (" %s " % where_op).join(sql_where_clauses),
+                params
+            )
+
+        if isinstance(where_clauses, dict):
+            cls._combine_params(params, where_clauses)
+            return (
+                (" %s " % where_op).join([
+                    '"{0}" = %({0})s'.format(key)
+                    for key in where_clauses
+                ]),
+                params
+            )
+        raise PgDBUnsupportedWHEREClauses(where_clauses)
+
+    @classmethod
+    def _add_where_clauses(cls, sql, params, where_clauses, where_op=None):
+        """
+        Add WHERE clauses to an SQL query
+
+        :param sql: The SQL query to complete
+        :param params: The dict of parameters of the SQL query to complete
+        :param where_clauses: The WHERE clause (see _format_where_clauses())
+        :param where_op: SQL operator used to combine WHERE clauses together (optional, default: see _format_where_clauses())
+
+        :return:
+        :rtype: A tuple of two elements: raw SQL WHERE combined clauses and parameters
+        """
+        if where_clauses:
+            sql_where, params = cls._format_where_clauses(where_clauses, params=params, where_op=where_op)
+            sql += " WHERE " + sql_where
+        return (sql, params)
+
+
+    def insert(self, table, values, just_try=False):
+        """ Run INSERT SQL query """
+        sql = 'INSERT INTO "%s" ("%s") VALUES (%s)' % (
+            table,
+            '", "'.join(values.keys()),
+            ", ".join([
+                '%({0})s'.format(key)
+                for key in values
+            ])
+        )
+
+        if just_try:
+            log.debug("Just-try mode : execute INSERT query : %s", sql)
+            return True
+
+        log.debug(sql)
+        if not self.doSQL(sql, params=values):
+            log.error("Fail to execute INSERT query (SQL : %s)", sql)
+            return False
+        return True
+
+    def update(self, table, values, where_clauses, where_op=None, just_try=False):
+        """ Run UPDATE SQL query """
+        sql = 'UPDATE "%s" SET %s' % (
+            table,
+            ", ".join([
+                '"{0}" = %({0})s'.format(key)
+                for key in values
+            ])
+        )
+        params = values
+
+        try:
+            sql, params = self._add_where_clauses(sql, params, where_clauses, where_op=where_op)
+        except (PgDBDuplicatedSQLParameter, PgDBUnsupportedWHEREClauses):
+            log.error('Fail to add WHERE clauses', exc_info=True)
+            return False
+
+        if just_try:
+            log.debug("Just-try mode : execute UPDATE query : %s", sql)
+            return True
+
+        log.debug(sql)
+        if not self.doSQL(sql, params=params):
+            log.error("Fail to execute UPDATE query (SQL : %s)", sql)
+            return False
+        return True
+
+    def delete(self, table, where_clauses, where_op='AND', just_try=False):
+        """ Run DELETE SQL query """
+        sql = 'DELETE FROM "%s"' % table
+        params = dict()
+
+        try:
+            sql, params = self._add_where_clauses(sql, params, where_clauses, where_op=where_op)
+        except (PgDBDuplicatedSQLParameter, PgDBUnsupportedWHEREClauses):
+            log.error('Fail to add WHERE clauses', exc_info=True)
+            return False
+
+        if just_try:
+            log.debug("Just-try mode : execute UPDATE query : %s", sql)
+            return True
+
+        log.debug(sql)
+        if not self.doSQL(sql, params=params):
+            log.error("Fail to execute UPDATE query (SQL : %s)", sql)
+            return False
+        return True
+
+    def truncate(self, table, just_try=False):
+        """ Run TRUNCATE SQL query """
+
+        sql = 'TRUNCATE "%s"' % table
+
+        if just_try:
+            log.debug("Just-try mode : execute TRUNCATE query : %s", sql)
+            return True
+
+        log.debug(sql)
+        if not self.doSQL(sql):
+            log.error("Fail to execute TRUNCATE query (SQL : %s)", sql)
+            return False
+        return True
+
+    def select(self, table, where_clauses=None, fields=None, where_op='AND', order_by=None, just_try=False):
+        """ Run SELECT SQL query """
+        sql = "SELECT "
+        if fields is None:
+            sql += "*"
+        elif isinstance(fields, str):
+            sql += '"%s"' % fields
+        else:
+            sql += '"' + '", "'.join(fields) + '"'
+
+        sql += ' FROM "%s"' % table
+        params = dict()
+
+        try:
+            sql, params = self._add_where_clauses(sql, params, where_clauses, where_op=where_op)
+        except (PgDBDuplicatedSQLParameter, PgDBUnsupportedWHEREClauses):
+            log.error('Fail to add WHERE clauses', exc_info=True)
+            return False
+
+        if order_by:
+            sql += ' ORDER BY %s' % order_by
+
+        if just_try:
+            log.debug("Just-try mode : execute SELECT query : %s", sql)
+            return just_try
+
+        return self.doSelect(sql, params=params)
+
+
+    #
+    # Depreated helpers
+    #
+
+    @classmethod
+    def _quote_value(cls, value):
         """ Quote a value for SQL query """
+        if value is None:
+            return 'NULL'
+
         if isinstance(value, (int, float)):
             return str(value)
 
         if isinstance(value, datetime.datetime):
-            value = self._format_datetime(value)
+            value = cls._format_datetime(value)
         elif isinstance(value, datetime.date):
-            value = self._format_date(value)
+            value = cls._format_date(value)
 
-        return u"'%s'" % value.replace(u"'", u"''")
+        return "'%s'" % value.replace("'", "''")
 
-    def _format_where_clauses(self, where_clauses, where_op=u'AND'):
+    @classmethod
-        """ Format WHERE clauses """
+    def _format_datetime(cls, value):
-        if isinstance(where_clauses, str):
-            return where_clauses
-        if isinstance(where_clauses, list):
-            return (u" %s " % where_op).join(where_clauses)
-        if isinstance(where_clauses, dict):
-            return (u" %s " % where_op).join(map(lambda x: "%s=%s" % (x, self._quote_value(where_clauses[x])), where_clauses))
-        log.error('Unsupported where clauses type %s', type(where_clauses))
-        return False
-
-    def _format_datetime(self, value):
         """ Format datetime object as string """
         assert isinstance(value, datetime.datetime)
-        return value.strftime(self.datetime_format)
+        return value.strftime(cls.datetime_format)
 
-    def _format_date(self, value):
+    @classmethod
+    def _format_date(cls, value):
         """ Format date object as string """
         assert isinstance(value, (datetime.date, datetime.datetime))
-        return value.strftime(self.date_format)
+        return value.strftime(cls.date_format)
 
-    def time2datetime(self, time):
+    @classmethod
+    def time2datetime(cls, time):
         """ Convert timestamp to datetime string """
-        return self._format_datetime(datetime.datetime.fromtimestamp(int(time)))
+        return cls._format_datetime(datetime.datetime.fromtimestamp(int(time)))
 
-    def time2date(self, time):
+    @classmethod
+    def time2date(cls, time):
         """ Convert timestamp to date string """
-        return self._format_date(datetime.date.fromtimestamp(int(time)))
+        return cls._format_date(datetime.date.fromtimestamp(int(time)))
-
-    def insert(self, table, values, just_try=False):
-        """ Run INSERT SQL query """
-        sql = u"INSERT INTO %s (%s) VALUES (%s)" % (table, u', '.join(values.keys()), u", ".join(map(lambda x: self._quote_value(values[x]), values)))
-
-        if just_try:
-            log.debug(u"Just-try mode : execute INSERT query : %s", sql)
-            return True
-
-        log.debug(sql)
-        if not self.doSQL(sql):
-            log.error(u"Fail to execute INSERT query (SQL : %s)", sql)
-            return False
-        return True
-
-    def update(self, table, values, where_clauses, where_op=u'AND', just_try=False):
-        """ Run UPDATE SQL query """
-        where = self._format_where_clauses(where_clauses, where_op=where_op)
-        if not where:
-            return False
-
-        sql = u"UPDATE %s SET  %s WHERE %s" % (table, u", ".join(map(lambda x: "%s=%s" % (x, self._quote_value(values[x])), values)), where)
-
-        if just_try:
-            log.debug(u"Just-try mode : execute UPDATE query : %s", sql)
-            return True
-
-        log.debug(sql)
-        if not self.doSQL(sql):
-            log.error(u"Fail to execute UPDATE query (SQL : %s)", sql)
-            return False
-        return True
-
-    def delete(self, table, where_clauses, where_op=u'AND', just_try=False):
-        """ Run DELETE SQL query """
-        where = self._format_where_clauses(where_clauses, where_op=where_op)
-        if not where:
-            return False
-
-        sql = u"DELETE FROM %s WHERE %s" % (table, where)
-
-        if just_try:
-            log.debug(u"Just-try mode : execute DELETE query : %s", sql)
-            return True
-
-        log.debug(sql)
-        if not self.doSQL(sql):
-            log.error(u"Fail to execute DELETE query (SQL : %s)", sql)
-            return False
-        return True
-
-    def select(self, table, where_clauses=None, fields=None, where_op=u'AND', order_by=None):
-        """ Run SELECT SQL query """
-        sql = u"SELECT "
-        if fields is None:
-            sql += "*"
-        elif isinstance(fields, str):
-            sql += fields
-        else:
-            sql += u", ".join(fields)
-
-        sql += u" FROM " + table
-        if where_clauses:
-            where = self._format_where_clauses(where_clauses, where_op=where_op)
-            if not where:
-                return False
-
-            sql += u" WHERE " + where
-
-        if order_by:
-            sql += u"ORDER %s" % order_by
-
-        return self.doSelect(sql)

tests/test_pgsql.py

@@ -0,0 +1,385 @@
+""" Tests on opening hours helpers """
+
+import pytest
+
+from mylib.pgsql import PgDB
+
+class FakePsycopg2Cursor:
+
+    def __init__(self, expected_sql, expected_params, expected_return, expected_just_try, expected_exception):
+        self.expected_sql = expected_sql
+        self.expected_params = expected_params
+        self.expected_return = expected_return
+        self.expected_just_try = expected_just_try
+        self.expected_exception = expected_exception
+
+    def execute(self, sql, params=None):
+        if self.expected_exception:
+            raise Exception("%s.execute(%s, %s): expected exception" % (self, sql, params))
+        if self.expected_just_try and not sql.lower().startswith('select '):
+            assert False, "%s.execute(%s, %s) may not be executed in just try mode" % (self, sql, params)
+        assert sql == self.expected_sql, "%s.execute(): Invalid SQL query:\n  '%s'\nMay be:\n  '%s'" % (self, sql, self.expected_sql)
+        assert params == self.expected_params, "%s.execute(): Invalid params:\n  %s\nMay be:\n  %s" % (self, params, self.expected_params)
+        return self.expected_return
+
+    def fetchall(self):
+        return self.expected_return
+
+    def __repr__(self):
+        return "FakePsycopg2Cursor(%s, %s, %s, %s)" % (
+            self.expected_sql, self.expected_params,
+            self.expected_return, self.expected_just_try
+        )
+
+
+class FakePsycopg2:
+
+    expected_sql = None
+    expected_params = None
+    expected_return = True
+    expected_just_try = False
+    expected_exception = False
+    just_try = False
+
+    def __init__(self, **kwargs):
+        allowed_kwargs = dict(dbname=str, user=str, password=(str, None), host=str)
+        for arg, value in kwargs.items():
+            assert arg in allowed_kwargs, "Invalid arg %s='%s'" % (arg, value)
+            assert isinstance(value, allowed_kwargs[arg]), "Arg %s not a %s (%s)" % (arg, allowed_kwargs[arg], type(value))
+            setattr(self, arg, value)
+
+    def close(self):
+        return self.expected_return
+
+    def set_client_encoding(self, *arg):
+        self._check_just_try()
+        assert len(arg) == 1 and isinstance(arg[0], str)
+        if self.expected_exception:
+            raise Exception("set_client_encoding(%s): Expected exception", arg[0])
+        return self.expected_return
+
+    def cursor(self):
+        return FakePsycopg2Cursor(
+            self.expected_sql, self.expected_params,
+            self.expected_return, self.expected_just_try or self.just_try,
+            self.expected_exception
+        )
+
+    def commit(self):
+        self._check_just_try()
+        return self.expected_return
+
+    def rollback(self):
+        self._check_just_try()
+        return self.expected_return
+
+    def _check_just_try(self):
+        if self.just_try:
+            assert false, "May not be executed in just try mode"
+
+def fake_psycopg2_connect(**kwargs):
+    return FakePsycopg2(**kwargs)
+
+def fake_psycopg2_connect_just_try(**kwargs):
+    con = FakePsycopg2(**kwargs)
+    con.just_try = True
+    return con
+
+@pytest.fixture
+def test_pgdb():
+    return PgDB('127.0.0.1', 'user', 'password', 'dbname')   
+
+@pytest.fixture
+def fake_pgdb(mocker):
+    mocker.patch('psycopg2.connect', fake_psycopg2_connect)
+    return PgDB('127.0.0.1', 'user', 'password', 'dbname')
+
+@pytest.fixture
+def fake_just_try_pgdb(mocker):
+    mocker.patch('psycopg2.connect', fake_psycopg2_connect_just_try)
+    return PgDB('127.0.0.1', 'user', 'password', 'dbname', just_try=True)
+
+@pytest.fixture
+def fake_connected_pgdb(fake_pgdb):
+    fake_pgdb.connect()
+    return fake_pgdb
+
+@pytest.fixture
+def fake_connected_just_try_pgdb(fake_just_try_pgdb):
+    fake_just_try_pgdb.connect()
+    return fake_just_try_pgdb
+
+def generate_mock_args(expected_args=(), expected_kwargs=dict(), expected_return=True):
+    def mock_args(*args, **kwargs):
+        assert args == expected_args, "Invalid call args:\n  %s\nMay be:\n  %s" % (args, expected_args)
+        assert kwargs == expected_kwargs, "Invalid call kwargs:\n  %s\nMay be:\n  %s" % (kwargs, expected_kwargs)
+        return expected_return
+    return mock_args
+
+def mock_doSQL_just_try(self, sql, params=None):
+    assert False, "doSQL() may not be executed in just try mode"
+
+def generate_mock_doSQL(expected_sql, expected_params=dict(), expected_return=True):
+    def mock_doSQL(self, sql, params=None):
+        assert sql == expected_sql, "Invalid generated SQL query:\n  '%s'\nMay be:\n  '%s'" % (sql, expected_sql)
+        assert params == expected_params, "Invalid generated params:\n  %s\nMay be:\n  %s" % (params, expected_params)
+        return expected_return
+    return mock_doSQL
+
+# PgDB.doSelect() have same expected parameters as PgDB.doSQL()
+generate_mock_doSelect = generate_mock_doSQL
+mock_doSelect_just_try = mock_doSQL_just_try
+
+#
+# Test on PgDB helper methods
+#
+def test_combine_params_with_to_add_parameter():
+    assert PgDB._combine_params(dict(test1=1), dict(test2=2)) == dict(
+        test1=1, test2=2
+    )
+
+def test_combine_params_with_kargs():
+    assert PgDB._combine_params(dict(test1=1), test2=2) == dict(
+        test1=1, test2=2
+    )
+
+def test_combine_params_with_kargs_and_to_add_parameter():
+    assert PgDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict(
+        test1=1, test2=2, test3=3
+    )
+
+def test_format_where_clauses_params_are_preserved():
+    args=('test = test', dict(test1=1))
+    assert PgDB._format_where_clauses(*args) == args
+
+def test_format_where_clauses_raw():
+    assert PgDB._format_where_clauses('test = test') == (('test = test'), dict())
+
+def test_format_where_clauses_tuple_clause_with_params():
+    where_clauses = (
+        'test1 = %(test1)s AND test2 = %(test2)s',
+        dict(test1=1, test2=2)
+    )
+    assert PgDB._format_where_clauses(where_clauses) == where_clauses
+
+def test_format_where_clauses_dict():
+    where_clauses = dict(test1=1, test2=2)
+    assert PgDB._format_where_clauses(where_clauses) == (
+        '"test1" = %(test1)s AND "test2" = %(test2)s',
+        where_clauses
+    )
+
+def test_format_where_clauses_combined_types():
+    where_clauses = (
+        'test1 = 1',
+        ('test2 LIKE %(test2)s', dict(test2=2)),
+        dict(test3=3, test4=4)
+    )
+    assert PgDB._format_where_clauses(where_clauses) == (
+        'test1 = 1 AND test2 LIKE %(test2)s AND "test3" = %(test3)s AND "test4" = %(test4)s',
+        dict(test2=2, test3=3, test4=4)
+    )
+
+def test_format_where_clauses_with_where_op():
+    where_clauses = dict(test1=1, test2=2)
+    assert PgDB._format_where_clauses(where_clauses, where_op='OR') == (
+        '"test1" = %(test1)s OR "test2" = %(test2)s',
+        where_clauses
+    )
+
+def test_add_where_clauses():
+    sql="SELECT * FROM table"
+    where_clauses = dict(test1=1, test2=2)
+    assert PgDB._add_where_clauses(sql, None, where_clauses) == (
+        sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s',
+        where_clauses
+    )
+
+def test_add_where_clauses_preserved_params():
+    sql="SELECT * FROM table"
+    where_clauses = dict(test1=1, test2=2)
+    params = dict(fake1=1)
+    assert PgDB._add_where_clauses(sql, params.copy(), where_clauses) == (
+        sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s',
+        dict(**where_clauses, **params)
+    )
+
+def test_add_where_clauses_with_op():
+    sql="SELECT * FROM table"
+    where_clauses = ('test1=1', 'test2=2')
+    assert PgDB._add_where_clauses(sql, None, where_clauses, where_op='OR') == (
+        sql + ' WHERE test1=1 OR test2=2',
+        dict()
+    )
+
+def test_insert(mocker, test_pgdb):
+    values = dict(test1=1, test2=2)
+    mocker.patch(
+        'mylib.pgsql.PgDB.doSQL',
+        generate_mock_doSQL(
+            'INSERT INTO "mytable" ("test1", "test2") VALUES (%(test1)s, %(test2)s)',
+            values
+        )
+    )
+
+    assert test_pgdb.insert('mytable', values)
+
+def test_insert_just_try(mocker, test_pgdb):
+    mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSQL_just_try)
+    assert test_pgdb.insert('mytable', dict(test1=1, test2=2), just_try=True)
+
+def test_update(mocker, test_pgdb):
+    values = dict(test1=1, test2=2)
+    where_clauses = dict(test3=3, test4=4)
+    mocker.patch(
+        'mylib.pgsql.PgDB.doSQL',
+        generate_mock_doSQL(
+            'UPDATE "mytable" SET "test1" = %(test1)s, "test2" = %(test2)s WHERE "test3" = %(test3)s AND "test4" = %(test4)s',
+            dict(**values, **where_clauses)
+        )
+    )
+
+    assert test_pgdb.update('mytable', values, where_clauses)
+
+def test_update_just_try(mocker, test_pgdb):
+    mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSQL_just_try)
+    assert test_pgdb.update('mytable', dict(test1=1, test2=2), None, just_try=True)
+
+def test_delete(mocker, test_pgdb):
+    where_clauses = dict(test1=1, test2=2)
+    mocker.patch(
+        'mylib.pgsql.PgDB.doSQL',
+        generate_mock_doSQL(
+            'DELETE FROM "mytable" WHERE "test1" = %(test1)s AND "test2" = %(test2)s',
+            where_clauses
+        )
+    )
+
+    assert test_pgdb.delete('mytable', where_clauses)
+
+def test_delete_just_try(mocker, test_pgdb):
+    mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSQL_just_try)
+    assert test_pgdb.delete('mytable', None, just_try=True)
+
+def test_truncate(mocker, test_pgdb):
+    mocker.patch(
+        'mylib.pgsql.PgDB.doSQL',
+        generate_mock_doSQL('TRUNCATE "mytable"', None)
+    )
+
+    assert test_pgdb.truncate('mytable')
+
+def test_truncate_just_try(mocker, test_pgdb):
+    mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSelect_just_try)
+    assert test_pgdb.truncate('mytable', just_try=True)
+
+def test_select(mocker, test_pgdb):
+    fields = ('field1', 'field2')
+    where_clauses = dict(test3=3, test4=4)
+    expected_return = [
+        dict(field1=1, field2=2),
+        dict(field1=2, field2=3),
+    ]
+    order_by = "field1, DESC"
+    mocker.patch(
+        'mylib.pgsql.PgDB.doSelect',
+        generate_mock_doSQL(
+            'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = %(test3)s AND "test4" = %(test4)s ORDER BY ' + order_by,
+            where_clauses, expected_return
+        )
+    )
+
+    assert test_pgdb.select('mytable', where_clauses, fields, order_by=order_by) == expected_return
+
+def test_select_without_field_and_order_by(mocker, test_pgdb):
+    mocker.patch(
+        'mylib.pgsql.PgDB.doSelect',
+        generate_mock_doSQL(
+            'SELECT * FROM "mytable"'
+        )
+    )
+
+    assert test_pgdb.select('mytable')
+
+def test_select_just_try(mocker, test_pgdb):
+    mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSelect_just_try)
+    assert test_pgdb.select('mytable', None, None, just_try=True)
+
+#
+# Tests on main methods
+#
+
+def test_connect(mocker, test_pgdb):
+    expected_kwargs = dict(
+        dbname=test_pgdb.db,
+        user=test_pgdb.user,
+        host=test_pgdb.host,
+        password=test_pgdb.pwd
+    )
+    def mock_psycopg2_connect(**kwargs):
+        assert expected_kwargs == kwargs
+        
+    mocker.patch(
+        'psycopg2.connect',
+        generate_mock_args(
+            expected_kwargs=expected_kwargs
+        )
+    )
+
+    assert test_pgdb.connect()
+
+def test_close(fake_pgdb):
+    assert fake_pgdb.close() is None
+
+def test_close_connected(fake_connected_pgdb):
+    assert fake_connected_pgdb.close() is None
+
+def test_setEncoding(fake_connected_pgdb):
+    assert fake_connected_pgdb.setEncoding('utf8')
+
+def test_setEncoding_not_connected(fake_pgdb):
+    assert fake_pgdb.setEncoding('utf8') is False
+
+def test_setEncoding_on_exception(fake_connected_pgdb):
+    fake_connected_pgdb.con.expected_exception = True
+    assert fake_connected_pgdb.setEncoding('utf8') is False
+
+def test_doSQL(fake_connected_pgdb):
+    fake_connected_pgdb.con.expected_sql = 'DELETE FROM table WHERE test1 = %(test1)s'
+    fake_connected_pgdb.con.expected_params = dict(test1=1)
+    fake_connected_pgdb.doSQL(fake_connected_pgdb.con.expected_sql, fake_connected_pgdb.con.expected_params)
+
+def test_doSQL_without_params(fake_connected_pgdb):
+    fake_connected_pgdb.con.expected_sql = 'DELETE FROM table'
+    fake_connected_pgdb.doSQL(fake_connected_pgdb.con.expected_sql)
+
+def test_doSQL_just_try(fake_connected_just_try_pgdb):
+    assert fake_connected_just_try_pgdb.doSQL('DELETE FROM table')
+
+def test_doSQL_on_exception(fake_connected_pgdb):
+    fake_connected_pgdb.con.expected_exception = True
+    assert fake_connected_pgdb.doSQL('DELETE FROM table') is False
+
+def test_doSelect(fake_connected_pgdb):
+    fake_connected_pgdb.con.expected_sql = 'SELECT * FROM table WHERE test1 = %(test1)s'
+    fake_connected_pgdb.con.expected_params = dict(test1=1)
+    fake_connected_pgdb.con.expected_return = [dict(test1=1)]
+    assert fake_connected_pgdb.doSelect(fake_connected_pgdb.con.expected_sql, fake_connected_pgdb.con.expected_params) == fake_connected_pgdb.con.expected_return
+
+def test_doSelect_without_params(fake_connected_pgdb):
+    fake_connected_pgdb.con.expected_sql = 'SELECT * FROM table'
+    fake_connected_pgdb.con.expected_return = [dict(test1=1)]
+    assert fake_connected_pgdb.doSelect(fake_connected_pgdb.con.expected_sql) == fake_connected_pgdb.con.expected_return
+
+def test_doSelect_on_exception(fake_connected_pgdb):
+    fake_connected_pgdb.con.expected_exception = True
+    assert fake_connected_pgdb.doSelect('SELECT * FROM table') is False
+
+def test_doSelect_just_try(fake_connected_just_try_pgdb):
+    fake_connected_just_try_pgdb.con.expected_sql = 'SELECT * FROM table WHERE test1 = %(test1)s'
+    fake_connected_just_try_pgdb.con.expected_params = dict(test1=1)
+    fake_connected_just_try_pgdb.con.expected_return = [dict(test1=1)]
+    assert fake_connected_just_try_pgdb.doSelect(
+        fake_connected_just_try_pgdb.con.expected_sql,
+        fake_connected_just_try_pgdb.con.expected_params
+    ) == fake_connected_just_try_pgdb.con.expected_return