# Path: norm/interface.py
# class IAsyncCursor(Interface):
# 
# 
#     def execute(query, params=None):
#         """
#         Execute the given sql and params.
# 
#         Return a C{Deferred} which will fire with the DB-API results.
#         """
# 
# 
#     def fetchone():
#         pass
# 
# 
#     def fetchall():
#         pass
# 
# 
#     def lastRowId():
#         """
#         Return a C{Deferred} id of the most recently inserted row.
#         """
# 
# 
#     def close():
#         pass
# 
# Path: norm/common.py
# class BlockingCursor(object):
#     """
#     I wrap a single DB-API2 db cursor in an asynchronous api.
#     """
# 
#     implements(IAsyncCursor)
# 
# 
#     def __init__(self, cursor):
#         self.cursor = cursor
# 
# 
#     def execute(self, sql, params=()):
#         return defer.maybeDeferred(self.cursor.execute, sql, params)
# 
# 
#     def fetchone(self):
#         return defer.maybeDeferred(self.cursor.fetchone)
# 
# 
#     def fetchall(self):
#         return defer.maybeDeferred(self.cursor.fetchall)
# 
# 
#     def lastRowId(self):
#         return defer.succeed(self.cursor.lastrowid)
# 
# 
#     def close(self):
#         return defer.maybeDeferred(self.cursor.close)
# 
# Path: norm/postgres.py
# class PostgresCursorWrapper(object):
# 
# 
#     implements(IAsyncCursor)
# 
# 
#     def __init__(self, cursor):
#         self.cursor = cursor
# 
# 
#     def execute(self, sql, params=()):
#         sql = translateSQL(sql)
#         ret = self.cursor.execute(sql, params)
#         return ret
# 
# 
#     def lastRowId(self):
#         d = self.cursor.execute('select lastval()')
#         d.addCallback(lambda _: self.cursor.fetchone())
#         return d.addCallback(lambda row: row[0])
# 
# 
#     def fetchone(self):
#         return self.cursor.fetchone()
# 
# 
#     def fetchall(self):
#         return self.cursor.fetchall()
# 
# 
#     def close(self):
#         return self.cursor.close()
# 
# Path: norm/test/util.py
# def postgresConnStr():
#     if not postgres_url:
#         raise SkipTest(skip_postgres)
#     return mkConnStr(parseURI(postgres_url))


from twisted.trial.unittest import TestCase
from twisted.internet import defer
from zope.interface.verify import verifyObject
from mock import MagicMock
from norm.interface import IAsyncCursor
from norm.common import BlockingCursor
from norm.postgres import PostgresCursorWrapper
from norm.test.util import postgresConnStr
        import psycopg2
# Copyright (c) Matt Haggard.
# See LICENSE for details.






class PostgresCursorWrapperTest(TestCase):


    def test_IAsyncCursor(self):
        verifyObject(IAsyncCursor, PostgresCursorWrapper(None))


    def assertCallThrough(self, name, *args, **kwargs):
        mock = MagicMock()
        setattr(mock, name, MagicMock(return_value=defer.succeed('foo')))


# Path: norm/interface.py
# class IAsyncCursor(Interface):
# 
# 
#     def execute(query, params=None):
#         """
#         Execute the given sql and params.
# 
#         Return a C{Deferred} which will fire with the DB-API results.
#         """
# 
# 
#     def fetchone():
#         pass
# 
# 
#     def fetchall():
#         pass
# 
# 
#     def lastRowId():
#         """
#         Return a C{Deferred} id of the most recently inserted row.
#         """
# 
# 
#     def close():
#         pass
# 
# Path: norm/common.py
# class BlockingCursor(object):
#     """
#     I wrap a single DB-API2 db cursor in an asynchronous api.
#     """
# 
#     implements(IAsyncCursor)
# 
# 
#     def __init__(self, cursor):
#         self.cursor = cursor
# 
# 
#     def execute(self, sql, params=()):
#         return defer.maybeDeferred(self.cursor.execute, sql, params)
# 
# 
#     def fetchone(self):
#         return defer.maybeDeferred(self.cursor.fetchone)
# 
# 
#     def fetchall(self):
#         return defer.maybeDeferred(self.cursor.fetchall)
# 
# 
#     def lastRowId(self):
#         return defer.succeed(self.cursor.lastrowid)
# 
# 
#     def close(self):
#         return defer.maybeDeferred(self.cursor.close)
# 
# Path: norm/postgres.py
# class PostgresCursorWrapper(object):
# 
# 
#     implements(IAsyncCursor)
# 
# 
#     def __init__(self, cursor):
#         self.cursor = cursor
# 
# 
#     def execute(self, sql, params=()):
#         sql = translateSQL(sql)
#         ret = self.cursor.execute(sql, params)
#         return ret
# 
# 
#     def lastRowId(self):
#         d = self.cursor.execute('select lastval()')
#         d.addCallback(lambda _: self.cursor.fetchone())
#         return d.addCallback(lambda row: row[0])
# 
# 
#     def fetchone(self):
#         return self.cursor.fetchone()
# 
# 
#     def fetchall(self):
#         return self.cursor.fetchall()
# 
# 
#     def close(self):
#         return self.cursor.close()
# 
# Path: norm/test/util.py
# def postgresConnStr():
#     if not postgres_url:
#         raise SkipTest(skip_postgres)
#     return mkConnStr(parseURI(postgres_url))


# Path: norm/test/test_postgres.py
from twisted.trial.unittest import TestCase
from twisted.internet import defer
from zope.interface.verify import verifyObject
from mock import MagicMock
from norm.interface import IAsyncCursor
from norm.common import BlockingCursor
from norm.postgres import PostgresCursorWrapper
from norm.test.util import postgresConnStr
        import psycopg2

# Copyright (c) Matt Haggard.
# See LICENSE for details.






class PostgresCursorWrapperTest(TestCase):


    def test_IAsyncCursor(self):
        verifyObject(IAsyncCursor, PostgresCursorWrapper(None))


    def assertCallThrough(self, name, *args, **kwargs):
        mock = MagicMock()
        setattr(mock, name, MagicMock(return_value=defer.succeed('foo')))


        cursor = PostgresCursorWrapper(mock)