Changeset 697


Ignore:
Timestamp:
Jan 7, 2016, 11:59:10 AM (4 years ago)
Author:
cito
Message:

Test rowcount attribute with copy methods

Location:
trunk/module
Files:
2 edited

Legend:

Unmodified
Added
Removed
  • trunk/module/pgdb.py

    r693 r697  
    651651
    652652        def copy():
     653            self.rowcount = 0
    653654            while True:
    654655                row = getdata(decode)
  • trunk/module/tests/test_dbapi20_copy.py

    r694 r697  
    191191    def check_table(self):
    192192        self.assertEqual(self.table_data, self.data)
     193
     194    def check_rowcount(self, number=len(data)):
     195        self.assertEqual(self.cursor.rowcount, number)
    193196
    194197
     
    230233        self.assertIs(ret, self.cursor)
    231234        self.assertEqual(self.table_data, [(42, 'Hello, world!')])
     235        self.check_rowcount(1)
    232236
    233237    def test_input_string_with_newline(self):
    234238        self.copy_from('42\tHello, world!\n')
    235239        self.assertEqual(self.table_data, [(42, 'Hello, world!')])
     240        self.check_rowcount(1)
    236241
    237242    def test_input_string_multiple_rows(self):
     
    239244        self.assertIs(ret, self.cursor)
    240245        self.check_table()
     246        self.check_rowcount()
    241247
    242248    if str is unicode:
     
    261267        self.copy_from(self.data_text.splitlines())
    262268        self.check_table()
     269        self.check_rowcount()
    263270
    264271    def test_input_iterable_with_newlines(self):
     
    300307        self.assertEqual(self.table_data, [
    301308            (1, None), (2, None), (3, 'Three'), (4, 'Four'), (5, 'Five')])
     309        self.check_rowcount(5)
    302310        self.assertRaises(pgdb.ProgrammingError, self.copy_from,
    303311            '6\t42', columns=['id', 'age'])
     312        self.check_rowcount(-1)
    304313
    305314    def test_csv(self):
     
    311320        self.copy_from(stream, format='csv', sep=';')
    312321        self.check_table()
     322        self.check_rowcount()
    313323
    314324    def test_binary(self):
    315325        self.assertRaises(IOError, self.copy_from,
    316326            b'NOPGCOPY\n', format='binary')
     327        self.check_rowcount(-1)
    317328
    318329    def test_binary_with_sep(self):
     
    333344        self.assertEqual(len(stream), 0)
    334345        self.assertEqual(stream.sizes, [8192])
     346        self.check_rowcount()
    335347
    336348    def test_size_positive(self):
     
    342354        self.assertEqual(len(stream), 0)
    343355        self.assertEqual(stream.sizes, [size] * num_chunks)
     356        self.check_rowcount()
    344357
    345358    def test_size_negative(self):
     
    349362        self.assertEqual(len(stream), 0)
    350363        self.assertEqual(stream.sizes, [None])
     364        self.check_rowcount()
    351365
    352366
     
    397411        self.assertIsInstance(rows, str)
    398412        self.assertEqual(rows, self.data_text)
     413        self.check_rowcount()
    399414
    400415    if str is unicode:
     
    420435            self.assertEqual(rows, self.data_text.decode('utf-8'))
    421436
     437    def test_rowcount_increment(self):
     438        ret = self.copy_to()
     439        self.assertIsInstance(ret, Iterable)
     440        for n, row in enumerate(ret):
     441            self.check_rowcount(n + 1)
     442
    422443    def test_decode(self):
    423444        ret_raw = b''.join(self.copy_to(decode=False))
     
    426447        self.assertIsInstance(ret_decoded, unicode)
    427448        self.assertEqual(ret_decoded, ret_raw.decode('utf-8'))
     449        self.check_rowcount()
    428450
    429451    def test_sep(self):
     
    470492        self.assertIsInstance(rows, str)
    471493        self.assertEqual(rows, self.data_csv)
     494        self.check_rowcount(3)
    472495
    473496    def test_csv_with_sep(self):
     
    481504            self.assertTrue(row.startswith(b'PGCOPY\n\377\r\n\0'))
    482505            break
     506        self.check_rowcount(1)
    483507
    484508    def test_binary_with_sep(self):
     
    497521        self.assertIsInstance(rows[0], str)
    498522        self.assertEqual(rows[0], '%s!\n' % self.data[1][1])
     523        self.check_rowcount(1)
    499524
    500525    def test_file(self):
     
    508533        sizes = [len(row) + 1 for row in data.splitlines()]
    509534        self.assertEqual(stream.sizes, sizes)
     535        self.check_rowcount()
    510536
    511537
     
    517543        self.cursor.copy_from(self.data_text, 'copytest', format='text')
    518544        self.check_table()
     545        self.check_rowcount()
    519546        # get data back in binary format
    520547        ret = self.cursor.copy_to(None, 'copytest', format='binary')
     
    522549        data_binary = b''.join(ret)
    523550        self.assertTrue(data_binary.startswith(b'PGCOPY\n\377\r\n\0'))
     551        self.check_rowcount()
    524552        self.truncate_table()
    525553        # fill table from binary data
    526554        self.cursor.copy_from(data_binary, 'copytest', format='binary')
    527555        self.check_table()
     556        self.check_rowcount()
    528557
    529558
Note: See TracChangeset for help on using the changeset viewer.