KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _expect method in Playwright Python
Best Python code snippet using playwright-python
node.py
Source:node.py 
...105 print("%s" % cmd)106 self.pexpect = pexpect.spawn(cmd, timeout=4)107 # Add delay to ensure that the process is ready to receive commands.108 time.sleep(0.2)109 self._expect('spinel-cli >')110 self.debug(int(os.getenv('DEBUG', '0')))111 def _expect(self, pattern, timeout=-1, *args, **kwargs):112 """ Process simulator events until expected the pattern. """113 if timeout == -1:114 timeout = self.pexpect.timeout115 assert timeout > 0116 while timeout > 0:117 try:118 return self.pexpect.expect(pattern, 0.1, *args, **kwargs)119 except pexpect.TIMEOUT:120 timeout -= 0.1121 self.simulator.go(0)122 if timeout <= 0:123 raise124 def __init_soc(self, nodeid):125 """ Initialize a System-on-a-chip node connected via UART. """126 import fdpexpect127 serialPort = '/dev/ttyUSB%d' % ((nodeid - 1) * 2)128 self.pexpect = fdpexpect.fdspawn(129 os.open(serialPort, os.O_RDWR | os.O_NONBLOCK | os.O_NOCTTY)130 )131 def __del__(self):132 self.destroy()133 def destroy(self):134 if not self._initialized:135 return136 if (137 hasattr(self.pexpect, 'proc')138 and self.pexpect.proc.poll() is None139 or not hasattr(self.pexpect, 'proc')140 and self.pexpect.isalive()141 ):142 print("%d: exit" % self.nodeid)143 self.pexpect.send('exit\n')144 self.pexpect.expect(pexpect.EOF)145 self.pexpect.wait()146 self._initialized = False147 def read_cert_messages_in_commissioning_log(self, timeout=-1):148 """Get the log of the traffic after DTLS handshake.149 """150 format_str = br"=+?\[\[THCI\].*?type=%s.*?\].*?=+?[\s\S]+?-{40,}"151 join_fin_req = format_str % br"JOIN_FIN\.req"152 join_fin_rsp = format_str % br"JOIN_FIN\.rsp"153 dummy_format_str = br"\[THCI\].*?type=%s.*?"154 join_ent_ntf = dummy_format_str % br"JOIN_ENT\.ntf"155 join_ent_rsp = dummy_format_str % br"JOIN_ENT\.rsp"156 pattern = (157 b"("158 + join_fin_req159 + b")|("160 + join_fin_rsp161 + b")|("162 + join_ent_ntf163 + b")|("164 + join_ent_rsp165 + b")"166 )167 messages = []168 # There are at most 4 cert messages both for joiner and commissioner169 for _ in range(0, 4):170 try:171 self._expect(pattern, timeout=timeout)172 log = self.pexpect.match.group(0)173 messages.append(self._extract_cert_message(log))174 except BaseException:175 break176 return messages177 def _extract_cert_message(self, log):178 res = re.search(br"direction=\w+", log)179 assert res180 direction = res.group(0).split(b'=')[1].strip()181 res = re.search(br"type=\S+", log)182 assert res183 type = res.group(0).split(b'=')[1].strip()184 payload = bytearray([])185 payload_len = 0186 if type in [b"JOIN_FIN.req", b"JOIN_FIN.rsp"]:187 res = re.search(br"len=\d+", log)188 assert res189 payload_len = int(res.group(0).split(b'=')[1].strip())190 hex_pattern = br"\|(\s([0-9a-fA-F]{2}|\.\.))+?\s+?\|"191 while True:192 res = re.search(hex_pattern, log)193 if not res:194 break195 data = [196 int(hex, 16)197 for hex in res.group(0)[1:-1].split(b' ')198 if hex and hex != b'..'199 ]200 payload += bytearray(data)201 log = log[res.end() - 1:]202 assert len(payload) == payload_len203 return (direction, type, payload)204 def send_command(self, cmd, go=True):205 print("%d: %s" % (self.nodeid, cmd))206 self.pexpect.send(cmd + '\n')207 if go:208 self.simulator.go(0, nodeid=self.nodeid)209 sys.stdout.flush()210 def get_commands(self):211 self.send_command('?')212 self._expect('Commands:')213 commands = []214 while True:215 i = self._expect(['Done', r'(\S+)'])216 if i != 0:217 commands.append(self.pexpect.match.groups()[0])218 else:219 break220 return commands221 def set_mode(self, mode):222 cmd = 'mode %s' % mode223 self.send_command(cmd)224 self._expect('Done')225 def debug(self, level):226 # `debug` command will not trigger interaction with simulator227 self.send_command('debug %d' % level, go=False)228 def start(self):229 self.interface_up()230 self.thread_start()231 def stop(self):232 self.thread_stop()233 self.interface_down()234 def interface_up(self):235 self.send_command('ifconfig up')236 self._expect('Done')237 def interface_down(self):238 self.send_command('ifconfig down')239 self._expect('Done')240 def thread_start(self):241 self.send_command('thread start')242 self._expect('Done')243 def thread_stop(self):244 self.send_command('thread stop')245 self._expect('Done')246 def commissioner_start(self):247 cmd = 'commissioner start'248 self.send_command(cmd)249 self._expect('Done')250 def commissioner_add_joiner(self, addr, psk):251 cmd = 'commissioner joiner add %s %s' % (addr, psk)252 self.send_command(cmd)253 self._expect('Done')254 def joiner_start(self, pskd='', provisioning_url=''):255 cmd = 'joiner start %s %s' % (pskd, provisioning_url)256 self.send_command(cmd)257 self._expect('Done')258 def clear_whitelist(self):259 cmd = 'macfilter addr clear'260 self.send_command(cmd)261 self._expect('Done')262 def enable_whitelist(self):263 cmd = 'macfilter addr whitelist'264 self.send_command(cmd)265 self._expect('Done')266 def disable_whitelist(self):267 cmd = 'macfilter addr disable'268 self.send_command(cmd)269 self._expect('Done')270 def add_whitelist(self, addr, rssi=None):271 cmd = 'macfilter addr add %s' % addr272 if rssi is not None:273 cmd += ' %s' % rssi274 self.send_command(cmd)275 self._expect('Done')276 def remove_whitelist(self, addr):277 cmd = 'macfilter addr remove %s' % addr278 self.send_command(cmd)279 self._expect('Done')280 def get_addr16(self):281 self.send_command('rloc16')282 i = self._expect('([0-9a-fA-F]{4})')283 if i == 0:284 addr16 = int(self.pexpect.match.groups()[0], 16)285 self._expect('Done')286 return addr16287 def get_router_id(self):288 rloc16 = self.get_addr16()289 return rloc16 >> 10290 def get_addr64(self):291 self.send_command('extaddr')292 i = self._expect('([0-9a-fA-F]{16})')293 if i == 0:294 addr64 = self.pexpect.match.groups()[0].decode("utf-8")295 self._expect('Done')296 return addr64297 def get_eui64(self):298 self.send_command('eui64')299 i = self._expect('([0-9a-fA-F]{16})')300 if i == 0:301 addr64 = self.pexpect.match.groups()[0].decode("utf-8")302 self._expect('Done')303 return addr64304 def get_joiner_id(self):305 self.send_command('joiner id')306 i = self._expect('([0-9a-fA-F]{16})')307 if i == 0:308 addr = self.pexpect.match.groups()[0].decode("utf-8")309 self._expect('Done')310 return addr311 def get_channel(self):312 self.send_command('channel')313 i = self._expect(r'(\d+)\r?\n')314 if i == 0:315 channel = int(self.pexpect.match.groups()[0])316 self._expect('Done')317 return channel318 def set_channel(self, channel):319 cmd = 'channel %d' % channel320 self.send_command(cmd)321 self._expect('Done')322 def get_masterkey(self):323 self.send_command('masterkey')324 i = self._expect('([0-9a-fA-F]{32})')325 if i == 0:326 masterkey = self.pexpect.match.groups()[0].decode("utf-8")327 self._expect('Done')328 return masterkey329 def set_masterkey(self, masterkey):330 cmd = 'masterkey %s' % masterkey331 self.send_command(cmd)332 self._expect('Done')333 def get_key_sequence_counter(self):334 self.send_command('keysequence counter')335 i = self._expect(r'(\d+)\r?\n')336 if i == 0:337 key_sequence_counter = int(self.pexpect.match.groups()[0])338 self._expect('Done')339 return key_sequence_counter340 def set_key_sequence_counter(self, key_sequence_counter):341 cmd = 'keysequence counter %d' % key_sequence_counter342 self.send_command(cmd)343 self._expect('Done')344 def set_key_switch_guardtime(self, key_switch_guardtime):345 cmd = 'keysequence guardtime %d' % key_switch_guardtime346 self.send_command(cmd)347 self._expect('Done')348 def set_network_id_timeout(self, network_id_timeout):349 cmd = 'networkidtimeout %d' % network_id_timeout350 self.send_command(cmd)351 self._expect('Done')352 def get_network_name(self):353 self.send_command('networkname')354 while True:355 i = self._expect(['Done', r'(\S+)'])356 if i != 0:357 network_name = self.pexpect.match.groups()[0].decode('utf-8')358 else:359 break360 return network_name361 def set_network_name(self, network_name):362 cmd = 'networkname %s' % network_name363 self.send_command(cmd)364 self._expect('Done')365 def get_panid(self):366 self.send_command('panid')367 i = self._expect('([0-9a-fA-F]{4})')368 if i == 0:369 panid = int(self.pexpect.match.groups()[0], 16)370 self._expect('Done')371 return panid372 def set_panid(self, panid=config.PANID):373 cmd = 'panid %d' % panid374 self.send_command(cmd)375 self._expect('Done')376 def get_partition_id(self):377 self.send_command('leaderpartitionid')378 i = self._expect(r'(\d+)\r?\n')379 if i == 0:380 weight = self.pexpect.match.groups()[0]381 self._expect('Done')382 return weight383 def set_partition_id(self, partition_id):384 cmd = 'leaderpartitionid %d' % partition_id385 self.send_command(cmd)386 self._expect('Done')387 def set_router_upgrade_threshold(self, threshold):388 cmd = 'routerupgradethreshold %d' % threshold389 self.send_command(cmd)390 self._expect('Done')391 def set_router_downgrade_threshold(self, threshold):392 cmd = 'routerdowngradethreshold %d' % threshold393 self.send_command(cmd)394 self._expect('Done')395 def release_router_id(self, router_id):396 cmd = 'releaserouterid %d' % router_id397 self.send_command(cmd)398 self._expect('Done')399 def get_state(self):400 states = [r'\ndetached', r'\nchild', r'\nrouter', r'\nleader']401 self.send_command('state')402 match = self._expect(states)403 self._expect('Done')404 return states[match].strip(r'\n')405 def set_state(self, state):406 cmd = 'state %s' % state407 self.send_command(cmd)408 self._expect('Done')409 def get_timeout(self):410 self.send_command('childtimeout')411 i = self._expect(r'(\d+)\r?\n')412 if i == 0:413 timeout = self.pexpect.match.groups()[0]414 self._expect('Done')415 return timeout416 def set_timeout(self, timeout):417 cmd = 'childtimeout %d' % timeout418 self.send_command(cmd)419 self._expect('Done')420 def set_max_children(self, number):421 cmd = 'childmax %d' % number422 self.send_command(cmd)423 self._expect('Done')424 def get_weight(self):425 self.send_command('leaderweight')426 i = self._expect(r'(\d+)\r?\n')427 if i == 0:428 weight = self.pexpect.match.groups()[0]429 self._expect('Done')430 return weight431 def set_weight(self, weight):432 cmd = 'leaderweight %d' % weight433 self.send_command(cmd)434 self._expect('Done')435 def add_ipaddr(self, ipaddr):436 cmd = 'ipaddr add %s' % ipaddr437 self.send_command(cmd)438 self._expect('Done')439 def get_addrs(self):440 addrs = []441 self.send_command('ipaddr')442 while True:443 i = self._expect([r'(\S+(:\S*)+)\r?\n', 'Done'])444 if i == 0:445 addrs.append(self.pexpect.match.groups()[0].decode("utf-8"))446 elif i == 1:447 break448 return addrs449 def get_addr(self, prefix):450 network = ipaddress.ip_network(u'%s' % str(prefix))451 addrs = self.get_addrs()452 for addr in addrs:453 if isinstance(addr, bytearray):454 addr = bytes(addr)455 ipv6_address = ipaddress.ip_address(addr)456 if ipv6_address in network:457 return ipv6_address.exploded458 return None459 def get_addr_rloc(self):460 addrs = self.get_addrs()461 for addr in addrs:462 segs = addr.split(':')463 if (464 segs[4] == '0'465 and segs[5] == 'ff'466 and segs[6] == 'fe00'467 and segs[7] != 'fc00'468 ):469 return addr470 return None471 def get_addr_leader_aloc(self):472 addrs = self.get_addrs()473 for addr in addrs:474 segs = addr.split(':')475 if (476 segs[4] == '0'477 and segs[5] == 'ff'478 and segs[6] == 'fe00'479 and segs[7] == 'fc00'480 ):481 return addr482 return None483 def get_eidcaches(self):484 eidcaches = []485 self.send_command('eidcache')486 while True:487 i = self._expect([r'([a-fA-F0-9\:]+) ([a-fA-F0-9]+)\r?\n', 'Done'])488 if i == 0:489 eid = self.pexpect.match.groups()[0].decode("utf-8")490 rloc = self.pexpect.match.groups()[1].decode("utf-8")491 eidcaches.append((eid, rloc))492 elif i == 1:493 break494 return eidcaches495 def add_service(self, enterpriseNumber, serviceData, serverData):496 cmd = 'service add %s %s %s' % (497 enterpriseNumber,498 serviceData,499 serverData,500 )501 self.send_command(cmd)502 self._expect('Done')503 def remove_service(self, enterpriseNumber, serviceData):504 cmd = 'service remove %s %s' % (enterpriseNumber, serviceData)505 self.send_command(cmd)506 self._expect('Done')507 def __getLinkLocalAddress(self):508 for ip6Addr in self.get_addrs():509 if re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I):510 return ip6Addr511 return None512 def __getGlobalAddress(self):513 global_address = []514 for ip6Addr in self.get_addrs():515 if (516 (not re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I))517 and (518 not re.match(519 config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I520 )521 )522 and (523 not re.match(524 config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I525 )526 )527 ):528 global_address.append(ip6Addr)529 return global_address530 def __getRloc(self):531 for ip6Addr in self.get_addrs():532 if (533 re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I)534 and re.match(535 config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I536 )537 and not (538 re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I)539 )540 ):541 return ip6Addr542 return None543 def __getAloc(self):544 aloc = []545 for ip6Addr in self.get_addrs():546 if (547 re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I)548 and re.match(549 config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I550 )551 and re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I)552 ):553 aloc.append(ip6Addr)554 return aloc555 def __getMleid(self):556 for ip6Addr in self.get_addrs():557 if re.match(558 config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I559 ) and not (560 re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I)561 ):562 return ip6Addr563 return None564 def get_ip6_address(self, address_type):565 """Get specific type of IPv6 address configured on thread device.566 Args:567 address_type: the config.ADDRESS_TYPE type of IPv6 address.568 Returns:569 IPv6 address string.570 """571 if address_type == config.ADDRESS_TYPE.LINK_LOCAL:572 return self.__getLinkLocalAddress()573 elif address_type == config.ADDRESS_TYPE.GLOBAL:574 return self.__getGlobalAddress()575 elif address_type == config.ADDRESS_TYPE.RLOC:576 return self.__getRloc()577 elif address_type == config.ADDRESS_TYPE.ALOC:578 return self.__getAloc()579 elif address_type == config.ADDRESS_TYPE.ML_EID:580 return self.__getMleid()581 else:582 return None583 return None584 def get_context_reuse_delay(self):585 self.send_command('contextreusedelay')586 i = self._expect(r'(\d+)\r?\n')587 if i == 0:588 timeout = self.pexpect.match.groups()[0]589 self._expect('Done')590 return timeout591 def set_context_reuse_delay(self, delay):592 cmd = 'contextreusedelay %d' % delay593 self.send_command(cmd)594 self._expect('Done')595 def add_prefix(self, prefix, flags, prf='med'):596 cmd = 'prefix add %s %s %s' % (prefix, flags, prf)597 self.send_command(cmd)598 self._expect('Done')599 def remove_prefix(self, prefix):600 cmd = 'prefix remove %s' % prefix601 self.send_command(cmd)602 self._expect('Done')603 def add_route(self, prefix, prf='med'):604 cmd = 'route add %s %s' % (prefix, prf)605 self.send_command(cmd)606 self._expect('Done')607 def remove_route(self, prefix):608 cmd = 'route remove %s' % prefix609 self.send_command(cmd)610 self._expect('Done')611 def register_netdata(self):612 self.send_command('netdataregister')613 self._expect('Done')614 def energy_scan(self, mask, count, period, scan_duration, ipaddr):615 cmd = 'commissioner energy %d %d %d %d %s' % (616 mask,617 count,618 period,619 scan_duration,620 ipaddr,621 )622 self.send_command(cmd)623 if isinstance(self.simulator, simulator.VirtualTime):624 self.simulator.go(8)625 timeout = 1626 else:627 timeout = 8628 self._expect('Energy:', timeout=timeout)629 def panid_query(self, panid, mask, ipaddr):630 cmd = 'commissioner panid %d %d %s' % (panid, mask, ipaddr)631 self.send_command(cmd)632 if isinstance(self.simulator, simulator.VirtualTime):633 self.simulator.go(8)634 timeout = 1635 else:636 timeout = 8637 self._expect('Conflict:', timeout=timeout)638 def scan(self):639 self.send_command('scan')640 results = []641 while True:642 i = self._expect(643 [644 r'\|\s(\S+)\s+\|\s(\S+)\s+\|\s([0-9a-fA-F]{4})\s\|\s([0-9a-fA-F]{16})\s\|\s(\d+)\r?\n',645 'Done',646 ]647 )648 if i == 0:649 results.append(self.pexpect.match.groups())650 else:651 break652 return results653 def ping(self, ipaddr, num_responses=1, size=None, timeout=5):654 cmd = 'ping %s' % ipaddr655 if size is not None:656 cmd += ' %d' % size657 self.send_command(cmd)658 if isinstance(self.simulator, simulator.VirtualTime):659 self.simulator.go(timeout)660 result = True661 try:662 responders = {}663 while len(responders) < num_responses:664 i = self._expect([r'from (\S+):'])665 if i == 0:666 responders[self.pexpect.match.groups()[0]] = 1667 self._expect('\n')668 except (pexpect.TIMEOUT, socket.timeout):669 result = False670 if isinstance(self.simulator, simulator.VirtualTime):671 self.simulator.sync_devices()672 return result673 def reset(self):674 self.send_command('reset')675 time.sleep(0.1)676 def set_router_selection_jitter(self, jitter):677 cmd = 'routerselectionjitter %d' % jitter678 self.send_command(cmd)679 self._expect('Done')680 def set_active_dataset(681 self,682 timestamp,683 panid=None,684 channel=None,685 channel_mask=None,686 master_key=None,687 ):688 self.send_command('dataset clear')689 self._expect('Done')690 cmd = 'dataset activetimestamp %d' % timestamp691 self.send_command(cmd)692 self._expect('Done')693 if panid is not None:694 cmd = 'dataset panid %d' % panid695 self.send_command(cmd)696 self._expect('Done')697 if channel is not None:698 cmd = 'dataset channel %d' % channel699 self.send_command(cmd)700 self._expect('Done')701 if channel_mask is not None:702 cmd = 'dataset channelmask %d' % channel_mask703 self.send_command(cmd)704 self._expect('Done')705 if master_key is not None:706 cmd = 'dataset masterkey %s' % master_key707 self.send_command(cmd)708 self._expect('Done')709 self.send_command('dataset commit active')710 self._expect('Done')711 def set_pending_dataset(712 self, pendingtimestamp, activetimestamp, panid=None, channel=None713 ):714 self.send_command('dataset clear')715 self._expect('Done')716 cmd = 'dataset pendingtimestamp %d' % pendingtimestamp717 self.send_command(cmd)718 self._expect('Done')719 cmd = 'dataset activetimestamp %d' % activetimestamp720 self.send_command(cmd)721 self._expect('Done')722 if panid is not None:723 cmd = 'dataset panid %d' % panid724 self.send_command(cmd)725 self._expect('Done')726 if channel is not None:727 cmd = 'dataset channel %d' % channel728 self.send_command(cmd)729 self._expect('Done')730 self.send_command('dataset commit pending')731 self._expect('Done')732 def announce_begin(self, mask, count, period, ipaddr):733 cmd = 'commissioner announce %d %d %d %s' % (734 mask,735 count,736 period,737 ipaddr,738 )739 self.send_command(cmd)740 self._expect('Done')741 def send_mgmt_active_set(742 self,743 active_timestamp=None,744 channel=None,745 channel_mask=None,746 extended_panid=None,747 panid=None,748 master_key=None,749 mesh_local=None,750 network_name=None,751 binary=None,752 ):753 cmd = 'dataset mgmtsetcommand active '754 if active_timestamp is not None:755 cmd += 'activetimestamp %d ' % active_timestamp756 if channel is not None:757 cmd += 'channel %d ' % channel758 if channel_mask is not None:759 cmd += 'channelmask %d ' % channel_mask760 if extended_panid is not None:761 cmd += 'extpanid %s ' % extended_panid762 if panid is not None:763 cmd += 'panid %d ' % panid764 if master_key is not None:765 cmd += 'masterkey %s ' % master_key766 if mesh_local is not None:767 cmd += 'localprefix %s ' % mesh_local768 if network_name is not None:769 cmd += 'networkname %s ' % network_name770 if binary is not None:771 cmd += 'binary %s ' % binary772 self.send_command(cmd)773 self._expect('Done')774 def send_mgmt_pending_set(775 self,776 pending_timestamp=None,777 active_timestamp=None,778 delay_timer=None,779 channel=None,780 panid=None,781 master_key=None,782 mesh_local=None,783 network_name=None,784 ):785 cmd = 'dataset mgmtsetcommand pending '786 if pending_timestamp is not None:787 cmd += 'pendingtimestamp %d ' % pending_timestamp788 if active_timestamp is not None:789 cmd += 'activetimestamp %d ' % active_timestamp790 if delay_timer is not None:791 cmd += 'delaytimer %d ' % delay_timer792 if channel is not None:793 cmd += 'channel %d ' % channel794 if panid is not None:795 cmd += 'panid %d ' % panid796 if master_key is not None:797 cmd += 'masterkey %s ' % master_key798 if mesh_local is not None:799 cmd += 'localprefix %s ' % mesh_local800 if network_name is not None:801 cmd += 'networkname %s ' % network_name802 self.send_command(cmd)803 self._expect('Done')804 def coaps_start_psk(self, psk, pskIdentity):805 cmd = 'coaps psk %s %s' % (psk, pskIdentity)806 self.send_command(cmd)807 self._expect('Done')808 cmd = 'coaps start'809 self.send_command(cmd)810 self._expect('Done')811 def coaps_start_x509(self):812 cmd = 'coaps x509'813 self.send_command(cmd)814 self._expect('Done')815 cmd = 'coaps start'816 self.send_command(cmd)817 self._expect('Done')818 def coaps_set_resource_path(self, path):819 cmd = 'coaps resource %s' % path820 self.send_command(cmd)821 self._expect('Done')822 def coaps_stop(self):823 cmd = 'coaps stop'824 self.send_command(cmd)825 if isinstance(self.simulator, simulator.VirtualTime):826 self.simulator.go(5)827 timeout = 1828 else:829 timeout = 5830 self._expect('Done', timeout=timeout)831 def coaps_connect(self, ipaddr):832 cmd = 'coaps connect %s' % ipaddr833 self.send_command(cmd)834 if isinstance(self.simulator, simulator.VirtualTime):835 self.simulator.go(5)836 timeout = 1837 else:838 timeout = 5839 self._expect('coaps connected', timeout=timeout)840 def coaps_disconnect(self):841 cmd = 'coaps disconnect'842 self.send_command(cmd)843 self._expect('Done')844 self.simulator.go(5)845 def coaps_get(self):846 cmd = 'coaps get test'847 self.send_command(cmd)848 if isinstance(self.simulator, simulator.VirtualTime):849 self.simulator.go(5)850 timeout = 1851 else:852 timeout = 5853 self._expect('coaps response', timeout=timeout)854 def commissioner_mgmtset(self, tlvs_binary):855 cmd = 'commissioner mgmtset binary %s' % tlvs_binary856 self.send_command(cmd)857 self._expect('Done')858 def bytes_to_hex_str(self, src):859 return ''.join(format(x, '02x') for x in src)860 def commissioner_mgmtset_with_tlvs(self, tlvs):861 payload = bytearray()862 for tlv in tlvs:863 payload += tlv.to_hex()864 self.commissioner_mgmtset(self.bytes_to_hex_str(payload))865 def udp_start(self, local_ipaddr, local_port):866 cmd = 'udp open'867 self.send_command(cmd)868 self._expect('Done')869 cmd = 'udp bind %s %s' % (local_ipaddr, local_port)870 self.send_command(cmd)871 self._expect('Done')872 def udp_stop(self):873 cmd = 'udp close'874 self.send_command(cmd)875 self._expect('Done')876 def udp_send(self, bytes, ipaddr, port, success=True):877 cmd = 'udp send %s %d -s %d ' % (ipaddr, port, bytes)878 self.send_command(cmd)879 if success:880 self._expect('Done')881 else:882 self._expect('Error')883 def udp_check_rx(self, bytes_should_rx):884 self._expect('%d bytes' % bytes_should_rx)885 def router_list(self):886 cmd = 'router list'887 self.send_command(cmd)888 self._expect([r'(\d+)((\s\d+)*)'])889 g = self.pexpect.match.groups()890 router_list = g[0] + ' ' + g[1]891 router_list = [int(x) for x in router_list.split()]892 self._expect('Done')893 return router_list894 def router_table(self):895 cmd = 'router table'896 self.send_command(cmd)897 self._expect(r'(.*)Done')898 g = self.pexpect.match.groups()899 output = g[0]900 lines = output.strip().split('\n')901 lines = [l.strip() for l in lines]902 router_table = {}903 for i, line in enumerate(lines):904 if not line.startswith('|') or not line.endswith('|'):905 if i not in (0, 2):906 # should not happen907 print("unexpected line %d: %s" % (i, line))908 continue909 line = line[1:][:-1]910 line = [x.strip() for x in line.split('|')]911 if len(line) != 8:...
shtoken.py
Source:shtoken.py
1import unittest2from web3 import Web33from time import sleep4import settings as sts5import config6class SHToken(unittest.TestCase):7 @classmethod8 def setUpClass(cls):9 # åå§åæ°æ® sht-class, w3, sht10 while True:11 cls.w3 = Web3(Web3.HTTPProvider(sts.SIBBAY_SHT_NODE_HTTP, {"timeout": 30}))12 if cls.w3.isConnected() == False:13 print("node is not connected, wait " + str(5) + " second")14 sleep(5)15 else:16 print("connect to node by http: " + sts.SIBBAY_SHT_NODE_HTTP)17 break18 cls.sht = cls.w3.eth.contract(address=Web3.toChecksumAddress(sts.SIBBAY_SHT_ADDRESS), abi=sts.SIBBAY_SHT_ABI)19# @classmethod20# def tearDownClass(cls):21# cls.w3.providers[0]._socket.sock.close()22 # çå¾
确认23 def wait_tx_confirm(self, tx_hash):24 while True:25 sleep(config.waitting_time)26 logs = self.w3.eth.getTransactionReceipt(tx_hash)27 if logs is None:28 continue29 break30 return logs31 def create_account(self, _pwd):32 # å建账æ·33 ret = self.w3.personal.newAccount(_pwd)34 return ret35 # åéether36 def send_ether(self, _from, _to, _value, _pwd):37 _from = Web3.toChecksumAddress(_from)38 _to = Web3.toChecksumAddress(_to)39 # è®°å½è´¦æ·_toä½é¢40 balance_old = self.w3.eth.getBalance(_to)41 # 解é_fromè´¦æ·ï¼å¹¶è½¬è´¦42 self.w3.personal.unlockAccount(_from, _pwd)43 tx_hash = self.w3.eth.sendTransaction({"from": _from, "to": _to, "value": _value, "gas": config.gas, "gasPrice": config.gas_price})44 # çå¾
确认45 ret = self.wait_tx_confirm(tx_hash)46 print("send_ether gas used:", ret["gasUsed"])47 # æ¥çè´¦æ·_toä½é¢48 balance_new = self.w3.eth.getBalance(_to)49 self.assertEqual(balance_new - balance_old, _value)50 # åétoken51 # _from åéæ¹52 # _to æ¥æ¶æ¹53 # _value tokençæ°é54 # _pwd _fromç解éå¯ç 55 # _expect è´¦æ·_to tokençååå¼56 def transfer(self, _from, _to, _value, _pwd, _expect):57 _from = Web3.toChecksumAddress(_from)58 _to = Web3.toChecksumAddress(_to)59 # è®°å½è´¦æ·_fromçä½é¢60 balance_old_from = self.sht.functions.balanceOf(_from).call()61 # è®°å½è´¦æ·_toçä½é¢62 balance_old = self.sht.functions.balanceOf(_to).call()63 # 解é_fromè´¦æ·ï¼å¹¶åé_vaue个tokenç»è´¦æ·_to64 self.w3.personal.unlockAccount(_from, _pwd)65 tx_hash = self.sht.functions.transfer(_to, _value).transact({"from": _from, "gas": config.gas, "gasPrice": config.gas_price})66 # çå¾
确认67 ret = self.wait_tx_confirm(tx_hash)68 print("transfer gas used:", ret["gasUsed"])69 # æ¥çè´¦æ·_fromçä½é¢70 balance_new_from = self.sht.functions.balanceOf(_from).call()71 # æ¥çè´¦æ·_toçä½é¢72 balance_new = self.sht.functions.balanceOf(_to).call()73 self.assertEqual(balance_new - balance_old, _expect)74 self.assertEqual(balance_old_from - balance_new_from, _expect)75 def transfer_from(self, _spender, _from, _to, _value, _pwd, _expect):76 _spender = Web3.toChecksumAddress(_spender)77 _from = Web3.toChecksumAddress(_from)78 _to = Web3.toChecksumAddress(_to)79 # è®°å½è´¦æ·_fromçä½é¢80 balance_old_from = self.sht.functions.balanceOf(_from).call()81 # è®°å½è´¦æ·_toçä½é¢82 balance_old = self.sht.functions.balanceOf(_to).call()83 # è®°å½è´¦æ· _spender 代çä½é¢84 approve_balance_old = self.sht.functions.allowance(_from, _spender).call()85 # 解é_spenderè´¦æ·ï¼å¹¶åé_vaue个tokenç»è´¦æ·_to86 self.w3.personal.unlockAccount(_spender, _pwd)87 tx_hash = self.sht.functions.transferFrom(_from, _to, _value).transact({"from": _spender, "gas": config.gas, "gasPrice": config.gas_price})88 # çå¾
确认89 ret = self.wait_tx_confirm(tx_hash)90 print("transfer_from gas used:", ret["gasUsed"])91 # æ¥çè´¦æ·_fromçä½é¢92 balance_new_from = self.sht.functions.balanceOf(_from).call()93 # æ¥çè´¦æ·_toçä½é¢94 balance_new = self.sht.functions.balanceOf(_to).call()95 # æ¥çè´¦æ· _spender 代çä½é¢96 approve_balance_new = self.sht.functions.allowance(_from, _spender).call()97 self.assertEqual(balance_new - balance_old, _expect)98 self.assertEqual(balance_old_from - balance_new_from, _expect)99 self.assertEqual(approve_balance_old - approve_balance_new , _expect)100 def approve(self, _owner, _spender, _value, _pwd, _expect):101 # 转æ¢å°å102 _owner = Web3.toChecksumAddress(_owner)103 _spender = Web3.toChecksumAddress(_spender)104 # æ¥è¯¢ä»£çä½é¢105 ret = self.sht.functions.allowance(_owner, _spender).call()106 if ret == _value:107 return108 # 解é_ownerè´¦æ·ï¼å¹¶è®¾ç½®ä»£çä½é¢109 self.w3.personal.unlockAccount(_owner, _pwd)110 tx_hash = self.sht.functions.approve(_spender, _value).transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})111 # çå¾
确认112 ret = self.wait_tx_confirm(tx_hash)113 print("approve gas used:", ret["gasUsed"])114 # æ¥è¯¢ä»£çä½é¢115 ret = self.sht.functions.allowance(_owner, _spender).call()116 self.assertEqual(ret, _expect)117 def increase_approval(self, _owner, _spender, _value, _pwd, _expect):118 _owner = Web3.toChecksumAddress(_owner)119 _spender = Web3.toChecksumAddress(_spender)120 # è®°å½ä»£çä½é¢121 balance_old = self.sht.functions.allowance(_owner, _spender).call()122 # 解é_ownerè´¦æ·ï¼å¹¶è®¾ç½®ä»£çä½é¢123 self.w3.personal.unlockAccount(_owner, _pwd)124 tx_hash = self.sht.functions.increaseApproval(_spender, _value).transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})125 # çå¾
确认126 ret = self.wait_tx_confirm(tx_hash)127 print("increase_approval gas used:", ret["gasUsed"])128 # æ¥è¯¢ä»£çä½é¢129 balance_new = self.sht.functions.allowance(_owner, _spender).call()130 self.assertEqual(balance_new - balance_old, _expect)131 def decrease_approval(self, _owner, _spender, _value, _pwd, _expect):132 _owner = Web3.toChecksumAddress(_owner)133 _spender = Web3.toChecksumAddress(_spender)134 # è®°å½ä»£çä½é¢135 balance_old = self.sht.functions.allowance(_owner, _spender).call()136 # 解é_ownerè´¦æ·ï¼å¹¶è®¾ç½®ä»£çä½é¢137 self.w3.personal.unlockAccount(_owner, _pwd)138 tx_hash = self.sht.functions.decreaseApproval(_spender, _value).transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})139 # çå¾
确认140 ret = self.wait_tx_confirm(tx_hash)141 print("decrease_approval gas used:", ret["gasUsed"])142 # æ¥è¯¢ä»£çä½é¢143 balance_new = self.sht.functions.allowance(_owner, _spender).call()144 self.assertEqual(balance_old - balance_new, _expect)145 def batch_transfer(self, _from, _receivers, _values, _pwd, _expects):146 _from = Web3.toChecksumAddress(_from)147 # è®°å½è´¦æ·_fromçä½é¢148 balance_old_from = self.sht.functions.balanceOf(_from).call()149 # 转æ¢å°åï¼å¹¶è®°å½ä½é¢150 balances_old = []151 for i in range(len(_receivers)):152 _receivers[i] = Web3.toChecksumAddress(_receivers[i])153 balances_old.append(self.sht.functions.balanceOf(_receivers[i]).call())154 # 解é_fromè´¦æ·ï¼å¹¶æ¹éåé155 self.w3.personal.unlockAccount(_from, _pwd)156 tx_hash = self.sht.functions.batchTransfer(_receivers, _values).transact({"from": _from, "gas": config.gas, "gasPrice": config.gas_price})157 # çå¾
确认158 ret = self.wait_tx_confirm(tx_hash)159 print("batch_transfer with", str(len(_receivers)), "receivers gas used:", ret["gasUsed"])160 # æ¥çè´¦æ·_fromçä½é¢161 balance_new_from = self.sht.functions.balanceOf(_from).call()162 # æ¥è¯¢ä½é¢ï¼å¹¶éªè¯163 for i in range(len(_receivers)):164 ret = self.sht.functions.balanceOf(_receivers[i]).call()165 self.assertEqual(ret - balances_old[i], _expects[i])166 # 计ç®expectï¼éªè¯_fromä½é¢167 expect = 0168 for i in range(len(_expects)):169 expect = expect + _expects[i]170 self.assertEqual(balance_old_from - balance_new_from, expect)171 def batch_transfer_from(self, _spender, _from, _receivers, _values, _pwd, _expects):172 _spender = Web3.toChecksumAddress(_spender)173 _from = Web3.toChecksumAddress(_from)174 # è®°å½è´¦æ·_fromçä½é¢175 balance_old_from = self.sht.functions.balanceOf(_from).call()176 # 转æ¢å°åï¼å¹¶è®°å½ä½é¢177 balances_old = []178 for i in range(len(_receivers)):179 _receivers[i] = Web3.toChecksumAddress(_receivers[i])180 balances_old.append(self.sht.functions.balanceOf(_receivers[i]).call())181 # 解é_fromè´¦æ·ï¼å¹¶æ¹éåé182 self.w3.personal.unlockAccount(_spender, _pwd)183 tx_hash = self.sht.functions.batchTransferFrom(_from, _receivers, _values).transact({"from": _spender, "gas": config.gas, "gasPrice": config.gas_price})184 # çå¾
确认185 ret = self.wait_tx_confirm(tx_hash)186 print("batch_transfer_from with", str(len(_receivers)), "receivers gas used:", ret["gasUsed"])187 # æ¥çè´¦æ·_fromçä½é¢188 balance_new_from = self.sht.functions.balanceOf(_from).call()189 # æ¥è¯¢ä½é¢ï¼å¹¶éªè¯190 for i in range(len(_receivers)):191 ret = self.sht.functions.balanceOf(_receivers[i]).call()192 self.assertEqual(ret - balances_old[i], _expects[i])193 # 计ç®expectï¼éªè¯_fromä½é¢194 expect = 0195 for i in range(len(_expects)):196 expect = expect + _expects[i]197 self.assertEqual(balance_old_from - balance_new_from, expect)198 def transfer_by_date(self, _from, _to, _values, _dates, _pwd, _expect):199 _from = Web3.toChecksumAddress(_from)200 _to = Web3.toChecksumAddress(_to)201 # è®°å½è´¦æ·_fromçä½é¢202 balance_old_from = self.sht.functions.totalBalanceOf(_from).call()203 # è®°å½ä½é¢204 balance_old = self.sht.functions.totalBalanceOf(_to).call()205 # 解é_fromè´¦æ·ï¼å¹¶æ¹éåé206 self.w3.personal.unlockAccount(_from, _pwd)207 tx_hash = self.sht.functions.transferByDate(_to, _values, _dates).transact({"from": _from, "gas": config.gas, "gasPrice": config.gas_price})208 # çå¾
确认209 ret = self.wait_tx_confirm(tx_hash)210 print("transfer_by_date with", str(len(_dates)), "dates gas used:", ret["gasUsed"])211 # æ¥çè´¦æ·_fromçä½é¢,并éªè¯212 balance_new_from = self.sht.functions.totalBalanceOf(_from).call()213 self.assertEqual(balance_old_from - balance_new_from, _expect)214 # æ¥è¯¢ä½é¢ï¼å¹¶éªè¯215 balance_new = self.sht.functions.totalBalanceOf(_to).call()216 self.assertEqual(balance_new - balance_old, _expect)217 def transfer_from_by_date(self, _spender, _from, _to, _values, _dates, _pwd, _expect):218 _spender = Web3.toChecksumAddress(_spender)219 _from = Web3.toChecksumAddress(_from)220 _to = Web3.toChecksumAddress(_to)221 # è®°å½è´¦æ·_fromçä½é¢,并éªè¯222 balance_old_from = self.sht.functions.totalBalanceOf(_from).call()223 # è®°å½ä½é¢224 balance_old = self.sht.functions.totalBalanceOf(_to).call()225 # 解é_fromè´¦æ·ï¼å¹¶æ¹éåé226 self.w3.personal.unlockAccount(_spender, _pwd)227 tx_hash = self.sht.functions.transferFromByDate(_from, _to, _values, _dates).transact({"from": _spender, "gas": config.gas, "gasPrice": config.gas_price})228 # çå¾
确认229 ret = self.wait_tx_confirm(tx_hash)230 print("transfer_from_by_date with", str(len(_dates)), "dates gas used:", ret["gasUsed"])231 # æ¥çè´¦æ·_fromçä½é¢,并éªè¯232 balance_new_from = self.sht.functions.totalBalanceOf(_from).call()233 self.assertEqual(balance_old_from - balance_new_from, _expect)234 # æ¥è¯¢ä½é¢ï¼å¹¶éªè¯235 balance_new = self.sht.functions.totalBalanceOf(_to).call()236 self.assertEqual(balance_new - balance_old, _expect)237 def add_token_to_fund(self, _owner, _values, _pwd, _expect):238 _owner = Web3.toChecksumAddress(_owner)239 # è®°å½ _owner ä½é¢240 balance_old = self.sht.functions.balanceOf(_owner).call()241 # è®°å½ fundAccount ä½é¢242 fund_account = self.sht.functions.fundAccount().call()243 fund_account_balance_old = self.sht.functions.balanceOf(fund_account).call()244 # 解é _owner è´¦æ·ï¼å¹¶éæ¾tokenå°fund245 self.w3.personal.unlockAccount(_owner, _pwd)246 tx_hash = self.sht.functions.addTokenToFund(_values).transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})247 # çå¾
确认248 ret = self.wait_tx_confirm(tx_hash)249 print("add_token_to_fund value", str(_values), "gas used:", ret["gasUsed"])250 # æ¥çè´¦æ·_fromçä½é¢,并éªè¯251 balance_new = self.sht.functions.balanceOf(_owner).call()252 self.assertEqual(balance_old - balance_new, _expect)253 # æ¥è¯¢ä½é¢ï¼å¹¶éªè¯254 fund_account_balance_new = self.sht.functions.balanceOf(fund_account).call()255 self.assertEqual(fund_account_balance_new - fund_account_balance_old , _expect)256 # æ¸
空账æ·ä½é¢257 # _who å°è¦æ¸
空çè´¦æ·258 # _pwd è´¦æ·_whoç解éå¯ç 259 def clear_all_sht(self, _who, _collect, _pwd):260 _who = Web3.toChecksumAddress(_who)261 # è·åè´¦æ·ä½é¢262 balance = self.sht.functions.balanceOf(_who).call()263 # å°ææä½é¢è½¬è´¦å°åæ¶è´¦æ·264 if balance > 0:265 self.transfer(_who, _collect, balance, _pwd, balance)266 # æ¸
空ææå¯ç¨ä½é¢267 # å 为å¯è½ææªå°æçä½é¢ï¼è¿é¨åä½é¢ä¸è½æä½268 def clear_all_ava_sht(self, _who, _collect, _pwd):269 _who = Web3.toChecksumAddress(_who)270 # è·åè´¦æ·ä½é¢271 balance = self.sht.functions.balanceOf(_who).call()272 # å°ææä½é¢è½¬è´¦å°åæ¶è´¦æ·273 if balance > 0:274 self.transfer(_who, _collect, balance, _pwd, balance)275 # å»ç»è´¦æ·276 # _admin 管çå277 # _who 被å»ç»è´¦æ·278 # _pwd 管çå解éå¯ç 279 def froze(self, _admin, _who, _pwd):280 _admin = Web3.toChecksumAddress(_admin)281 _who = Web3.toChecksumAddress(_who)282 # æ¥çç¶æ283 ret = self.sht.functions.frozenList(_who).call()284 if ret == True:285 return286 # 解é_adminè´¦æ·å¹¶å»ç»è´¦æ·287 self.w3.personal.unlockAccount(_admin, _pwd)288 tx_hash = self.sht.functions.froze(_who).transact({"from": _admin, "gas": config.gas, "gasPrice": config.gas_price})289 # çå¾
确认290 ret = self.wait_tx_confirm(tx_hash)291 print("froze gas used:", ret["gasUsed"])292 ret = self.sht.functions.frozenList(_who).call()293 self.assertEqual(ret, True)294 # 解é¤å»ç»è´¦æ·295 # _admin 管çå296 # _who 被解é¤å»ç»è´¦æ·297 # _pwd 管çå解éå¯ç 298 def unfroze(self, _admin, _who, _pwd):299 _admin = Web3.toChecksumAddress(_admin)300 _who = Web3.toChecksumAddress(_who)301 # æ¥çç¶æ302 ret = self.sht.functions.frozenList(_who).call()303 if ret == False:304 return305 # 解é_adminè´¦æ·å¹¶è§£å»è´¦æ·306 self.w3.personal.unlockAccount(_admin, _pwd)307 tx_hash = self.sht.functions.unfroze(_who).transact({"from": _admin, "gasPrice": config.gas_price})308 # çå¾
确认309 ret = self.wait_tx_confirm(tx_hash)310 print("unfroze gas used:", ret["gasUsed"])311 ret = self.sht.functions.frozenList(_who).call()312 self.assertEqual(ret, False)313 def set_sell_price(self, _admin, _price, _pwd):314 _admin = Web3.toChecksumAddress(_admin)315 # æ¥çsell price316 ret = self.sht.functions.sellPrice().call()317 if ret == _price:318 return319 # ç»ç®_adminè´¦æ·å¹¶è®¾ç½®ä»·æ ¼320 self.w3.personal.unlockAccount(_admin, _pwd)321 tx_hash = self.sht.functions.setSellPrice(_price).transact({"from": _admin, "gasPrice": config.gas_price})322 # çå¾
确认323 ret = self.wait_tx_confirm(tx_hash)324 print("set_sell_price gas used:", ret["gasUsed"])325 ret = self.sht.functions.sellPrice().call()326 self.assertEqual(ret, _price)327 def set_buy_price(self, _admin, _price, _pwd):328 _admin = Web3.toChecksumAddress(_admin)329 # æ¥çsell price330 ret = self.sht.functions.buyPrice().call()331 if ret == _price:332 return333 # ç»ç®_adminè´¦æ·å¹¶è®¾ç½®ä»·æ ¼334 self.w3.personal.unlockAccount(_admin, _pwd)335 tx_hash = self.sht.functions.setBuyPrice(_price).transact({"from": _admin, "gasPrice": config.gas_price})336 # çå¾
确认337 ret = self.wait_tx_confirm(tx_hash)338 print("set_buy_price gas used:", ret["gasUsed"])339 ret = self.sht.functions.buyPrice().call()340 self.assertEqual(ret, _price)341 def open_buy(self, _owner, _pwd):342 _owner = Web3.toChecksumAddress(_owner)343 # æ¥çç¶æ344 ret = self.sht.functions.buyFlag().call()345 if ret == True:346 return347 # 解éownerè´¦æ·ï¼å¹¶æå¼è´ä¹°åèµåå¼å
³348 self.w3.personal.unlockAccount(_owner, _pwd)349 tx_hash = self.sht.functions.openBuy().transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})350 # çå¾
确认351 ret = self.wait_tx_confirm(tx_hash)352 print("open_buy_sell gas used:", ret["gasUsed"])353 ret = self.sht.functions.buyFlag().call()354 self.assertEqual(ret, True)355 def close_buy(self, _owner, _pwd):356 _owner = Web3.toChecksumAddress(_owner)357 # æ¥çç¶æ358 ret = self.sht.functions.buyFlag().call()359 if ret == False:360 return361 # 解éownerè´¦æ·ï¼å¹¶æå¼è´ä¹°åèµåå¼å
³362 self.w3.personal.unlockAccount(_owner, _pwd)363 tx_hash = self.sht.functions.closeBuy().transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})364 # çå¾
确认365 ret = self.wait_tx_confirm(tx_hash)366 print("close_buy_sell gas used:", ret["gasUsed"])367 ret = self.sht.functions.buyFlag().call()368 self.assertEqual(ret, False)369 def open_sell(self, _owner, _pwd):370 _owner = Web3.toChecksumAddress(_owner)371 # æ¥çç¶æ372 ret = self.sht.functions.sellFlag().call()373 if ret == True:374 return375 # 解éownerè´¦æ·ï¼å¹¶æå¼è´ä¹°åèµåå¼å
³376 self.w3.personal.unlockAccount(_owner, _pwd)377 tx_hash = self.sht.functions.openSell().transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})378 # çå¾
确认379 ret = self.wait_tx_confirm(tx_hash)380 print("open_buy_sell gas used:", ret["gasUsed"])381 ret = self.sht.functions.sellFlag().call()382 self.assertEqual(ret, True)383 def close_sell(self, _owner, _pwd):384 _owner = Web3.toChecksumAddress(_owner)385 # æ¥çç¶æ386 ret = self.sht.functions.sellFlag().call()387 if ret == False:388 return389 # 解éownerè´¦æ·ï¼å¹¶æå¼è´ä¹°åèµåå¼å
³390 self.w3.personal.unlockAccount(_owner, _pwd)391 tx_hash = self.sht.functions.closeSell().transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})392 # çå¾
确认393 ret = self.wait_tx_confirm(tx_hash)394 print("close_buy_sell gas used:", ret["gasUsed"])395 ret = self.sht.functions.sellFlag().call()396 self.assertEqual(ret, False)397 # _value: ether value398 def buy(self, _who, _value, _pwd, _expect):399 _who = Web3.toChecksumAddress(_who)400 # è®°å½è´¦æ·_whoçä½é¢401 balance_old = self.sht.functions.balanceOf(_who).call()402 who_eth_balance_old = self.w3.eth.getBalance(_who)403 # è®°å½ fundAccount eth ä½é¢404 fund_account = self.sht.functions.fundAccount().call()405 fund_account_eth_balance_old = self.w3.eth.getBalance(fund_account)406 fund_account_balance_old = self.sht.functions.balanceOf(fund_account).call()407 # 解é_whoè´¦æ·ï¼å¹¶è´ä¹°_value 以太å¸çtoken408 self.w3.personal.unlockAccount(_who, _pwd)409 tx_hash = self.sht.functions.buy().transact({"from": _who, "value": _value, "gas": config.gas, "gasPrice": config.gas_price})410 # çå¾
确认411 ret = self.wait_tx_confirm(tx_hash)412 print("buy gas used:", ret["gasUsed"])413 # è®°å½è´¦æ·_toçä½é¢414 balance_new = self.sht.functions.balanceOf(_who).call()415 self.assertEqual(balance_new - balance_old, _expect)416 # _who åå°ç eth æ¯å¦æ£ç¡®417 who_eth_balance_new = self.w3.eth.getBalance(_who)418 self.assertEqual(who_eth_balance_old - who_eth_balance_new, ret["gasUsed"] * config.gas_price + Web3.toWei(1, "ether"))419 # fundAccount å¢å ç eth æ¯å¦æ£ç¡®420 fund_account_eth_balance_new = self.w3.eth.getBalance( fund_account)421 self.assertEqual(fund_account_eth_balance_new - fund_account_eth_balance_old, Web3.toWei(1, "ether"))422 # fundAccount åå°ç token423 fund_account_balance_new = self.sht.functions.balanceOf(fund_account).call()424 self.assertEqual(fund_account_balance_old - fund_account_balance_new, _expect)425 def sell(self, _who, _value, _pwd, _expect):426 _who = Web3.toChecksumAddress(_who)427 # è®°å½è´¦æ·_whoçä½é¢428 balance_old = self.sht.functions.balanceOf(_who).call()429 eth_balance_old = self.w3.eth.getBalance(_who)430 # è®°å½ fundAccount token ä½é¢431 fund_account = self.sht.functions.fundAccount().call()432 fund_account_balance_old = self.sht.functions.balanceOf(fund_account).call()433 # è®°å½ å约 eth ä½é¢434 eth_sht_account_balance_old = self.w3.eth.getBalance(Web3.toChecksumAddress(sts.SIBBAY_SHT_ADDRESS))435 # 解é_whoè´¦æ·ï¼å¹¶è´ä¹°_value 以太å¸çtoken436 self.w3.personal.unlockAccount(_who, _pwd)437 tx_hash = self.sht.functions.sell(_value).transact({"from": _who, "gas": config.gas, "gasPrice": config.gas_price})438 # çå¾
确认439 ret = self.wait_tx_confirm(tx_hash)440 print("sell gas used:", ret["gasUsed"])441 sell_price = self.sht.functions.sellPrice().call()442 # è®°å½è´¦æ·_toçä½é¢443 balance_new = self.sht.functions.balanceOf(_who).call()444 self.assertEqual(balance_old - balance_new, _expect)445 # who eth446 eth_balance_new = self.w3.eth.getBalance(_who)447 self.assertEqual(eth_balance_new - eth_balance_old + ret["gasUsed"] * config.gas_price, int(_expect * sell_price / config.magnitude))448 # è®°å½ fundAccount ä½é¢449 fund_account_balance_new = self.sht.functions.balanceOf(fund_account).call()450 # è®°å½ å约 eth ä½é¢451 eth_sht_account_balance_new = self.w3.eth.getBalance(Web3.toChecksumAddress(sts.SIBBAY_SHT_ADDRESS))452 self.assertEqual(fund_account_balance_new - fund_account_balance_old, _expect)453 self.assertEqual(eth_sht_account_balance_old - eth_sht_account_balance_new, int(_expect * sell_price / config.magnitude))454 def pause(self, _owner, _pwd):455 _owner = Web3.toChecksumAddress(_owner)456 # æ¥çå约ç¶æ457 ret = self.sht.functions.paused().call()458 if ret == True:459 return460 # 解é_ownerè´¦æ·ï¼æåå约461 self.w3.personal.unlockAccount(_owner, _pwd)462 tx_hash = self.sht.functions.pause().transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})463 # çå¾
确认464 ret = self.wait_tx_confirm(tx_hash)465 print("pause gas used:", ret["gasUsed"])466 # æ¥çå约ç¶æ467 ret = self.sht.functions.paused().call()468 self.assertEqual(ret, True)469 def unpause(self, _owner, _pwd):470 _owner = Web3.toChecksumAddress(_owner)471 # æ¥çå约ç¶æ472 ret = self.sht.functions.paused().call()473 if ret == False:474 return475 # 解é_ownerè´¦æ·ï¼åæ¶æåå约476 self.w3.personal.unlockAccount(_owner, _pwd)477 tx_hash = self.sht.functions.unpause().transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})478 # çå¾
确认479 ret = self.wait_tx_confirm(tx_hash)480 print("unpause gas used:", ret["gasUsed"])481 # æ¥çå约ç¶æ482 ret = self.sht.functions.paused().call()483 self.assertEqual(ret, False)484 def add_administrator(self, _owner, _who, _pwd):485 _owner = Web3.toChecksumAddress(_owner)486 _who = Web3.toChecksumAddress(_who)487 # æ¥ç管çåç¶æ488 ret = self.sht.functions.adminList(_who).call()489 if ret == True:490 return491 # 解é_ownerè´¦æ·ï¼æ·»å _who为管çå492 self.w3.personal.unlockAccount(_owner, _pwd)493 tx_hash = self.sht.functions.addAdministrator(_who).transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})494 # çå¾
确认495 ret = self.wait_tx_confirm(tx_hash)496 print("add_administrator gas used:", ret["gasUsed"])497 # æ¥ç管çåç¶æ498 ret = self.sht.functions.adminList(_who).call()499 self.assertEqual(ret, True)500 def del_administrator(self, _owner, _who, _pwd):501 _owner = Web3.toChecksumAddress(_owner)502 _who = Web3.toChecksumAddress(_who)503 # æ¥ç管çåç¶æ504 ret = self.sht.functions.adminList(_who).call()505 if ret == False:506 return507 # 解é_ownerè´¦æ·ï¼æ·»å _who为管çå508 self.w3.personal.unlockAccount(_owner, _pwd)509 tx_hash = self.sht.functions.delAdministrator(_who).transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})510 # çå¾
确认511 ret = self.wait_tx_confirm(tx_hash)512 print("del_administrator gas used:", ret["gasUsed"])513 # æ¥ç管çåç¶æ514 ret = self.sht.functions.adminList(_who).call()515 self.assertEqual(ret, False)516 def withdraw(self, _owner, _pwd):517 _owner = Web3.toChecksumAddress(_owner)518 # æ¥çå约ä½é¢519 ret = self.w3.eth.getBalance(Web3.toChecksumAddress(sts.SIBBAY_SHT_ADDRESS))520 if ret == 0:521 return522 # 解é_ownerè´¦æ·ï¼ååå约ä½é¢523 self.w3.personal.unlockAccount(_owner, _pwd)524 tx_hash = self.sht.functions.withdraw().transact({"from": _owner, "gas": config.gas, "gasPrice": config.gas_price})525 # çå¾
确认526 ret = self.wait_tx_confirm(tx_hash)527 print("withdraw gas used:", ret["gasUsed"])528 # æ¥çå约ä½é¢529 ret = self.w3.eth.getBalance(Web3.toChecksumAddress(sts.SIBBAY_SHT_ADDRESS))530 self.assertEqual(ret, 0)531 def balance_of(self, address, _expect):532 # æ¥çè´¦æ· address çä½é¢533 _who = Web3.toChecksumAddress(address)534 balance_value = self.sht.functions.balanceOf(_who).call()535 self.assertEqual(balance_value, _expect)536 def eth_balance_of(self, address, _expect):537 ret = self.w3.eth.getBalance(Web3.toChecksumAddress(address))538 self.assertEqual(ret, _expect)539 def eth_balance_of_sht_address(self, _expect):...
node_cli.py
Source:node_cli.py
...99 print ("%s" % cmd)100 self.pexpect = pexpect.spawn(cmd, timeout=4)101 # Add delay to ensure that the process is ready to receive commands.102 time.sleep(0.2)103 self._expect('spinel-cli >')104 self.debug(int(os.getenv('DEBUG', '0')))105 def _expect(self, pattern, timeout=-1, *args, **kwargs):106 """ Process simulator events until expected the pattern. """107 if timeout == -1:108 timeout = self.pexpect.timeout109 assert timeout > 0110 while timeout > 0:111 try:112 return self.pexpect.expect(pattern, 0.1, *args, **kwargs)113 except pexpect.TIMEOUT:114 timeout -= 0.1115 self.simulator.go(0)116 if timeout <= 0:117 raise118 def __init_soc(self, nodeid):119 """ Initialize a System-on-a-chip node connected via UART. """120 import fdpexpect121 serialPort = '/dev/ttyUSB%d' % ((nodeid-1)*2)122 self.pexpect = fdpexpect.fdspawn(os.open(serialPort, os.O_RDWR|os.O_NONBLOCK|os.O_NOCTTY))123 def __del__(self):124 self.destroy()125 def destroy(self):126 if not self.pexpect:127 return128 if hasattr(self.pexpect, 'proc') and self.pexpect.proc.poll() is None or \129 not hasattr(self.pexpect, 'proc') and self.pexpect.isalive():130 print("%d: exit" % self.nodeid)131 self.pexpect.send('exit\n')132 self.pexpect.expect(pexpect.EOF)133 self.pexpect = None134 def read_cert_messages_in_commissioning_log(self, timeout=-1):135 """Get the log of the traffic after DTLS handshake.136 """137 format_str = br"=+?\[\[THCI\].*?type=%s.*?\].*?=+?[\s\S]+?-{40,}"138 join_fin_req = format_str % br"JOIN_FIN\.req"139 join_fin_rsp = format_str % br"JOIN_FIN\.rsp"140 dummy_format_str = br"\[THCI\].*?type=%s.*?"141 join_ent_ntf = dummy_format_str % br"JOIN_ENT\.ntf"142 join_ent_rsp = dummy_format_str % br"JOIN_ENT\.rsp"143 pattern = (b"(" + join_fin_req + b")|(" + join_fin_rsp + b")|("+ join_ent_ntf + b")|(" + join_ent_rsp + b")")144 messages = []145 # There are at most 4 cert messages both for joiner and commissioner146 for _ in range(0, 4):147 try:148 self._expect(pattern, timeout=timeout)149 log = self.pexpect.match.group(0)150 messages.append(self._extract_cert_message(log))151 except:152 break153 return messages154 def _extract_cert_message(self, log):155 res = re.search(br"direction=\w+", log)156 assert res157 direction = res.group(0).split(b'=')[1].strip()158 res = re.search(br"type=\S+", log)159 assert res160 type = res.group(0).split(b'=')[1].strip()161 payload = bytearray([])162 payload_len = 0163 if type in [b"JOIN_FIN.req", b"JOIN_FIN.rsp"]:164 res = re.search(br"len=\d+", log)165 assert res166 payload_len = int(res.group(0).split(b'=')[1].strip())167 hex_pattern = br"\|(\s([0-9a-fA-F]{2}|\.\.))+?\s+?\|"168 while True:169 res = re.search(hex_pattern, log)170 if not res:171 break172 data = [int(hex, 16) for hex in res.group(0)[1:-1].split(b' ') if hex and hex != b'..']173 payload += bytearray(data)174 log = log[res.end()-1:]175 assert len(payload) == payload_len176 return (direction, type, payload)177 def send_command(self, cmd, go=True):178 print("%d: %s" % (self.nodeid, cmd))179 self.pexpect.send(cmd + '\n')180 if go:181 self.simulator.go(0, nodeid=self.nodeid)182 sys.stdout.flush()183 def get_commands(self):184 self.send_command('?')185 self._expect('Commands:')186 commands = []187 while True:188 i = self._expect(['Done', r'(\S+)'])189 if i != 0:190 commands.append(self.pexpect.match.groups()[0])191 else:192 break193 return commands194 def set_mode(self, mode):195 cmd = 'mode ' + mode196 self.send_command(cmd)197 self._expect('Done')198 def debug(self, level):199 # `debug` command will not trigger interaction with simulator200 self.send_command('debug '+ str(level), go=False)201 def interface_up(self):202 self.send_command('ifconfig up')203 self._expect('Done')204 def interface_down(self):205 self.send_command('ifconfig down')206 self._expect('Done')207 def thread_start(self):208 self.send_command('thread start')209 self._expect('Done')210 def thread_stop(self):211 self.send_command('thread stop')212 self._expect('Done')213 def commissioner_start(self):214 cmd = 'commissioner start'215 self.send_command(cmd)216 self._expect('Done')217 def commissioner_add_joiner(self, addr, psk):218 cmd = 'commissioner joiner add ' + addr + ' ' + psk219 self.send_command(cmd)220 self._expect('Done')221 def joiner_start(self, pskd='', provisioning_url=''):222 cmd = 'joiner start ' + pskd + ' ' + provisioning_url223 self.send_command(cmd)224 self._expect('Done')225 def clear_whitelist(self):226 cmd = 'macfilter addr clear'227 self.send_command(cmd)228 self._expect('Done')229 def enable_whitelist(self):230 cmd = 'macfilter addr whitelist'231 self.send_command(cmd)232 self._expect('Done')233 def disable_whitelist(self):234 cmd = 'macfilter addr disable'235 self.send_command(cmd)236 self._expect('Done')237 def add_whitelist(self, addr, rssi=None):238 cmd = 'macfilter addr add ' + addr239 if rssi != None:240 cmd += ' ' + str(rssi)241 self.send_command(cmd)242 self._expect('Done')243 def remove_whitelist(self, addr):244 cmd = 'macfilter addr remove ' + addr245 self.send_command(cmd)246 self._expect('Done')247 def get_addr16(self):248 self.send_command('rloc16')249 i = self._expect('([0-9a-fA-F]{4})')250 if i == 0:251 addr16 = int(self.pexpect.match.groups()[0], 16)252 self._expect('Done')253 return addr16254 def get_router_id(self):255 rloc16 = self.get_addr16()256 return (rloc16 >> 10)257 def get_addr64(self):258 self.send_command('extaddr')259 i = self._expect('([0-9a-fA-F]{16})')260 if i == 0:261 addr64 = self.pexpect.match.groups()[0].decode("utf-8")262 self._expect('Done')263 return addr64264 def get_eui64(self):265 self.send_command('eui64')266 i = self._expect('([0-9a-fA-F]{16})')267 if i == 0:268 addr64 = self.pexpect.match.groups()[0].decode("utf-8")269 self._expect('Done')270 return addr64271 def get_joiner_id(self):272 self.send_command('joinerid')273 i = self._expect('([0-9a-fA-F]{16})')274 if i == 0:275 addr = self.pexpect.match.groups()[0].decode("utf-8")276 self._expect('Done')277 return addr278 def get_channel(self):279 self.send_command('channel')280 i = self._expect(r'(\d+)\r?\n')281 if i == 0:282 channel = int(self.pexpect.match.groups()[0])283 self._expect('Done')284 return channel285 def set_channel(self, channel):286 cmd = 'channel %d' % channel287 self.send_command(cmd)288 self._expect('Done')289 def get_masterkey(self):290 self.send_command('masterkey')291 i = self._expect('([0-9a-fA-F]{32})')292 if i == 0:293 masterkey = self.pexpect.match.groups()[0].decode("utf-8")294 self._expect('Done')295 return masterkey296 def set_masterkey(self, masterkey):297 cmd = 'masterkey ' + masterkey298 self.send_command(cmd)299 self._expect('Done')300 def get_key_sequence_counter(self):301 self.send_command('keysequence counter')302 i = self._expect(r'(\d+)\r?\n')303 if i == 0:304 key_sequence_counter = int(self.pexpect.match.groups()[0])305 self._expect('Done')306 return key_sequence_counter307 def set_key_sequence_counter(self, key_sequence_counter):308 cmd = 'keysequence counter %d' % key_sequence_counter309 self.send_command(cmd)310 self._expect('Done')311 def set_key_switch_guardtime(self, key_switch_guardtime):312 cmd = 'keysequence guardtime %d' % key_switch_guardtime313 self.send_command(cmd)314 self._expect('Done')315 def set_network_id_timeout(self, network_id_timeout):316 cmd = 'networkidtimeout %d' % network_id_timeout317 self.send_command(cmd)318 self._expect('Done')319 def get_network_name(self):320 self.send_command('networkname')321 while True:322 i = self._expect(['Done', r'(\S+)'])323 if i != 0:324 network_name = self.pexpect.match.groups()[0].decode('utf-8')325 else:326 break327 return network_name328 def set_network_name(self, network_name):329 cmd = 'networkname ' + network_name330 self.send_command(cmd)331 self._expect('Done')332 def get_panid(self):333 self.send_command('panid')334 i = self._expect('([0-9a-fA-F]{4})')335 if i == 0:336 panid = int(self.pexpect.match.groups()[0], 16)337 self._expect('Done')338 return panid339 def set_panid(self, panid):340 cmd = 'panid %d' % panid341 self.send_command(cmd)342 self._expect('Done')343 def get_partition_id(self):344 self.send_command('leaderpartitionid')345 i = self._expect(r'(\d+)\r?\n')346 if i == 0:347 weight = self.pexpect.match.groups()[0]348 self._expect('Done')349 return weight350 def set_partition_id(self, partition_id):351 cmd = 'leaderpartitionid %d' % partition_id352 self.send_command(cmd)353 self._expect('Done')354 def set_router_upgrade_threshold(self, threshold):355 cmd = 'routerupgradethreshold %d' % threshold356 self.send_command(cmd)357 self._expect('Done')358 def set_router_downgrade_threshold(self, threshold):359 cmd = 'routerdowngradethreshold %d' % threshold360 self.send_command(cmd)361 self._expect('Done')362 def release_router_id(self, router_id):363 cmd = 'releaserouterid %d' % router_id364 self.send_command(cmd)365 self._expect('Done')366 def get_state(self):367 states = ['detached', 'child', 'router', 'leader']368 self.send_command('state')369 match = self._expect(states)370 self._expect('Done')371 return states[match]372 def set_state(self, state):373 cmd = 'state ' + state374 self.send_command(cmd)375 self._expect('Done')376 def get_timeout(self):377 self.send_command('childtimeout')378 i = self._expect(r'(\d+)\r?\n')379 if i == 0:380 timeout = self.pexpect.match.groups()[0]381 self._expect('Done')382 return timeout383 def set_timeout(self, timeout):384 cmd = 'childtimeout %d' % timeout385 self.send_command(cmd)386 self._expect('Done')387 def set_max_children(self, number):388 cmd = 'childmax %d' % number389 self.send_command(cmd)390 self._expect('Done')391 def get_weight(self):392 self.send_command('leaderweight')393 i = self._expect(r'(\d+)\r?\n')394 if i == 0:395 weight = self.pexpect.match.groups()[0]396 self._expect('Done')397 return weight398 def set_weight(self, weight):399 cmd = 'leaderweight %d' % weight400 self.send_command(cmd)401 self._expect('Done')402 def add_ipaddr(self, ipaddr):403 cmd = 'ipaddr add ' + ipaddr404 self.send_command(cmd)405 self._expect('Done')406 def get_addrs(self):407 addrs = []408 self.send_command('ipaddr')409 while True:410 i = self._expect([r'(\S+(:\S*)+)\r?\n', 'Done'])411 if i == 0:412 addrs.append(self.pexpect.match.groups()[0].decode("utf-8"))413 elif i == 1:414 break415 return addrs416 def get_addr(self, prefix):417 network = ipaddress.ip_network(u'%s' % str(prefix))418 addrs = self.get_addrs()419 for addr in addrs:420 if isinstance(addr, bytearray):421 addr = bytes(addr)422 elif isinstance(addr, str) and sys.version_info[0] == 2:423 addr = addr.decode("utf-8")424 ipv6_address = ipaddress.ip_address(addr)425 if ipv6_address in network:426 return ipv6_address.exploded427 return None428 def get_eidcaches(self):429 eidcaches = []430 self.send_command('eidcache')431 while True:432 i = self._expect([r'([a-fA-F0-9\:]+) ([a-fA-F0-9]+)\r?\n', 'Done'])433 if i == 0:434 eid = self.pexpect.match.groups()[0].decode("utf-8")435 rloc = self.pexpect.match.groups()[1].decode("utf-8")436 eidcaches.append((eid, rloc))437 elif i == 1:438 break439 return eidcaches440 def add_service(self, enterpriseNumber, serviceData, serverData):441 cmd = 'service add ' + enterpriseNumber + ' ' + serviceData+ ' ' + serverData442 self.send_command(cmd)443 self._expect('Done')444 def remove_service(self, enterpriseNumber, serviceData):445 cmd = 'service remove ' + enterpriseNumber + ' ' + serviceData446 self.send_command(cmd)447 self._expect('Done')448 def __getLinkLocalAddress(self):449 for ip6Addr in self.get_addrs():450 if re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I):451 return ip6Addr452 return None453 def __getGlobalAddress(self):454 global_address = []455 for ip6Addr in self.get_addrs():456 if (not re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I)) and \457 (not re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I)) and \458 (not re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I)):459 global_address.append(ip6Addr)460 return global_address461 def __getRloc(self):462 for ip6Addr in self.get_addrs():463 if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \464 re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I) and \465 not(re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I)):466 return ip6Addr467 return None468 def __getAloc(self):469 aloc = []470 for ip6Addr in self.get_addrs():471 if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \472 re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I) and \473 re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I):474 aloc.append(ip6Addr)475 return aloc476 def __getMleid(self):477 for ip6Addr in self.get_addrs():478 if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \479 not(re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I)):480 return ip6Addr481 return None482 def get_ip6_address(self, address_type):483 """Get specific type of IPv6 address configured on thread device.484 Args:485 address_type: the config.ADDRESS_TYPE type of IPv6 address.486 Returns:487 IPv6 address string.488 """489 if address_type == config.ADDRESS_TYPE.LINK_LOCAL:490 return self.__getLinkLocalAddress()491 elif address_type == config.ADDRESS_TYPE.GLOBAL:492 return self.__getGlobalAddress()493 elif address_type == config.ADDRESS_TYPE.RLOC:494 return self.__getRloc()495 elif address_type == config.ADDRESS_TYPE.ALOC:496 return self.__getAloc()497 elif address_type == config.ADDRESS_TYPE.ML_EID:498 return self.__getMleid()499 else:500 return None501 return None502 def get_context_reuse_delay(self):503 self.send_command('contextreusedelay')504 i = self._expect(r'(\d+)\r?\n')505 if i == 0:506 timeout = self.pexpect.match.groups()[0]507 self._expect('Done')508 return timeout509 def set_context_reuse_delay(self, delay):510 cmd = 'contextreusedelay %d' % delay511 self.send_command(cmd)512 self._expect('Done')513 def add_prefix(self, prefix, flags, prf = 'med'):514 cmd = 'prefix add ' + prefix + ' ' + flags + ' ' + prf515 self.send_command(cmd)516 self._expect('Done')517 def remove_prefix(self, prefix):518 cmd = 'prefix remove ' + prefix519 self.send_command(cmd)520 self._expect('Done')521 def add_route(self, prefix, prf = 'med'):522 cmd = 'route add ' + prefix + ' ' + prf523 self.send_command(cmd)524 self._expect('Done')525 def remove_route(self, prefix):526 cmd = 'route remove ' + prefix527 self.send_command(cmd)528 self._expect('Done')529 def register_netdata(self):530 self.send_command('netdataregister')531 self._expect('Done')532 def energy_scan(self, mask, count, period, scan_duration, ipaddr):533 cmd = 'commissioner energy ' + str(mask) + ' ' + str(count) + ' ' + str(period) + ' ' + str(scan_duration) + ' ' + ipaddr534 self.send_command(cmd)535 if isinstance(self.simulator, simulator.VirtualTime):536 self.simulator.go(8)537 timeout = 1538 else:539 timeout = 8540 self._expect('Energy:', timeout=timeout)541 def panid_query(self, panid, mask, ipaddr):542 cmd = 'commissioner panid ' + str(panid) + ' ' + str(mask) + ' ' + ipaddr543 self.send_command(cmd)544 if isinstance(self.simulator, simulator.VirtualTime):545 self.simulator.go(8)546 timeout = 1547 else:548 timeout = 8549 self._expect('Conflict:', timeout=timeout)550 def scan(self):551 self.send_command('scan')552 results = []553 while True:554 i = self._expect([r'\|\s(\S+)\s+\|\s(\S+)\s+\|\s([0-9a-fA-F]{4})\s\|\s([0-9a-fA-F]{16})\s\|\s(\d+)\r?\n',555 'Done'])556 if i == 0:557 results.append(self.pexpect.match.groups())558 else:559 break560 return results561 def ping(self, ipaddr, num_responses=1, size=None, timeout=5):562 cmd = 'ping ' + ipaddr563 if size != None:564 cmd += ' ' + str(size)565 self.send_command(cmd)566 if isinstance(self.simulator, simulator.VirtualTime):567 self.simulator.go(timeout)568 result = True569 try:570 responders = {}571 while len(responders) < num_responses:572 i = self._expect([r'from (\S+):'])573 if i == 0:574 responders[self.pexpect.match.groups()[0]] = 1575 self._expect('\n')576 except (pexpect.TIMEOUT, socket.timeout):577 result = False578 if isinstance(self.simulator, simulator.VirtualTime):579 self.simulator.sync_devices()580 return result581 def reset(self):582 self.send_command('reset')583 time.sleep(0.1)584 def set_router_selection_jitter(self, jitter):585 cmd = 'routerselectionjitter %d' % jitter586 self.send_command(cmd)587 self._expect('Done')588 def set_active_dataset(self, timestamp, panid=None, channel=None, channel_mask=None, master_key=None):589 self.send_command('dataset clear')590 self._expect('Done')591 cmd = 'dataset activetimestamp %d' % timestamp592 self.send_command(cmd)593 self._expect('Done')594 if panid != None:595 cmd = 'dataset panid %d' % panid596 self.send_command(cmd)597 self._expect('Done')598 if channel != None:599 cmd = 'dataset channel %d' % channel600 self.send_command(cmd)601 self._expect('Done')602 if channel_mask != None:603 cmd = 'dataset channelmask %d' % channel_mask604 self.send_command(cmd)605 self._expect('Done')606 if master_key != None:607 cmd = 'dataset masterkey ' + master_key608 self.send_command(cmd)609 self._expect('Done')610 self.send_command('dataset commit active')611 self._expect('Done')612 def set_pending_dataset(self, pendingtimestamp, activetimestamp, panid=None, channel=None):613 self.send_command('dataset clear')614 self._expect('Done')615 cmd = 'dataset pendingtimestamp %d' % pendingtimestamp616 self.send_command(cmd)617 self._expect('Done')618 cmd = 'dataset activetimestamp %d' % activetimestamp619 self.send_command(cmd)620 self._expect('Done')621 if panid != None:622 cmd = 'dataset panid %d' % panid623 self.send_command(cmd)624 self._expect('Done')625 if channel != None:626 cmd = 'dataset channel %d' % channel627 self.send_command(cmd)628 self._expect('Done')629 self.send_command('dataset commit pending')630 self._expect('Done')631 def announce_begin(self, mask, count, period, ipaddr):632 cmd = 'commissioner announce ' + str(mask) + ' ' + str(count) + ' ' + str(period) + ' ' + ipaddr633 self.send_command(cmd)634 self._expect('Done')635 def send_mgmt_active_set(self, active_timestamp=None, channel=None, channel_mask=None, extended_panid=None,636 panid=None, master_key=None, mesh_local=None, network_name=None, binary=None):637 cmd = 'dataset mgmtsetcommand active '638 if active_timestamp != None:639 cmd += 'activetimestamp %d ' % active_timestamp640 if channel != None:641 cmd += 'channel %d ' % channel642 if channel_mask != None:643 cmd += 'channelmask %d ' % channel_mask644 if extended_panid != None:645 cmd += 'extpanid ' + extended_panid + ' '646 if panid != None:647 cmd += 'panid %d ' % panid648 if master_key != None:649 cmd += 'masterkey ' + master_key + ' '650 if mesh_local != None:651 cmd += 'localprefix ' + mesh_local + ' '652 if network_name != None:653 cmd += 'networkname ' + network_name + ' '654 if binary != None:655 cmd += 'binary ' + binary + ' '656 self.send_command(cmd)657 self._expect('Done')658 def send_mgmt_pending_set(self, pending_timestamp=None, active_timestamp=None, delay_timer=None, channel=None,659 panid=None, master_key=None, mesh_local=None, network_name=None):660 cmd = 'dataset mgmtsetcommand pending '661 if pending_timestamp != None:662 cmd += 'pendingtimestamp %d ' % pending_timestamp663 if active_timestamp != None:664 cmd += 'activetimestamp %d ' % active_timestamp665 if delay_timer != None:666 cmd += 'delaytimer %d ' % delay_timer667 if channel != None:668 cmd += 'channel %d ' % channel669 if panid != None:670 cmd += 'panid %d ' % panid671 if master_key != None:672 cmd += 'masterkey ' + master_key + ' '673 if mesh_local != None:674 cmd += 'localprefix ' + mesh_local + ' '675 if network_name != None:676 cmd += 'networkname ' + network_name + ' '677 self.send_command(cmd)678 self._expect('Done')679 def coaps_start_psk(self, psk, pskIdentity):680 cmd = 'coaps set psk ' + psk + ' ' + pskIdentity681 self.send_command(cmd)682 self._expect('Done')683 cmd = 'coaps start'684 self.send_command(cmd)685 self._expect('Done')686 def coaps_start_x509(self):687 cmd = 'coaps set x509'688 self.send_command(cmd)689 self._expect('Done')690 cmd = 'coaps start'691 self.send_command(cmd)692 self._expect('Done')693 def coaps_set_resource_path(self, path):694 cmd = 'coaps resource ' + path695 self.send_command(cmd)696 self._expect('Done')697 def coaps_stop(self):698 cmd = 'coaps stop'699 self.send_command(cmd)700 if isinstance(self.simulator, simulator.VirtualTime):701 self.simulator.go(5)702 timeout = 1703 else:704 timeout = 5705 self._expect('Done', timeout=timeout)706 def coaps_connect(self, ipaddr):707 cmd = 'coaps connect ' + ipaddr708 self.send_command(cmd)709 if isinstance(self.simulator, simulator.VirtualTime):710 self.simulator.go(5)711 timeout = 1712 else:713 timeout = 5714 self._expect('CoAP Secure connected!', timeout=timeout)715 def coaps_disconnect(self):716 cmd = 'coaps disconnect'717 self.send_command(cmd)718 self._expect('Done')719 self.simulator.go(5)720 def coaps_get(self):721 cmd = 'coaps get test'722 self.send_command(cmd)723 if isinstance(self.simulator, simulator.VirtualTime):724 self.simulator.go(5)725 timeout = 1726 else:727 timeout = 5...
test_videosize.py
Source:test_videosize.py
...38 dar[0], dar[1])39 else:40 result += " "41 return result42 def _expect(self, iw, ih, ipar, ow, oh, opar, mw, mh, pref, ew, eh, epar):43 rw, rh, rpar = videosize.getVideoSize(iw, ih, ipar,44 ow, oh, opar,45 mw, mh, pref)46 #print "IN: %s - PREF: %s - OUT: %s" % (self._format(iw, ih, ipar),47 # self._format(ow, oh, opar),48 # self._format(rw, rh, rpar))49 idar = (float(iw) * ipar[0]) / (float(ih) * ipar[1])50 rdar = (float(rw) * rpar[0]) / (float(rh) * rpar[1])51 self.assertTrue(abs(idar - rdar) < 0.03, "Display Aspect Ratio Changed")52 self.assertEquals((rw, rh, rpar), (ew, eh, epar))53 def testMaxSizeWithPreferredSize(self):54 self._expect(320, 240, (1, 1),55 200, None, (1, 1),56 300, 200, None,57 200, 150, (1, 1))58 self._expect(320, 240, (1, 1),59 200, None, (1, 1),60 150, 100, None,61 133, 100, (1, 1))62 self._expect(320, 240, (1, 1),63 200, None, (1, 1),64 100, 50, None,65 67, 50, (1, 1))66 self._expect(320, 240, (1, 1),67 None, 180, (1, 1),68 300, 200, None,69 240, 180, (1, 1))70 self._expect(320, 240, (1, 1),71 None, 180, (1, 1),72 150, 100, None,73 133, 100, (1, 1))74 self._expect(320, 240, (1, 1),75 None, 180, (1, 1),76 100, 50, None,77 67, 50, (1, 1))78 self._expect(320, 240, (1, 1),79 180, 180, (1, 1),80 300, 200, None,81 180, 135, (1, 1))82 self._expect(320, 240, (1, 1),83 180, 180, (1, 1),84 150, 100, None,85 133, 100, (1, 1))86 self._expect(320, 240, (1, 1),87 180, 180, (1, 1),88 100, 50, None,89 67, 50, (1, 1))90 def testMaxSizeWithoutPreferredSize(self):91 self._expect(320, 240, (1, 1),92 None, None, (1, 1),93 200, 300, VideoScaleMethodEnum.upscale,94 200, 150, (1, 1))95 self._expect(320, 240, (1, 1),96 None, None, (1, 2),97 200, 300, VideoScaleMethodEnum.upscale,98 200, 75, (1, 2))99 self._expect(320, 240, (1, 1),100 None, None, (1, 3),101 200, 300, VideoScaleMethodEnum.upscale,102 200, 50, (1, 3))103 self._expect(320, 240, (1, 1),104 None, None, (2, 1),105 200, 300, VideoScaleMethodEnum.upscale,106 200, 300, (2, 1))107 self._expect(320, 240, (1, 1),108 None, None, (3, 1),109 200, 300, VideoScaleMethodEnum.upscale,110 133, 300, (3, 1))111 self._expect(320, 240, (1, 1),112 None, None, (1, 1),113 400, 500, VideoScaleMethodEnum.upscale,114 320, 240, (1, 1))115 self._expect(320, 240, (1, 1),116 None, None, (1, 2),117 400, 500, VideoScaleMethodEnum.upscale,118 400, 150, (1, 2))119 self._expect(320, 240, (1, 1),120 None, None, (1, 3),121 400, 500, VideoScaleMethodEnum.upscale,122 400, 100, (1, 3))123 self._expect(320, 240, (1, 1),124 None, None, (2, 1),125 400, 500, VideoScaleMethodEnum.upscale,126 320, 480, (2, 1))127 self._expect(320, 240, (1, 1),128 None, None, (3, 1),129 400, 500, VideoScaleMethodEnum.upscale,130 222, 500, (3, 1))131 def testPreferredSize(self):132 self._expect(320, 240, (1, 1),133 200, 200, (1, 1),134 None, None, None,135 200, 150, (1, 1))136 self._expect(240, 320, (1, 1),137 200, 200, (1, 1),138 None, None, None,139 150, 200, (1, 1))140 self._expect(320, 240, (3, 4),141 200, 300, (1, 2),142 None, None, None,143 200, 100, (1, 2))144 self._expect(320, 240, (3, 4),145 200, 300, (1, 3),146 None, None, None,147 200, 67, (1, 3))148 self._expect(320, 240, (3, 4),149 200, 300, (2, 1),150 None, None, None,151 150, 300, (2, 1))152 self._expect(320, 240, (3, 4),153 200, 300, (3, 1),154 None, None, None,155 100, 300, (3, 1))156 self._expect(240, 320, (3, 4),157 200, 300, (1, 2),158 None, None, None,159 200, 178, (1, 2))160 self._expect(240, 320, (3, 4),161 200, 300, (1, 3),162 None, None, None,163 200, 119, (1, 3))164 self._expect(240, 320, (3, 4),165 200, 300, (2, 1),166 None, None, None,167 84, 300, (2, 1))168 self._expect(240, 320, (3, 4),169 200, 300, (3, 1),170 None, None, None,171 56, 300, (3, 1))172 self._expect(900, 900, (7, 8),173 800, 200, (1, 1),174 None, None, None,175 175, 200, (1, 1))176 self._expect(900, 900, (7, 8),177 200, 800, (1, 1),178 None, None, None,179 200, 229, (1, 1))180 def testPreferredHeight(self):181 self._expect(320, 240, (1, 1),182 None, 200, (1, 1),183 None, None, None,184 267, 200, (1, 1))185 self._expect(240, 320, (1, 1),186 None, 200, (1, 1),187 None, None, None,188 150, 200, (1, 1))189 self._expect(320, 240, (8, 7),190 None, 200, None,191 None, None, None,192 267, 200, (8, 7))193 self._expect(240, 320, (8, 7),194 None, 200, None,195 None, None, None,196 150, 200, (8, 7))197 self._expect(320, 240, (3, 4),198 None, 200, (1, 1),199 None, None, None,200 200, 200, (1, 1))201 self._expect(320, 240, (3, 4),202 None, 200, (1, 2),203 None, None, None,204 400, 200, (1, 2))205 self._expect(320, 240, (3, 4),206 None, 200, (1, 3),207 None, None, None,208 600, 200, (1, 3))209 self._expect(320, 240, (3, 4),210 None, 200, (2, 1),211 None, None, None,212 100, 200, (2, 1))213 self._expect(320, 240, (3, 4),214 None, 200, (3, 1),215 None, None, None,216 67, 200, (3, 1))217 self._expect(240, 320, (3, 4),218 None, 200, (1, 1),219 None, None, None,220 113, 200, (1, 1))221 self._expect(240, 320, (3, 4),222 None, 200, (1, 2),223 None, None, None,224 225, 200, (1, 2))225 self._expect(240, 320, (3, 4),226 None, 200, (1, 3),227 None, None, None,228 338, 200, (1, 3))229 self._expect(240, 320, (3, 4),230 None, 200, (2, 1),231 None, None, None,232 56, 200, (2, 1))233 self._expect(240, 320, (3, 4),234 None, 200, (3, 1),235 None, None, None,236 38, 200, (3, 1))237 def testPreferredWidth(self):238 self._expect(320, 240, (1, 1),239 200, None, (1, 1),240 None, None, None,241 200, 150, (1, 1))242 self._expect(240, 320, (1, 1),243 200, None, (1, 1),244 None, None, None,245 200, 267, (1, 1))246 self._expect(320, 240, (8, 7),247 200, None, None,248 None, None, None,249 200, 150, (8, 7))250 self._expect(240, 320, (8, 7),251 200, None, None,252 None, None, None,253 200, 267, (8, 7))254 self._expect(320, 240, (4, 3),255 200, None, (1, 1),256 None, None, None,257 200, 113, (1, 1))258 self._expect(320, 240, (4, 3),259 200, None, (1, 2),260 None, None, None,261 200, 56, (1, 2))262 self._expect(320, 240, (4, 3),263 200, None, (1, 3),264 None, None, None,265 200, 38, (1, 3))266 self._expect(320, 240, (4, 3),267 200, None, (2, 1),268 None, None, None,269 200, 225, (2, 1))270 self._expect(320, 240, (4, 3),271 200, None, (3, 1),272 None, None, None,273 200, 338, (3, 1))274 self._expect(240, 320, (4, 3),275 200, None, (1, 1),276 None, None, None,277 200, 200, (1, 1))278 self._expect(240, 320, (4, 3),279 200, None, (1, 2),280 None, None, None,281 200, 100, (1, 2))282 self._expect(240, 320, (4, 3),283 200, None, (1, 3),284 None, None, None,285 200, 67, (1, 3))286 self._expect(240, 320, (4, 3),287 200, None, (2, 1),288 None, None, None,289 200, 400, (2, 1))290 self._expect(240, 320, (4, 3),291 200, None, (3, 1),292 None, None, None,293 200, 600, (3, 1))294 def testDefaultPreferredMethod(self):295 #The default preferred method is "height"296 self._expect(320, 240, (1, 1),297 None, None, (1, 2),298 None, None, None,299 640, 240, (1, 2))300 self._expect(240, 320, (1, 1),301 None, None, (1, 2),302 None, None, None,303 480, 320, (1, 2))304 self._expect(320, 240, (1, 1),305 None, None, (2, 1),306 None, None, None,307 160, 240, (2, 1))308 self._expect(240, 320, (1, 1),309 None, None, (2, 1),310 None, None, None,311 120, 320, (2, 1))312 def testPreserveHeightMethod(self):313 self._expect(320, 240, (1, 1),314 None, None, (1, 2),315 None, None, VideoScaleMethodEnum.height,316 640, 240, (1, 2))317 self._expect(320, 240, (1, 1),318 None, None, (1, 3),319 None, None, VideoScaleMethodEnum.height,320 960, 240, (1, 3))321 self._expect(320, 240, (1, 1),322 None, None, (2, 1),323 None, None, VideoScaleMethodEnum.height,324 160, 240, (2, 1))325 self._expect(320, 240, (1, 1),326 None, None, (3, 1),327 None, None, VideoScaleMethodEnum.height,328 107, 240, (3, 1))329 self._expect(240, 320, (1, 1),330 None, None, (1, 2),331 None, None, VideoScaleMethodEnum.height,332 480, 320, (1, 2))333 self._expect(240, 320, (1, 1),334 None, None, (1, 3),335 None, None, VideoScaleMethodEnum.height,336 720, 320, (1, 3))337 self._expect(240, 320, (1, 1),338 None, None, (2, 1),339 None, None, VideoScaleMethodEnum.height,340 120, 320, (2, 1))341 self._expect(240, 320, (1, 1),342 None, None, (3, 1),343 None, None, VideoScaleMethodEnum.height,344 80, 320, (3, 1))345 self._expect(320, 240, (5, 4),346 None, None, (1, 2),347 None, None, VideoScaleMethodEnum.height,348 800, 240, (1, 2))349 self._expect(320, 240, (5, 4),350 None, None, (1, 3),351 None, None, VideoScaleMethodEnum.height,352 1200, 240, (1, 3))353 self._expect(320, 240, (5, 4),354 None, None, (2, 1),355 None, None, VideoScaleMethodEnum.height,356 200, 240, (2, 1))357 self._expect(320, 240, (5, 4),358 None, None, (3, 1),359 None, None, VideoScaleMethodEnum.height,360 133, 240, (3, 1))361 self._expect(240, 320, (5, 4),362 None, None, (1, 2),363 None, None, VideoScaleMethodEnum.height,364 600, 320, (1, 2))365 self._expect(240, 320, (5, 4),366 None, None, (1, 3),367 None, None, VideoScaleMethodEnum.height,368 900, 320, (1, 3))369 self._expect(240, 320, (5, 4),370 None, None, (2, 1),371 None, None, VideoScaleMethodEnum.height,372 150, 320, (2, 1))373 self._expect(240, 320, (5, 4),374 None, None, (3, 1),375 None, None, VideoScaleMethodEnum.height,376 100, 320, (3, 1))377 def testPreserveWidthMethod(self):378 self._expect(320, 240, (1, 1),379 None, None, (1, 2),380 None, None, VideoScaleMethodEnum.width,381 320, 120, (1, 2))382 self._expect(320, 240, (1, 1),383 None, None, (1, 3),384 None, None, VideoScaleMethodEnum.width,385 320, 80, (1, 3))386 self._expect(320, 240, (1, 1),387 None, None, (2, 1),388 None, None, VideoScaleMethodEnum.width,389 320, 480, (2, 1))390 self._expect(320, 240, (1, 1),391 None, None, (3, 1),392 None, None, VideoScaleMethodEnum.width,393 320, 720, (3, 1))394 self._expect(240, 320, (1, 1),395 None, None, (1, 2),396 None, None, VideoScaleMethodEnum.width,397 240, 160, (1, 2))398 self._expect(240, 320, (1, 1),399 None, None, (1, 3),400 None, None, VideoScaleMethodEnum.width,401 240, 107, (1, 3))402 self._expect(240, 320, (1, 1),403 None, None, (2, 1),404 None, None, VideoScaleMethodEnum.width,405 240, 640, (2, 1))406 self._expect(240, 320, (1, 1),407 None, None, (3, 1),408 None, None, VideoScaleMethodEnum.width,409 240, 960, (3, 1))410 self._expect(320, 240, (7, 8),411 None, None, (1, 2),412 None, None, VideoScaleMethodEnum.width,413 320, 137, (1, 2))414 self._expect(320, 240, (7, 8),415 None, None, (1, 3),416 None, None, VideoScaleMethodEnum.width,417 320, 91, (1, 3))418 self._expect(320, 240, (7, 8),419 None, None, (2, 1),420 None, None, VideoScaleMethodEnum.width,421 320, 549, (2, 1))422 self._expect(320, 240, (7, 8),423 None, None, (3, 1),424 None, None, VideoScaleMethodEnum.width,425 320, 823, (3, 1))426 self._expect(240, 320, (7, 8),427 None, None, (1, 2),428 None, None, VideoScaleMethodEnum.width,429 240, 183, (1, 2))430 self._expect(240, 320, (7, 8),431 None, None, (1, 3),432 None, None, VideoScaleMethodEnum.width,433 240, 122, (1, 3))434 self._expect(240, 320, (7, 8),435 None, None, (2, 1),436 None, None, VideoScaleMethodEnum.width,437 240, 731, (2, 1))438 self._expect(240, 320, (7, 8),439 None, None, (3, 1),440 None, None, VideoScaleMethodEnum.width,441 240, 1097, (3, 1))442 def testUpscaleMethod(self):443 self._expect(320, 240, (1, 1),444 None, None, (1, 2),445 None, None, VideoScaleMethodEnum.upscale,446 640, 240, (1, 2))447 self._expect(320, 240, (1, 1),448 None, None, (1, 3),449 None, None, VideoScaleMethodEnum.upscale,450 960, 240, (1, 3))451 self._expect(320, 240, (1, 1),452 None, None, (2, 1),453 None, None, VideoScaleMethodEnum.upscale,454 320, 480, (2, 1))455 self._expect(320, 240, (1, 1),456 None, None, (3, 1),457 None, None, VideoScaleMethodEnum.upscale,458 320, 720, (3, 1))459 self._expect(240, 320, (1, 1),460 None, None, (1, 2),461 None, None, VideoScaleMethodEnum.upscale,462 480, 320, (1, 2))463 self._expect(240, 320, (1, 1),464 None, None, (1, 3),465 None, None, VideoScaleMethodEnum.upscale,466 720, 320, (1, 3))467 self._expect(240, 320, (1, 1),468 None, None, (2, 1),469 None, None, VideoScaleMethodEnum.upscale,470 240, 640, (2, 1))471 self._expect(240, 320, (1, 1),472 None, None, (3, 1),473 None, None, VideoScaleMethodEnum.upscale,474 240, 960, (3, 1))475 self._expect(320, 240, (2, 3),476 None, None, (1, 2),477 None, None, VideoScaleMethodEnum.upscale,478 427, 240, (1, 2))479 self._expect(320, 240, (2, 3),480 None, None, (1, 3),481 None, None, VideoScaleMethodEnum.upscale,482 640, 240, (1, 3))483 self._expect(320, 240, (2, 3),484 None, None, (2, 1),485 None, None, VideoScaleMethodEnum.upscale,486 320, 720, (2, 1))487 self._expect(320, 240, (2, 3),488 None, None, (3, 1),489 None, None, VideoScaleMethodEnum.upscale,490 320, 1080, (3, 1))491 self._expect(240, 320, (2, 3),492 None, None, (1, 2),493 None, None, VideoScaleMethodEnum.upscale,494 320, 320, (1, 2))495 self._expect(240, 320, (2, 3),496 None, None, (1, 3),497 None, None, VideoScaleMethodEnum.upscale,498 480, 320, (1, 3))499 self._expect(240, 320, (2, 3),500 None, None, (2, 1),501 None, None, VideoScaleMethodEnum.upscale,502 240, 960, (2, 1))503 self._expect(240, 320, (2, 3),504 None, None, (3, 1),505 None, None, VideoScaleMethodEnum.upscale,506 240, 1440, (3, 1))507 def testDownscaleMethod(self):508 self._expect(320, 240, (1, 1),509 None, None, (1, 2),510 None, None, VideoScaleMethodEnum.downscale,511 320, 120, (1, 2))512 self._expect(320, 240, (1, 1),513 None, None, (1, 3),514 None, None, VideoScaleMethodEnum.downscale,515 320, 80, (1, 3))516 self._expect(320, 240, (1, 1),517 None, None, (2, 1),518 None, None, VideoScaleMethodEnum.downscale,519 160, 240, (2, 1))520 self._expect(320, 240, (1, 1),521 None, None, (3, 1),522 None, None, VideoScaleMethodEnum.downscale,523 107, 240, (3, 1))524 self._expect(240, 320, (1, 1),525 None, None, (1, 2),526 None, None, VideoScaleMethodEnum.downscale,527 240, 160, (1, 2))528 self._expect(240, 320, (1, 1),529 None, None, (1, 3),530 None, None, VideoScaleMethodEnum.downscale,531 240, 107, (1, 3))532 self._expect(240, 320, (1, 1),533 None, None, (2, 1),534 None, None, VideoScaleMethodEnum.downscale,535 120, 320, (2, 1))536 self._expect(240, 320, (1, 1),537 None, None, (3, 1),538 None, None, VideoScaleMethodEnum.downscale,539 80, 320, (3, 1))540 self._expect(320, 240, (1, 4),541 None, None, (1, 2),542 None, None, VideoScaleMethodEnum.downscale,543 160, 240, (1, 2))544 self._expect(320, 240, (1, 4),545 None, None, (1, 3),546 None, None, VideoScaleMethodEnum.downscale,547 240, 240, (1, 3))548 self._expect(320, 240, (1, 4),549 None, None, (2, 1),550 None, None, VideoScaleMethodEnum.downscale,551 40, 240, (2, 1))552 self._expect(320, 240, (1, 4),553 None, None, (3, 1),554 None, None, VideoScaleMethodEnum.downscale,555 27, 240, (3, 1))556 self._expect(240, 320, (1, 4),557 None, None, (1, 2),558 None, None, VideoScaleMethodEnum.downscale,559 120, 320, (1, 2))560 self._expect(240, 320, (1, 4),561 None, None, (1, 3),562 None, None, VideoScaleMethodEnum.downscale,563 180, 320, (1, 3))564 self._expect(240, 320, (1, 4),565 None, None, (2, 1),566 None, None, VideoScaleMethodEnum.downscale,567 30, 320, (2, 1))568 self._expect(240, 320, (1, 4),569 None, None, (3, 1),570 None, None, VideoScaleMethodEnum.downscale,571 20, 320, (3, 1))572 def testWithoutPreferredSizeAndPAR(self):573 self._expect(320, 240, (1, 1),574 None, None, None,575 None, None, None,576 320, 240, (1, 1))577 self._expect(320, 240, (2, 1),578 None, None, None,579 None, None, None,580 320, 240, (2, 1))581 self._expect(320, 240, (3, 1),582 None, None, None,583 None, None, None,584 320, 240, (3, 1))585 self._expect(320, 240, (1, 2),586 None, None, None,587 None, None, None,588 320, 240, (1, 2))589 self._expect(320, 240, (1, 3),590 None, None, None,591 None, None, None,592 320, 240, (1, 3))593 self._expect(320, 100, (1, 1),594 None, None, None,595 None, None, None,596 320, 100, (1, 1))597 self._expect(100, 240, (1, 1),598 None, None, None,599 None, None, None,...
Playwright tutorial
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Chapters:
- What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
- How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
- Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
- What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
- Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
- Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
- Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
- Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
- Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
- Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
- Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
- Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.
Run Playwright Python 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?
