KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _process_command method in lisa
Best Python code snippet using lisa_python
test_engine.py
Source:test_engine.py 
...56 )57 self.localization_patcher.start()58 def tearDown(self) -> None:59 self.localization_patcher.stop()60 def _process_command(self, command) -> DialogEventBatch:61 persist_dialog_call_count = self.repo.persist_dialog_state.call_count62 process_command(command, str(self.next_seq), repo=self.repo)63 self.next_seq += 164 self.assertEqual(65 persist_dialog_call_count + 1, len(self.repo.persist_dialog_state.call_args_list)66 )67 return self.repo.persist_dialog_state.call_args[0][0]68 def _assert_event_types(self, batch: DialogEventBatch, *args: DialogEventType):69 self.assertEqual(len(args), len(batch.events), f"{args} vs {batch.events}")70 for i in range(len(batch.events)):71 self.assertEqual(args[i], batch.events[i].event_type)72 def _set_current_prompt(self, prompt_index: int, should_advance: bool):73 self.dialog_state.current_drill = self.drill74 underlying_prompt = self.drill.prompts[prompt_index]75 prompt = Mock(wraps=underlying_prompt)76 prompt.slug = underlying_prompt.slug77 prompt.messages = underlying_prompt.messages78 prompt.response_user_profile_key = underlying_prompt.response_user_profile_key79 prompt.max_failures = underlying_prompt.max_failures80 prompt.should_advance_with_answer.return_value = should_advance81 self.drill.prompts[prompt_index] = prompt82 self.dialog_state.current_prompt_state = PromptState(slug=prompt.slug, start_time=self.now)83 def test_skip_processed_sequence_numbers(self, get_drill_mock):84 command = Mock(wraps=ProcessSMSMessage(self.phone_number, "hey"))85 process_command(command, "0", repo=self.repo)86 self.assertFalse(command.execute.called)87 def test_advance_sequence_numbers(self, get_drill_mock):88 validator = MagicMock()89 validation_payload = CodeValidationPayload(valid=True, account_info={"company": "WeWork"})90 validator.validate_code = MagicMock(return_value=validation_payload)91 command = ProcessSMSMessage(self.phone_number, "hey", registration_validator=validator)92 batch = self._process_command(command)93 self.assertEqual(1, len(batch.events))94 self.assertEqual("1", self.dialog_state.seq)95 def test_first_message_validates_user(self, get_drill_mock):96 validator = MagicMock()97 validation_payload = CodeValidationPayload(valid=True, account_info={"company": "WeWork"})98 validator.validate_code = MagicMock(return_value=validation_payload)99 command = ProcessSMSMessage(self.phone_number, "hey", registration_validator=validator)100 self.assertFalse(self.dialog_state.user_profile.validated)101 batch = self._process_command(command)102 self._assert_event_types(batch, DialogEventType.USER_VALIDATED)103 self.assertEqual(104 validation_payload, batch.events[0].code_validation_payload # type: ignore105 )106 def test_revalidate_demo_user(self, get_drill_mock):107 validator = MagicMock()108 validation_payload = CodeValidationPayload(valid=True, is_demo=True)109 validator.validate_code = MagicMock(return_value=validation_payload)110 self.assertFalse(self.dialog_state.user_profile.validated)111 command = ProcessSMSMessage(self.phone_number, "hey", registration_validator=validator)112 batch = self._process_command(command)113 self._assert_event_types(batch, DialogEventType.USER_VALIDATED)114 command = StartDrill(self.phone_number, self.drill.slug)115 self._process_command(command)116 validation_payload = CodeValidationPayload(valid=True, account_info={"company": "WeWork"})117 validator.validate_code = MagicMock(return_value=validation_payload)118 command = ProcessSMSMessage(self.phone_number, "hey", registration_validator=validator)119 batch = self._process_command(command)120 self._assert_event_types(batch, DialogEventType.USER_VALIDATED)121 def test_advance_demo_user(self, get_drill_mock):122 validator = MagicMock()123 validation_payload = CodeValidationPayload(valid=True, is_demo=True)124 validator.validate_code = MagicMock(return_value=validation_payload)125 self.assertFalse(self.dialog_state.user_profile.validated)126 command = ProcessSMSMessage(self.phone_number, "hey", registration_validator=validator)127 batch = self._process_command(command)128 self._assert_event_types(batch, DialogEventType.USER_VALIDATED)129 command = StartDrill(self.phone_number, self.drill.slug)130 self._process_command(command)131 # the user's next message isn't a validation code - so we just keep going132 validation_payload = CodeValidationPayload(valid=False)133 validator.validate_code = MagicMock(return_value=validation_payload)134 command = ProcessSMSMessage(self.phone_number, "hey", registration_validator=validator)135 batch = self._process_command(command)136 self._assert_event_types(137 batch, DialogEventType.COMPLETED_PROMPT, DialogEventType.ADVANCED_TO_NEXT_PROMPT138 )139 def test_first_message_does_not_validate_user(self, get_drill_mock):140 validator = MagicMock()141 validation_payload = CodeValidationPayload(valid=False)142 validator.validate_code = MagicMock(return_value=validation_payload)143 command = ProcessSMSMessage(self.phone_number, "hey", registration_validator=validator)144 self.assertFalse(self.dialog_state.user_profile.validated)145 batch = self._process_command(command)146 self._assert_event_types(batch, DialogEventType.USER_VALIDATION_FAILED)147 def test_start_drill(self, get_drill_mock):148 self.dialog_state.user_profile.validated = True149 command = StartDrill(self.phone_number, self.drill.slug)150 batch = self._process_command(command)151 self._assert_event_types(batch, DialogEventType.DRILL_STARTED)152 event: DrillStarted = batch.events[0] # type: ignore153 self.assertEqual(self.drill, event.drill)154 self.assertEqual(self.drill.first_prompt(), event.first_prompt)155 self.assertIsNotNone(event.drill_instance_id)156 def test_start_drill_not_validated(self, get_drill_mock):157 self.dialog_state.user_profile.validated = False158 self.dialog_state.user_profile.opted_out = False159 command = StartDrill(self.phone_number, self.drill.slug)160 batch = self._process_command(command)161 self.assertEqual(0, len(batch.events))162 def test_start_drill_opted_out(self, get_drill_mock):163 self.dialog_state.user_profile.validated = True164 self.dialog_state.user_profile.opted_out = True165 command = StartDrill(self.phone_number, self.drill.slug)166 batch = self._process_command(command)167 self.assertEqual(0, len(batch.events))168 def test_complete_and_advance(self, get_drill_mock):169 self.dialog_state.user_profile.validated = True170 self._set_current_prompt(0, should_advance=True)171 command = ProcessSMSMessage(self.phone_number, "go")172 batch = self._process_command(command)173 self._assert_event_types(174 batch, DialogEventType.COMPLETED_PROMPT, DialogEventType.ADVANCED_TO_NEXT_PROMPT175 )176 completed_event: CompletedPrompt = batch.events[0] # type: ignore177 self.assertEqual(completed_event.prompt, self.drill.prompts[0])178 self.assertEqual(completed_event.response, "go")179 self.assertEqual(completed_event.drill_instance_id, self.dialog_state.drill_instance_id)180 advanced_event: AdvancedToNextPrompt = batch.events[1] # type: ignore181 self.assertEqual(self.drill.prompts[1], advanced_event.prompt)182 self.assertEqual(self.dialog_state.drill_instance_id, advanced_event.drill_instance_id)183 def test_repeat_with_wrong_answer(self, get_drill_mock):184 self.dialog_state.user_profile.validated = True185 self._set_current_prompt(2, should_advance=False)186 command = ProcessSMSMessage(self.phone_number, "completely wrong answer")187 batch = self._process_command(command)188 self._assert_event_types(batch, DialogEventType.FAILED_PROMPT)189 failed_event: FailedPrompt = batch.events[0] # type: ignore190 self.assertEqual(failed_event.prompt, self.drill.prompts[2])191 self.assertFalse(failed_event.abandoned)192 self.assertEqual(failed_event.response, "completely wrong answer")193 self.assertEqual(failed_event.drill_instance_id, self.dialog_state.drill_instance_id)194 def test_advance_with_too_many_wrong_answers(self, get_drill_mock):195 self.dialog_state.user_profile.validated = True196 self._set_current_prompt(2, should_advance=False)197 self.dialog_state.current_prompt_state.failures = 1198 command = ProcessSMSMessage(self.phone_number, "completely wrong answer")199 batch = self._process_command(command)200 self._assert_event_types(201 batch, DialogEventType.FAILED_PROMPT, DialogEventType.ADVANCED_TO_NEXT_PROMPT202 )203 failed_event: FailedPrompt = batch.events[0] # type: ignore204 self.assertEqual(failed_event.prompt, self.drill.prompts[2])205 self.assertTrue(failed_event.abandoned)206 self.assertEqual(failed_event.response, "completely wrong answer")207 self.assertEqual(failed_event.drill_instance_id, self.dialog_state.drill_instance_id)208 advanced_event: AdvancedToNextPrompt = batch.events[1] # type: ignore209 self.assertEqual(self.drill.prompts[3], advanced_event.prompt)210 self.assertEqual(self.dialog_state.drill_instance_id, advanced_event.drill_instance_id)211 def test_conclude_with_right_answer(self, get_drill_mock):212 self.dialog_state.user_profile.validated = True213 self._set_current_prompt(3, should_advance=True)214 command = ProcessSMSMessage(self.phone_number, "foo")215 batch = self._process_command(command)216 self._assert_event_types(217 batch,218 DialogEventType.COMPLETED_PROMPT,219 DialogEventType.ADVANCED_TO_NEXT_PROMPT,220 DialogEventType.DRILL_COMPLETED,221 )222 completed_event: CompletedPrompt = batch.events[0] # type: ignore223 self.assertEqual(completed_event.prompt, self.drill.prompts[3])224 self.assertEqual(completed_event.response, "foo")225 self.assertEqual(completed_event.drill_instance_id, self.dialog_state.drill_instance_id)226 advanced_event: AdvancedToNextPrompt = batch.events[1] # type: ignore227 self.assertEqual(self.drill.prompts[4], advanced_event.prompt)228 self.assertEqual(self.dialog_state.drill_instance_id, advanced_event.drill_instance_id)229 drill_completed_event: DrillCompleted = batch.events[2] # type: ignore230 self.assertEqual(231 self.dialog_state.drill_instance_id, drill_completed_event.drill_instance_id232 )233 def test_conclude_with_too_many_wrong_answers(self, get_drill_mock):234 self.dialog_state.user_profile.validated = True235 self._set_current_prompt(3, should_advance=False)236 self.dialog_state.current_prompt_state.failures = 1237 command = ProcessSMSMessage(self.phone_number, "completely wrong answer")238 batch = self._process_command(command)239 self._assert_event_types(240 batch,241 DialogEventType.FAILED_PROMPT,242 DialogEventType.ADVANCED_TO_NEXT_PROMPT,243 DialogEventType.DRILL_COMPLETED,244 )245 failed_event: FailedPrompt = batch.events[0] # type: ignore246 self.assertEqual(failed_event.prompt, self.drill.prompts[3])247 self.assertTrue(failed_event.abandoned)248 self.assertEqual(failed_event.response, "completely wrong answer")249 self.assertEqual(failed_event.drill_instance_id, self.dialog_state.drill_instance_id)250 advanced_event: AdvancedToNextPrompt = batch.events[1] # type: ignore251 self.assertEqual(self.drill.prompts[4], advanced_event.prompt)252 self.assertEqual(self.dialog_state.drill_instance_id, advanced_event.drill_instance_id)253 drill_completed_event: DrillCompleted = batch.events[2] # type: ignore254 self.assertEqual(255 self.dialog_state.drill_instance_id, drill_completed_event.drill_instance_id256 )257 def test_opt_out(self, get_drill_mock):258 self.dialog_state.user_profile.validated = True259 self._set_current_prompt(0, should_advance=True)260 command = ProcessSMSMessage(self.phone_number, "stop")261 batch = self._process_command(command)262 self._assert_event_types(batch, DialogEventType.OPTED_OUT)263 def test_message_during_opt_out(self, get_drill_mock):264 self.dialog_state.user_profile.validated = True265 self.dialog_state.user_profile.opted_out = True266 command = ProcessSMSMessage(self.phone_number, "it's not a bacteria")267 batch = self._process_command(command)268 self.assertEqual(0, len(batch.events))269 def test_opt_back_in(self, get_drill_mock):270 self.dialog_state.user_profile.validated = True271 self.dialog_state.user_profile.opted_out = True272 command = ProcessSMSMessage(self.phone_number, "start")273 batch = self._process_command(command)274 self._assert_event_types(batch, DialogEventType.NEXT_DRILL_REQUESTED)275 def test_ask_for_help(self, get_drill_mock):276 command = ProcessSMSMessage(self.phone_number, "help")277 batch = self._process_command(command)278 self.assertEqual(0, len(batch.events)) # response handled by twilio279 def test_ask_for_help_validated(self, get_drill_mock):280 self.dialog_state.user_profile.validated = True281 command = ProcessSMSMessage(self.phone_number, "help")282 batch = self._process_command(command)283 self.assertEqual(0, len(batch.events)) # response handled by twilio284 def test_ask_for_more(self, get_drill_mock):285 self.dialog_state.user_profile.validated = True286 command = ProcessSMSMessage(self.phone_number, "more")287 batch = self._process_command(command)288 self._assert_event_types(batch, DialogEventType.NEXT_DRILL_REQUESTED)289 def test_trigger_reminder(self, get_drill_mock):290 self.dialog_state.user_profile.validated = True291 self._set_current_prompt(2, should_advance=True)292 command = TriggerReminder(293 phone_number=self.phone_number,294 drill_instance_id=self.dialog_state.drill_instance_id, # type:ignore295 prompt_slug=self.drill.prompts[2].slug,296 )297 batch = self._process_command(command)298 self._assert_event_types(batch, DialogEventType.REMINDER_TRIGGERED)299 self.assertTrue(self.dialog_state.current_prompt_state.reminder_triggered)300 def test_trigger_reminder_not_validated(self, get_drill_mock):301 self.dialog_state.user_profile.validated = False302 self._set_current_prompt(2, should_advance=True)303 command = TriggerReminder(304 phone_number=self.phone_number,305 drill_instance_id=self.dialog_state.drill_instance_id, # type:ignore306 prompt_slug=self.drill.prompts[2].slug,307 )308 batch = self._process_command(command)309 self.assertEqual(0, len(batch.events))310 def test_trigger_reminder_opted_out(self, get_drill_mock):311 self.dialog_state.user_profile.validated = True312 self.dialog_state.user_profile.opted_out = True313 self._set_current_prompt(2, should_advance=True)314 command = TriggerReminder(315 phone_number=self.phone_number,316 drill_instance_id=self.dialog_state.drill_instance_id, # type:ignore317 prompt_slug=self.drill.prompts[2].slug,318 )319 batch = self._process_command(command)320 self.assertEqual(0, len(batch.events))321 def test_trigger_reminder_idempotence(self, get_drill_mock):322 self.dialog_state.user_profile.validated = True323 self._set_current_prompt(2, should_advance=True)324 self.dialog_state.current_prompt_state.reminder_triggered = True325 command = TriggerReminder(326 phone_number=self.phone_number,327 drill_instance_id=self.dialog_state.drill_instance_id, # type:ignore328 prompt_slug=self.drill.prompts[2].slug,329 )330 batch = self._process_command(command)331 self.assertEqual(0, len(batch.events))332 self.assertTrue(self.dialog_state.current_prompt_state.reminder_triggered)333 def test_trigger_late_reminder_later_prompt(self, get_drill_mock):334 self.dialog_state.user_profile.validated = True335 self._set_current_prompt(3, should_advance=True)336 command = TriggerReminder(337 phone_number=self.phone_number,338 drill_instance_id=self.dialog_state.drill_instance_id, # type:ignore339 prompt_slug=self.drill.prompts[2].slug,340 )341 batch = self._process_command(command)342 self.assertEqual(0, len(batch.events))343 self.assertFalse(self.dialog_state.current_prompt_state.reminder_triggered)344 def test_trigger_late_reminder_later_drill(self, get_drill_mock):345 self.dialog_state.user_profile.validated = True346 self._set_current_prompt(2, should_advance=True)347 command = TriggerReminder(self.phone_number, uuid.uuid4(), self.drill.prompts[2].slug)348 batch = self._process_command(command)349 self.assertEqual(0, len(batch.events))...
backend.py
Source:backend.py
...123 self.last_tx = ''124 self.last_eer = ''125 self.last_esr = ''126 # this function only should be used with commands that returns a response127 def _process_command(self, cmd):128 with self._lock:129 self._clear_lasts()130 log.info("Processing " + cmd)131 # If an error happens with socket it will raise an exception or if132 # it is not conn133 self.sock.execute_command(cmd)134 self.last_tx = cmd135 data = self.sock.read_response()136 self.last_rx = data137 self.check_if_error() # if there is an error it raises TTiCPXExc138 return data139 def _execute_command(self, cmd):140 with self._lock:141 self._clear_lasts()142 log.info("Executing " + cmd)143 # If an error happens with socket it will raise an exception or if144 # it is not conn145 self.sock.execute_command(cmd)146 self.last_tx = cmd147 self.check_if_error() # if there is an error it raises TTiCPXExc148 def _check_output(self, output):149 output = int(output) # can raise ValueError150 if output not in range(1, self.n_outputs + 1):151 raise TTiBackendExc(f"Only valid values for output are "152 f"{','.join([str(l) for l in range(1, self.n_outputs + 1)])}")153 return output154 def connect(self, ip, port=9221):155 if self.sock is None:156 self.sock = SockCommand(ip=ip, port=port, valid_commands=self._valid_commands)157 self.sock.connect(ip, port)158 def disconnect(self):159 self.sock.disconnect()160class CommonBackend(TTiBackend):161 """162 This Backend has been verified against a PL068 power supply.163 CommonBackend contains the commands that are common to CPx and PL power supplies. For each power supply series a new164 class that inherits from this one must be created.165 For commands that return values, parsing is implemented to return the value as a float or integer. I fear that not166 all power supply models will follow same templates on returning values. If you are using another model or series and167 the parsing fails, do not change it on this class. On its own class for the series, overwrite the methods that are168 wrongly parsed169 For commands that returns information instead of values, no parsing is done170 """171 def _get_status(self, output):172 cmd = "OP{}?".format(self._check_output(output))173 return self._process_command(cmd)174 def _set_mode(self, mode):175 m = int(mode)176 if m not in (0, 2):177 raise TTiBackendExc(178 "Only valid modes are 0 (independent) and 2 (tracking))")179 cmd = "CONFIG {}".format(mode)180 return self._execute_command(cmd)181 def _get_mode(self):182 return self._process_command("CONFIG?")183 # COMMANDS184 def set_mode_independent(self):185 self._set_mode(SlaveModes.independent)186 def set_mode_tracking(self):187 self._set_mode(SlaveModes.tracking)188 def is_independent_mode(self):189 return int(self._get_mode()) == SlaveModes.independent190 def is_tracking_mode(self):191 return int(self._get_mode()) == SlaveModes.tracking192 # Not exposed as it should only be read from check_error193 # def read_register_standard_event_status(self):194 # return self._process_command("*ESR?")195 def read_register_limit_event_status(self, output):196 cmd = "LSR{}?".format(self._check_output(output))197 return self._process_command(cmd)198 def read_register_execution_error(self):199 return self._process_command("EER?")200 def get_identifier(self):201 return self._process_command("*IDN?")202 def get_address(self):203 return self._process_command("ADDRESS?")204 def clear_status(self):205 self._execute_command("*CLS")206 def reset(self):207 self._execute_command("*RST")208 def clear_trip(self):209 self._execute_command("TRIPRST")210 def local(self):211 self._execute_command("LOCAL")212 def lock(self):213 return self._process_command("IFLOCK")214 def is_lock(self):215 return self._process_command("IFLOCK?")216 def unlock(self):217 return self._process_command("IFUNLOCK")218 def is_IST(self):219 return self._process_command("*IST?") == "1"220 def get_register_query_error(self):221 return self._process_command("QER?")222 def get_register_status_byte(self):223 return self._process_command("*STB?")224 def set_register_service_request(self, value):225 cmd = "*SRE {}".format(float(value))226 self._execute_command(cmd)227 def get_register_service_request(self):228 return self._process_command("*SRE?")229 def set_register_parallel_poll(self, value):230 cmd = "*PRE {}".format(value)231 self._execute_command(cmd)232 def get_register_parallel_poll(self):233 return self._process_command("*PRE?")234 def set_register_event_status(self, value):235 cmd = "*ESE {}".format(value)236 self._execute_command(cmd)237 def get_register_event_status(self):238 return self._process_command("*ESE?")239 def set_register_limit_event_status(self, output, value):240 cmd = "*LSE{} {}".format(output, value)241 self._execute_command(cmd)242 def get_register_limit_event_status(self, output):243 cmd = "*LSE{}?".format(output)244 return self._process_command(cmd)245 def save(self, output, store):246 cmd = "SAV{} {}".format(output, store)247 self._execute_command(cmd)248 def recall(self, output, store):249 cmd = "RCL{} {}".format(output, store)250 self._execute_command(cmd)251 def set_ratio(self, value):252 cmd = "RATIO {}".format(value)253 self._execute_command(cmd)254 def get_ratio(self):255 return self._process_command("RATIO?")256 def enable_output(self, output_1=False, output_2=False):257 if output_1:258 if output_2:259 cmd = "OPALL 1"260 else:261 cmd = "OP1 1"262 elif output_2:263 cmd = "OP2 1"264 else:265 return266 self._execute_command(cmd)267 def disable_output(self, output_1=False, output_2=False):268 if output_1:269 if output_2:270 cmd = "OPALL 0"271 else:272 cmd = "OP1 0"273 elif output_2:274 cmd = "OP2 0"275 else:276 return277 self._execute_command(cmd)278 def is_enabled(self, output):279 return int(self._get_status(self._check_output(output))) == 1280 def is_disabled(self, output):281 return int(self._get_status(self._check_output(output))) == 0282 def set_voltage(self, output, volts):283 cmd = "V{} {}".format(self._check_output(output), float(volts))284 self._execute_command(cmd)285 def set_voltage_verify(self, output, volts):286 cmd = "V{}V {}".format(self._check_output(output), float(volts))287 self._execute_command(cmd)288 # Returns the configured voltage289 def get_configured_voltage(self, output):290 """291 Return the output configured voltage value292 """293 cmd = "V{}?".format(self._check_output(output))294 data = self._process_command(cmd)295 if data:296 match = re.match(f"V{output} ([0-9,\.,\-,e]*)", data)297 if match:298 return float(match.groups()[0])299 raise TTiBackendExc("Command did not return a valid string. Received: {}".format(data))300 # Reads the voltage of an output301 def read_voltage(self, output):302 cmd = "V{}O?".format(self._check_output(output))303 return self._process_command(cmd)304 def get_OVP(self, output):305 """306 Over Voltage Protection307 """308 cmd = "OVP{}?".format(self._check_output(output))309 data = self._process_command(cmd)310 if data:311 return data.split()[1]312 raise TTiBackendExc("Command returned {}".format(data))313 def set_OVP(self, output, volts):314 cmd = "OVP{} {}".format(self._check_output(output), float(volts))315 self._execute_command(cmd)316 def set_delta_voltage(self, output, volts):317 cmd = "DELTAV{} {}".format(self._check_output(output), float(volts))318 self._execute_command(cmd)319 def get_delta_voltage(self, output):320 cmd = "DELTAV{}?".format(self._check_output(output))321 data = self._process_command(cmd)322 if data:323 return data.split()[1]324 raise TTiBackendExc("Command returned {}".format(data))325 def inc_voltage(self, output):326 """327 Increases voltage of output by a delta value, defined by set_delta_voltage328 :param output: which output of the power supply will be affected329 :return:330 """331 cmd = "INCV{}".format(self._check_output(output))332 self._execute_command(cmd)333 def inc_voltage_verify(self, output):334 cmd = "INCV{}V".format(self._check_output(output))335 self._execute_command(cmd)336 def dec_voltage(self, output):337 cmd = "DECV{}".format(self._check_output(output))338 self._execute_command(cmd)339 def dec_voltage_verify(self, output):340 cmd = "DECV{}V".format(self._check_output(output))341 self._execute_command(cmd)342 def set_current_limit(self, output, amps):343 cmd = "I{} {}".format(self._check_output(output), float(amps))344 self._execute_command(cmd)345 # Returns the configured current346 def get_current_limit(self, output):347 """348 Return the output configured voltage value349 """350 cmd = "I{}?".format(self._check_output(output))351 data = self._process_command(cmd)352 if data:353 match = re.match(fr"I{output} ([0-9,\.,\-,e]*)", data)354 if match:355 return float(match.groups()[0])356 raise TTiBackendExc("Command did not return a valid string. Received: {}".format(data))357 # Reads the current of an output358 def read_current(self, output):359 cmd = "I{}O?".format(self._check_output(output))360 return self._process_command(cmd)361 def set_OCP(self, output, amps):362 cmd = "OCP{} {}".format(self._check_output(output), float(amps))363 self._execute_command(cmd)364 def get_OCP(self, output):365 cmd = "OCP{}?".format(self._check_output(output))366 data = self._process_command(cmd)367 if data:368 return data.split()[1]369 raise TTiBackendExc("Command returned {}".format(data))370 def set_delta_current_limit(self, output, amps):371 cmd = "DELTAI{} {}".format(self._check_output(output), float(amps))372 self._execute_command(cmd)373 def get_delta_current(self, output):374 cmd = "DELTAI{}?".format(self._check_output(output))375 data = self._process_command(cmd)376 if data:377 return data.split()[1]378 raise TTiBackendExc("Command returned {}".format(data))379 def inc_current_limit(self, output):380 cmd = "INCI{}".format(self._check_output(output))381 self._execute_command(cmd)382 def dec_current(self, output):383 cmd = "DECI{}".format(self._check_output(output))384 self._execute_command(cmd)385class CPxBackend(CommonBackend):386 """387 There are no differences between common and CPx388 """389 def __init__(self, num_outputs=1):390 super().__init__(valid_commands=TTiCPxCommands(), num_outputs=num_outputs)391class IRangeValues:392 low = 1 # (500/800mA for PL series)393 high = 2394class PLBackend(CommonBackend):395 def __init__(self, num_outputs=1):396 super().__init__(valid_commands=TTiPLCommands(), num_outputs=num_outputs)397 def set_irange(self, output, value):398 if int(value) not in (1, 2):399 raise TTiBackendExc('irange can only be set to 1 or 2. 1 means low range (500/800 mA) - 2 means High range')400 cmd = f"IRANGE{self._check_output(output)} {value}"401 self._execute_command(cmd)402 def get_irange(self, output):403 cmd = f"IRANGE{self._check_output(output)}?"404 return int(self._process_command(cmd))405 def get_ipaddr(self):406 cmd = "IPADDR?"407 return self._process_command(cmd)408 def get_netmask(self):409 cmd = "NETMASK?"410 return self._process_command(cmd)411 def get_netconfig(self):412 cmd = "NETCONFIG?"413 return self._process_command(cmd)414 def get_OVP(self, output):415 """416 Over Voltage Protection417 We overwrite common function because PL068 model does not return VP<N> <NR2><RMT> as said on the specs,418 it simply returns the value in Volts419 """420 cmd = "OVP{}?".format(self._check_output(output))421 return float(self._process_command(cmd))422 def get_OCP(self, output):423 """424 Over Current Protection setting425 We overwrite common function because PL068 model does not return CP<N> <NR2><RMT> as said on the specs,426 it simply returns the value in Amperes427 """428 cmd = "OCP{}?".format(self._check_output(output))429 return float(self._process_command(cmd))430 def read_voltage(self, output):431 """432 Reads output voltage433 We overwrite common function to parse value and return it. We receive from Power Supply: <NR2>V<RMT>434 :param output: output to readback the voltage435 :return: voltage value, float436 """437 cmd = "V{}O?".format(self._check_output(output))438 data = self._process_command(cmd)439 if data:440 match = re.match(r"([0-9,\.,e,\-]*)V", data)441 if match:442 return float(match.groups()[0])443 raise TTiBackendExc(f"Data received not valid. Received: {data}")444 def read_current(self, output):445 """446 Reads output current447 We overwrite common function to parse value and return it. We receive from Power Supply: <NR2>A<RMT>448 :param output: output to readback the voltage449 :return: voltage value, float450 """451 cmd = "I{}O?".format(self._check_output(output))452 data = self._process_command(cmd)453 if data:454 match = re.match(r"([0-9,\.,e,\-]*)A", data)455 if match:456 return float(match.groups()[0])...
client.py
Source:client.py
...104 block_device_mapping = self._dict_to_string(block_device_mapping)105 poll = '--poll' if poll else ''106 _positional_args = ['name', 'poll']107 _response_type = responses.ServerResponse108 return self._process_command()109 def show(self, server_id):110 """Returns server details via nova show command"""111 _cmd = 'show'112 _response_type = responses.ServerResponse113 return self._process_command()114 def list(self):115 """Returns list of servers via nova list command"""116 _cmd = 'list'117 _response_type = responses.ServerListResponse118 return self._process_command()119 def delete(self, server_id):120 """Deletes server via nova delete command"""121 _cmd = 'delete'122 return self._process_command()123 def image_list(self):124 """Returns list of images via nova image-list command"""125 _cmd = 'image-list'126 _response_type = responses.ImageListResponse127 return self._process_command()128 def image_show(self, image_id):129 """Returns image details via nova image-show command"""130 _cmd = 'image-show'131 _response_type = responses.ImageShowResponse132 return self._process_command()133 def image_create(self, server_id, image_name, show=None, poll=None):134 """Creates an image via the nova image-create command"""135 _cmd = 'image-create'136 _response_type = responses.ImageShowResponse137 show = '--show' if show else ''138 poll = '--poll' if poll else ''139 _positional_args = ['show', 'poll', 'server_id', 'image_name']140 return self._process_command()141 def image_delete(self, image_id):142 """Deletes an image via the nova image-delete command"""143 _cmd = 'image-delete'144 return self._process_command()145 def flavor_list(self):146 """Returns list of flavors via nova flavor-list command"""147 _cmd = 'flavor-list'148 _response_type = responses.FlavorListResponse149 return self._process_command()150 def flavor_show(self, flavor_id):151 """Returns flavor details via nova flavor-show command"""152 _cmd = 'flavor-show'153 _response_type = responses.FlavorShowResponse154 return self._process_command()155 def volume_attach(self, server_id, volume_id, device=None):156 """Attaches specified volume to specified server as specified157 device via nova volume-attach command. Returns attachment info158 """159 _cmd = 'volume-attach'160 _response_type = responses.VolumeAttachResponse161 return self._process_command()162 def volume_create(163 self, size, snapshot_id=None, image_id=None, display_name=None,164 display_description=None, volume_type=None,165 availability_zone=None):166 """Creates a new volume with specified parameters via nova167 volume-create command"""168 _cmd = 'volume-create'169 _kwmap = {170 'snapshot_id': 'snapshot-id',171 'image_id': 'image-id',172 'display_name': 'display-name',173 'display_description': 'display-description',174 'volume_type': 'volume-type',175 'availability_zone': 'availability-zone'}176 _response_type = responses.VolumeCreateResponse177 return self._process_command()178 def volume_delete(self, volume_id):179 """Deletes specified volume via nova volume-delete command"""180 _cmd = 'volume-delete'181 return self._process_command()182 def volume_detach(self, server_id, volume_id):183 """Detaches specified volume from specified server via the nova184 volume-detach command"""185 _cmd = 'volume-detach'186 return self._process_command()187 def volume_list(self):188 """Returns list of all volumes via nova volume-list command"""189 _cmd = 'volume-list'190 _response_type = responses.VolumeListResponse191 return self._process_command()192 def volume_show(self, volume_id):193 """Returns details of specified volume via nova volume-show command"""194 _cmd = 'volume-show'195 _response_type = responses.VolumeCreateResponse196 return self._process_command()197 def volume_snapshot_create(198 self, volume_id, force=False, display_name=None,199 display_description=None):200 """Creates a snapshot of specified volume and returns the new volume201 details via the nova volume-snapshot-create command"""202 _cmd = 'volume-snapshot-create'203 force = True if force else False204 _response_type = responses.VolumeSnapshotCreateResponse205 return self._process_command()206 def volume_snapshot_delete(self, snapshot_id):207 """Deletes the specified volume snapshot via the nova208 volume-snapshot-delete command"""209 _cmd = 'volume-snapshot-delete'210 return self._process_command()211 def volume_snapshot_list(self):212 """Returns a list of all snapshots via nova volume-snapshot-list213 command"""214 _cmd = 'volume-snapshot-list'215 _response_type = responses.VolumeSnapshotListResponse216 return self._process_command()217 def volume_snapshot_show(self, snapshot_id):218 """Returns details of specified snapshot via219 nova volume-snapshot-show command"""220 _cmd = 'volume-snapshot-show'221 _response_type = responses.VolumeSnapshotShowResponse222 return self._process_command()223 def volume_type_create(self, name):224 """Create a new volume type via nova volume-type-create command"""225 _cmd = 'volume-type-create'226 _response_type = responses.VolumeTypeCreateResponse227 return self._process_command()228 def volume_type_delete(self, volume_type_id):229 """Delete a volume type via nova volume-type-delete command"""230 _cmd = 'volume-type-delete'231 return self._process_command()232 def volume_type_list(self):233 """Returns a list of all volume types via nova volume-type-list234 command"""235 _cmd = 'volume-type-list'236 _response_type = responses.VolumeTypeListResponse237 return self._process_command()238 def volume_update(self, server_id, attachment_id, volume_id):239 # TODO: Investigate this method. It's possibly a bug that this240 # updates the volume attachment. Currently seems broken.241 # Performs a PUT against242 # servers/{server-id}/os-volume_attachments/{volume-atachment-id}:243 # with this payload: '{"volumeAttachment": {"volumeId": "<a vol id>"}}'244 # Still not sure what this method does vs what it's supposed to do.245 _cmd = 'volume-update'...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
