assertsql.py 文件源码

def process_cursor_execute(self, statement, parameters, context,
                               executemany):
        if not context:
            return
        from sqlalchemy.schema import _DDLCompiles
        _received_parameters = list(context.compiled_parameters)

        # recompile from the context, using the default dialect

        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=DefaultDialect())
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=DefaultDialect(),
                    column_keys=context.compiled.column_keys)
            )
        _received_statement = re.sub(r'[\n\t]', '', str(compiled))
        equivalent = self.statement == _received_statement
        if self.params:
            if util.callable(self.params):
                params = self.params(context)
            else:
                params = self.params
            if not isinstance(params, list):
                params = [params]
            else:
                params = list(params)
            all_params = list(params)
            all_received = list(_received_parameters)
            while params:
                param = dict(params.pop(0))
                for k, v in context.compiled.params.items():
                    param.setdefault(k, v)
                if param not in _received_parameters:
                    equivalent = False
                    break
                else:
                    _received_parameters.remove(param)
            if _received_parameters:
                equivalent = False
        else:
            params = {}
            all_params = {}
            all_received = []
        self._result = equivalent
        if not self._result:
            print('Testing for compiled statement %r partial params '
                  '%r, received %r with params %r' %
                  (self.statement, all_params,
                   _received_statement, all_received))
            self._errmsg = (
                'Testing for compiled statement %r partial params %r, '
                'received %r with params %r' %
                (self.statement, all_params,
                 _received_statement, all_received))

            # print self._errmsg