status-mobile/test/appium/support/appium_container.py

112 lines
4.8 KiB
Python
Raw Normal View History

import datetime
import logging
import re
import subprocess
import time
import docker
from docker.errors import NotFound
DOCKER_SHARED_VOLUME_PATH = '/root/shared_volume'
class DeviceStats(object):
def __init__(self):
self.total_battery_capacity_mah = None
# Est power used (mAh) by the app
self.estimated_power_usage_mah = None
class AppiumContainer:
def start_appium_container(self, shared_volume):
docker_client = docker.from_env()
try:
self.container = docker_client.containers.get("appium")
self.container.restart()
except NotFound:
self.container = docker_client.containers.run("appium/appium:local", detach=True, name="appium",
ports={'4723/tcp': 4723},
volumes={shared_volume: {
'bind': DOCKER_SHARED_VOLUME_PATH, 'mode': 'rw'}})
logging.info("Running Appium container. ID: %s" % self.container.short_id)
def exec_run(self, cmd):
return self.container.exec_run(cmd, stdout=True, stderr=True, stdin=True)
def connect_device(self, device_ip):
device_address = device_ip + ':5555'
connected_state = "connected to %s" % device_address
cmd = self.exec_run(['adb', 'connect', device_address])
if connected_state in cmd.output.decode("utf-8"):
logging.info("adb is already connected with the device")
else:
logging.info("Connecting the device with adb..")
# Restart adb on host machine
subprocess.call(['adb', 'kill-server'])
input("Connect USB cable to the device and press ENTER..")
# Reset USB on host machine
subprocess.call(['adb', 'usb'])
time.sleep(5) # wait until adb usb is restarted
# Set the target device to listen for a TCP/IP connection on port 5555 on host machine
subprocess.call(['adb', 'tcpip', '5555'])
# restarting in TCP mode port: 5555
input("Now, disconnect the USB cable and press ENTER..")
# Connect to the device in docker container
self.exec_run(['adb', 'connect', device_address])
input("Please check your device and allow for USB debugging if necessary. Then press ENTER to continue "
"the test..\n")
def reset_battery_stats(self):
logging.info("Resetting device stats..")
self.exec_run(['adb', 'shell', 'dumpsys', 'batterystats', '--reset'])
def generate_bugreport(self, report_name):
now = datetime.datetime.now()
bugreport_name = "bugreport_%s_%s" % (report_name, now.strftime("%Y-%m-%d_%H-%M-%S"))
print("\nGenerating device report from the test..")
self.exec_run(['adb', 'bugreport', "/%s/%s" % (DOCKER_SHARED_VOLUME_PATH, bugreport_name)])
print("Device report saved in the shared volume as %s.zip" % bugreport_name)
def get_device_stats(self):
stats = DeviceStats()
# Find process uid
uid_line = self.exec_run(['adb', 'shell', 'ps', '|', 'grep', 'im.status.ethereum']).output \
.decode('utf-8')
# Match first word which is the uid
match = re.match(r'(?:^|(?:[.!?]\s))(\w+)', uid_line, re.M | re.I)
uid = match.group(1).replace('_', '')
# Battery stats
batterystats = self.exec_run(['adb', 'shell', 'dumpsys', 'batterystats', 'im.status.ethereum']).output \
.decode('utf-8').splitlines()
battery_usage_line = [s for s in batterystats if "Uid %s" % uid in s][0]
match = re.match(r'.* Uid %s: ([^\s]+)' % uid, battery_usage_line, re.M | re.I)
stats.estimated_power_usage_mah = float(match.group(1))
capacity_line = [s for s in batterystats if "Capacity" in s][0]
match = re.match(r'.* Capacity: (.*), Computed drain: (.*), .*', capacity_line, re.M | re.I)
stats.total_battery_capacity_mah = float(match.group(1))
stats.total_computed_drain_mah = float(match.group(2))
# Wi-Fi stats
wifi_stats = self.exec_run(['adb', 'shell', 'dumpsys', 'batterystats', 'im.status.ethereum', '|', 'grep',
2023-09-06 03:07:12 +00:00
'Wi-Fi total']).output.decode('utf-8')
stats.wifi_received = wifi_stats.split()[3].replace(',', '')
stats.wifi_sent = wifi_stats.split()[5]
# OS stats
stats.os_version = self.exec_run(['adb', 'shell', 'getprop', 'ro.build.version.release'])\
.output.decode('utf-8').rstrip("\n")
stats.device_model = self.exec_run(['adb', 'shell', 'getprop', 'ro.product.model'])\
.output.decode('utf-8').rstrip("\n")
return stats
def stop_container(self):
self.container.stop()