KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _set_signature method in autotest

Best Python code snippet using autotest_python

cursor.py

Source:cursor.py

...105    def _set_id(self, id):106        if self._id is not None and self._id != id:107            self._connection._client.close_statement(self._connection._id, self._id)108        self._id = id109    def _set_signature(self, signature):110        self._signature = signature111        self._column_data_types = []112        self._parameter_data_types = []113        if signature is None:114            return115        for column in signature.columns:116            dtype = TypeHelper.from_class(column.column_class_name)117            self._column_data_types.append(dtype)118        for parameter in signature.parameters:119            dtype = TypeHelper.from_class(parameter.class_name)120            self._parameter_data_types.append(dtype)121    def _set_frame(self, frame):122        self._frame = frame123        self._pos = None124        if frame is not None:125            if frame.rows:126                self._pos = 0127            elif not frame.done:128                raise InternalError('got an empty frame, but the statement is not done yet')129    def _fetch_next_frame(self):130        offset = self._frame.offset + len(self._frame.rows)131        frame = self._connection._client.fetch(132            self._connection._id, self._id,133            offset=offset, frame_max_size=self.itersize)134        self._set_frame(frame)135    def _process_results(self, results):136        if results:137            result = results[0]138            if result.own_statement:139                self._set_id(result.statement_id)140            self._set_signature(result.signature if result.HasField('signature') else None)141            self._set_frame(result.first_frame if result.HasField('first_frame') else None)142            self._updatecount = result.update_count143    def _transform_parameters(self, parameters):144        typed_parameters = []145        for value, data_type in zip(parameters, self._parameter_data_types):146            field_name, rep, mutate_to, cast_from = data_type147            typed_value = common_pb2.TypedValue()148            if value is None:149                typed_value.null = True150                typed_value.type = common_pb2.NULL151            else:152                typed_value.null = False153                # use the mutator function154                if mutate_to is not None:155                    value = mutate_to(value)156                typed_value.type = rep157                setattr(typed_value, field_name, value)158            typed_parameters.append(typed_value)159        return typed_parameters160    def execute(self, operation, parameters=None):161        if self._closed:162            raise ProgrammingError('the cursor is already closed')163        self._updatecount = -1164        self._set_frame(None)165        if parameters is None:166            if self._id is None:167                self._set_id(self._connection._client.create_statement(self._connection._id))168            results = self._connection._client.prepare_and_execute(169                self._connection._id, self._id,170                operation, first_frame_max_size=self.itersize)171            self._process_results(results)172        else:173            statement = self._connection._client.prepare(174                self._connection._id, operation)175            self._set_id(statement.id)176            self._set_signature(statement.signature)177            results = self._connection._client.execute(178                self._connection._id, self._id,179                statement.signature, self._transform_parameters(parameters),180                first_frame_max_size=self.itersize)181            self._process_results(results)182    def executemany(self, operation, seq_of_parameters):183        if self._closed:184            raise ProgrammingError('the cursor is already closed')185        self._updatecount = -1186        self._set_frame(None)187        statement = self._connection._client.prepare(188            self._connection._id, operation, max_rows_total=0)189        self._set_id(statement.id)190        self._set_signature(statement.signature)191        for parameters in seq_of_parameters:192            self._connection._client.execute(193                self._connection._id, self._id,194                statement.signature, self._transform_parameters(parameters),195                first_frame_max_size=0)196    def _transform_row(self, row):197        """Transforms a Row into Python values.198        :param row:199            A ``common_pb2.Row`` object.200        :returns:201            A list of values casted into the correct Python types.202        :raises:203            NotImplementedError204        """...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.