Changeset 763 for trunk/pg.py


Ignore:
Timestamp:
Jan 17, 2016, 4:01:04 PM (4 years ago)
Author:
cito
Message:

Achieve 100% test coverage for pg module on the trunk

Note that some lines are only covered in certain Pg or Py versions,
so you need to run tests with different versions to be sure.

Also added another synonym for transaction methods,
you can now pick your favorite for all three of them.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • trunk/pg.py

    r762 r763  
    374374        """Prepare and add a parameter to the list."""
    375375        if value is not None and typ != 'text':
     376            prepare = self._prepare_funcs[typ]
    376377            try:
    377                 prepare = self._prepare_funcs[typ]
    378             except KeyError:
    379                 pass
    380             else:
    381                 try:
    382                     value = prepare(self, value)
    383                 except ValueError:
    384                     return value
     378                value = prepare(self, value)
     379            except ValueError:
     380                return value
    385381        params.append(value)
    386382        return '$%d' % len(params)
     
    469465            qstr += ' TO ' + name
    470466        return self.query(qstr)
     467
     468    abort = rollback
    471469
    472470    def savepoint(self, name):
     
    724722                    self._prepare_qualified_param(table, 1))
    725723            names = self.db.query(q, (table,)).getresult()
    726             if not names:
    727                 raise KeyError('Table %s does not exist' % table)
    728724            if not self._regtypes:
    729725                names = ((name, _simpletype(typ)) for name, typ in names)
     
    788784            if isinstance(row, dict):
    789785                if qoid not in row:
    790                     raise _db_error('%s not in row' % qoid)
     786                    raise _prg_error('%s not in row' % qoid)
    791787            else:
    792788                row = {qoid: row}
     
    855851        self._do_debug(q, params)
    856852        q = self.db.query(q, params)
    857         res = q.dictresult()
    858         if not res:
    859             raise _int_error('Insert operation did not return new values')
     853        res = q.dictresult()  # this will always return a row
    860854        for n, value in res[0].items():
    861855            if n == 'oid':
     
    880874        qoid = _oid_key(table)
    881875        if 'oid' in kw:
    882             kw[qoid] = kw['oid']
    883             del kw['oid']
     876            kw[qoid] = kw.pop('oid')
    884877        if row is None:
    885878            row = {}
     
    991984        keyname = [keyname] if isinstance(
    992985            keyname, basestring) else sorted(keyname)
    993         try:
    994             target = ', '.join(col(k) for k in keyname)
    995         except KeyError:
    996             raise _prg_error('Upsert operation needs primary key or oid')
     986        target = ', '.join(col(k) for k in keyname)
    997987        update = []
    998988        keyname = set(keyname)
     
    1005995                        value = 'excluded.%s' % col(n)
    1006996                    update.append('%s = %s' % (col(n), value))
    1007         if not values and not update:
     997        if not values:
    1008998            return row
    1009999        do = 'update set %s' % ', '.join(update) if update else 'nothing'
     
    10221012            raise  # re-raise original error
    10231013        res = q.dictresult()
    1024         if res:  # may be empty with "do nothing"
     1014        if update:  # may be empty with "do nothing"
    10251015            for n, value in res[0].items():
    10261016                if n == 'oid':
    10271017                    n = _oid_key(table)
    1028                 elif attnames.get(n) == 'bytea':
     1018                elif attnames.get(n) == 'bytea' and value is not None:
    10291019                    value = self.unescape_bytea(value)
    10301020                row[n] = value
    1031         elif update:
    1032             raise _int_error('Upsert operation did not return new values')
    10331021        else:
    10341022            self.get(table, row)
     
    10741062        qoid = _oid_key(table)
    10751063        if 'oid' in kw:
    1076             kw[qoid] = kw['oid']
    1077             del kw['oid']
     1064            kw[qoid] = kw.pop('oid')
    10781065        if row is None:
    10791066            row = {}
Note: See TracChangeset for help on using the changeset viewer.