Changeset 548


Ignore:
Timestamp:
Nov 19, 2015, 10:39:13 AM (4 years ago)
Author:
cito
Message:

Add tests for direct socket access functions

File:
1 edited

Legend:

Unmodified
Added
Removed
  • branches/4.x/module/TEST_PyGreSQL_classic_connection.py

    r547 r548  
    221221        self.assertIsInstance(r, int)
    222222        self.assertGreaterEqual(r, 0)
    223 
    224 
    225223
    226224
     
    647645
    648646    def tearDown(self):
    649         self.c.query('truncate table test')
     647        self.c.query("truncate table test")
    650648        self.c.close()
    651649
     
    747745
    748746
    749 class TestNoticeReceiver(unittest.TestCase):
    750     """"Test notification support."""
     747class TestDirectSocketAccess(unittest.TestCase):
     748    """"Test copy command with direct socket access."""
     749
     750    @classmethod
     751    def setUpClass(cls):
     752        c = connect()
     753        c.query("drop table if exists test cascade")
     754        c.query("create table test (i int, v varchar(16))")
     755        c.close()
     756
     757    @classmethod
     758    def tearDownClass(cls):
     759        c = connect()
     760        c.query("drop table test cascade")
     761        c.close()
    751762
    752763    def setUp(self):
    753764        self.c = connect()
     765        self.c.query("set datestyle='ISO,YMD'")
     766
     767    def tearDown(self):
     768        self.c.query("truncate table test")
     769        self.c.close()
     770
     771    def testPutline(self):
     772        putline = self.c.putline
     773        query = self.c.query
     774        data = list(enumerate("apple pear plum cherry banana".split()))
     775        query("copy test from stdin")
     776        try:
     777            for i, v in data:
     778                putline("%d\t%s\n" % (i, v))
     779            putline("\\.\n")
     780        finally:
     781            self.c.endcopy()
     782        r = query("select * from test").getresult()
     783        self.assertEqual(r, data)
     784
     785    def testPutline(self):
     786        getline = self.c.getline
     787        query = self.c.query
     788        data = list(enumerate("apple banana pear plum strawberry".split()))
     789        n = len(data)
     790        self.c.inserttable('test', data)
     791        query("copy test to stdout")
     792        try:
     793            for i in range(n + 2):
     794                v = getline()
     795                if i < n:
     796                    self.assertEqual(v, '%d\t%s' % data[i])
     797                elif i == n:
     798                    self.assertEqual(v, '\\.')
     799                else:
     800                    self.assertIsNone(v)
     801        finally:
     802            try:
     803                self.c.endcopy()
     804            except IOError:
     805                pass
     806
     807    def testParameterChecks(self):
     808        self.assertRaises(TypeError, self.c.putline)
     809        self.assertRaises(TypeError, self.c.getline, 'invalid')
     810        self.assertRaises(TypeError, self.c.endcopy, 'invalid')
     811
     812
     813class TestNotificatons(unittest.TestCase):
     814    """"Test notification support."""
     815
     816    def setUp(self):
     817        self.c = connect()
    754818
    755819    def tearDown(self):
     
    757821
    758822    def testGetNotify(self):
    759         getnotify = self.connection.getnotify
     823        getnotify = self.c.getnotify
    760824        query = self.c.query
    761825        self.assertIsNone(getnotify())
Note: See TracChangeset for help on using the changeset viewer.