97 lines
3.7 KiB
Python
97 lines
3.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import unittest
|
|
import psycopg2
|
|
from tests.stub_psycopg2 import *
|
|
|
|
|
|
class StubPsycopg2Tests(unittest.TestCase):
|
|
def test_AuthenticationFailure(self):
|
|
with self.assertRaises(psycopg2.OperationalError) as context:
|
|
StubPsycopg2().add_auth_failure().connect()
|
|
self.assertIn("authentication failed", str(context.exception))
|
|
|
|
def test_ConnectionTimeout(self):
|
|
with self.assertRaises(psycopg2.OperationalError) as context:
|
|
StubPsycopg2().add_conn_timeout().connect()
|
|
self.assertIn("timeout expired", str(context.exception))
|
|
|
|
def test_ConnectionFailure(self):
|
|
with self.assertRaises(psycopg2.OperationalError) as context:
|
|
StubPsycopg2().add_conn_failure().connect()
|
|
self.assertIn("could not connect to server", str(context.exception))
|
|
|
|
def test_AdminStartup(self):
|
|
with self.assertRaises(psycopg2.OperationalError) as context:
|
|
StubPsycopg2().add_admin_startup().connect()
|
|
self.assertIn("system is starting up", str(context.exception))
|
|
|
|
def test_AdminShutdown(self):
|
|
with self.assertRaises(psycopg2.errors.AdminShutdown) as context:
|
|
StubPsycopg2().add_admin_shutdown().connect()
|
|
self.assertIn("terminating connection", str(context.exception))
|
|
|
|
def test_GivenNoFixtures_Connect_RaisesIndexError(self):
|
|
with self.assertRaises(IndexError) as context:
|
|
StubPsycopg2().connect()
|
|
|
|
def test_SuccessfulConnect(self):
|
|
conn = StubConnection()
|
|
StubPsycopg2().add_connection(conn)
|
|
|
|
|
|
class StubConnectionTests(unittest.TestCase):
|
|
def test_GivenLiveConnection_Cursor_ReturnsCursor(self):
|
|
cursor = StubCursor()
|
|
conn = StubConnection(cursor)
|
|
self.assertIs(cursor, conn.cursor())
|
|
|
|
def test_GivenClosedConnection_Cursor_ReturnsInterfaceError(self):
|
|
conn = StubConnection()
|
|
conn.close()
|
|
with self.assertRaises(psycopg2.InterfaceError):
|
|
conn.cursor()
|
|
|
|
def test_IntegrationWithPsycopg2Stub(self):
|
|
cursor1 = StubCursor()
|
|
cursor2 = StubCursor()
|
|
conn = StubConnection(cursor1, cursor2)
|
|
pg = StubPsycopg2(conn)
|
|
|
|
conn_result = pg.connect()
|
|
cursor1_result = conn_result.cursor()
|
|
cursor2_result = conn_result.cursor()
|
|
|
|
self.assertIs(conn, conn_result)
|
|
self.assertIs(cursor1, cursor1_result)
|
|
self.assertIs(cursor2, cursor2_result)
|
|
|
|
class StubCursorTests(unittest.TestCase):
|
|
def test_GivenNoExecuteCalled_FetchOne_RaisesProgrammingError(self):
|
|
cursor = StubCursor()
|
|
with self.assertRaises(psycopg2.ProgrammingError):
|
|
cursor.fetchone()
|
|
|
|
def test_GivenExecuteCalled_WithoutResultRows_FetchOneReturnsNone(self):
|
|
cursor = StubCursor(('OK', None))
|
|
cursor.execute("SELECT query")
|
|
self.assertIs(None, cursor.fetchone())
|
|
self.assertEqual("SELECT query", cursor.query)
|
|
self.assertEqual("OK", cursor.statusmessage)
|
|
|
|
def test_GivenExecuteCalled_WithResultRows_FetchOneRetursRow(self):
|
|
cursor = StubCursor(
|
|
("STATUS1", [['value1', 'value2'], ['value3', 'value4']]),
|
|
("STATUS2", [['value5', 'value6'], ['value7', 'value8']]))
|
|
|
|
cursor.execute("SELECT query 1")
|
|
self.assertEqual("STATUS1", cursor.statusmessage)
|
|
self.assertEqual("value1", cursor.fetchone()[0])
|
|
self.assertEqual("value3", cursor.fetchone()[0])
|
|
|
|
cursor.execute("SELECT query 2")
|
|
self.assertEqual("STATUS2", cursor.statusmessage)
|
|
self.assertEqual("value6", cursor.fetchone()[1])
|
|
self.assertEqual("value8", cursor.fetchone()[1])
|
|
self.assertIs(None, cursor.fetchone())
|