mirror of
https://github.com/budtmo/docker-android.git
synced 2026-07-31 04:07:25 +00:00
Big restructuring by moving to python
This commit is contained in:
budtmo
+213
@@ -0,0 +1,213 @@
|
||||
#!/usr/bin/env python3
|
||||
import subprocess
|
||||
from typing import Union
|
||||
|
||||
import click
|
||||
import logging
|
||||
import os
|
||||
|
||||
from enum import Enum
|
||||
|
||||
from src.application import Application
|
||||
from src.device import DeviceType
|
||||
from src.device.emulator import Emulator
|
||||
from src.device.geny_aws import GenyAWS
|
||||
from src.device.geny_saas import GenySAAS
|
||||
from src.helper import convert_str_to_bool, get_env_value_or_raise
|
||||
from src.constants import ENV
|
||||
from src.logger import log
|
||||
|
||||
log.init()
|
||||
logger = logging.getLogger("App")
|
||||
|
||||
|
||||
def get_device(given_input: str) -> Union[Emulator, GenyAWS, GenySAAS, None]:
|
||||
"""
|
||||
Get Device object based on given input
|
||||
|
||||
:param given_input: device in string
|
||||
:return: Platform object
|
||||
"""
|
||||
|
||||
input_lower = given_input.lower()
|
||||
|
||||
if input_lower == DeviceType.EMULATOR.value.lower():
|
||||
emu_av = get_env_value_or_raise(ENV.EMULATOR_ANDROID_VERSION)
|
||||
emu_img_type = get_env_value_or_raise(ENV.EMULATOR_IMG_TYPE)
|
||||
emu_sys_img = get_env_value_or_raise(ENV.EMULATOR_SYS_IMG)
|
||||
|
||||
emu_device = os.getenv(ENV.EMULATOR_DEVICE, "Nexus 5")
|
||||
emu_data_partition = os.getenv(ENV.EMULATOR_DATA_PARTITION, "550m")
|
||||
emu_additional_args = os.getenv(ENV.EMULATOR_ADDITIONAL_ARGS, "")
|
||||
|
||||
emu_name = os.getenv(ENV.EMULATOR_NAME, "{d}_{v}".format(
|
||||
d=emu_device.replace(" ", "_").lower(), v=emu_av))
|
||||
emu = Emulator(emu_name, emu_device, emu_av, emu_data_partition,
|
||||
emu_additional_args, emu_img_type, emu_sys_img)
|
||||
return emu
|
||||
elif input_lower == DeviceType.GENY_AWS.value.lower():
|
||||
return GenyAWS()
|
||||
elif input_lower == DeviceType.GENY_SAAS.value.lower():
|
||||
return GenySAAS()
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
@click.group(context_settings=dict(help_option_names=['-h', '--help']))
|
||||
def cli():
|
||||
pass
|
||||
|
||||
|
||||
def start_appium() -> None:
|
||||
if convert_str_to_bool(os.getenv(ENV.APPIUM)):
|
||||
cmd = f"/usr/bin/appium"
|
||||
app_appium = Application("Appium", cmd,
|
||||
os.getenv(ENV.APPIUM_ADDITIONAL_ARGS, ""), False)
|
||||
app_appium.start()
|
||||
else:
|
||||
logger.info("env APPIUM cannot be found, Appium is not started!")
|
||||
|
||||
|
||||
def start_device() -> None:
|
||||
given_pt = get_env_value_or_raise(ENV.DEVICE_TYPE)
|
||||
selected_device = get_device(given_pt)
|
||||
if selected_device is None:
|
||||
raise RuntimeError(f"'{given_pt}' is invalid! Please check again!")
|
||||
selected_device.create()
|
||||
selected_device.start()
|
||||
selected_device.wait_until_ready()
|
||||
selected_device.reconfigure()
|
||||
selected_device.keep_alive()
|
||||
|
||||
|
||||
def start_display_screen() -> None:
|
||||
cmd = "/usr/bin/Xvfb"
|
||||
args = f"{os.getenv(ENV.DISPLAY)} " \
|
||||
f"-screen {os.getenv(ENV.SCREEN_NUMBER)} " \
|
||||
f"{os.getenv(ENV.SCREEN_WIDTH)}x" \
|
||||
f"{os.getenv(ENV.SCREEN_HEIGHT)}x" \
|
||||
f"{os.getenv(ENV.SCREEN_DEPTH)}"
|
||||
d_screen = Application("d_screen", cmd, args, False)
|
||||
d_screen.start()
|
||||
|
||||
|
||||
def start_display_wm() -> None:
|
||||
cmd = "/usr/bin/openbox-session"
|
||||
d_wm = Application("d_wm", cmd)
|
||||
d_wm.start()
|
||||
|
||||
|
||||
def start_port_forwarder() -> None:
|
||||
import socket
|
||||
local_ip = socket.gethostbyname(socket.gethostname())
|
||||
cmd = f"/usr/bin/socat tcp-listen:5554,bind={local_ip},fork tcp:127.0.0.1:5554 & " \
|
||||
f"/usr/bin/socat tcp-listen:5555,bind={local_ip},fork tcp:127.0.0.1:5555"
|
||||
pf = Application("port_forwarder", cmd)
|
||||
pf.start()
|
||||
|
||||
|
||||
def start_vnc_server() -> None:
|
||||
cmd = "/usr/bin/x11vnc"
|
||||
vnc_pass = os.getenv(ENV.VNC_PASSWORD)
|
||||
if vnc_pass:
|
||||
pass_path = os.path.join(os.getenv(ENV.WORK_PATH), ".vncpass")
|
||||
subprocess.check_call(f"{cmd} -storepasswd {vnc_pass} {pass_path}", shell=True)
|
||||
last_arg = f"-rfbauth {pass_path}"
|
||||
else:
|
||||
last_arg = "-nopw"
|
||||
|
||||
display = os.getenv(ENV.DISPLAY)
|
||||
args = f"-display {display} -forever -shared {last_arg}"
|
||||
vnc_server = Application("vnc_web", cmd, args, False)
|
||||
vnc_server.start()
|
||||
|
||||
|
||||
def start_vnc_web() -> None:
|
||||
if convert_str_to_bool(os.getenv(ENV.WEB_VNC)):
|
||||
vnc_port = get_env_value_or_raise(ENV.VNC_PORT)
|
||||
vnc_web_port = get_env_value_or_raise(ENV.WEB_VNC_PORT)
|
||||
cmd = "/opt/noVNC/utils/novnc_proxy"
|
||||
args = f"--vnc localhost:{vnc_port} localhost:{vnc_web_port}"
|
||||
vnc_web = Application("vnc_web", cmd, args, False)
|
||||
vnc_web.start()
|
||||
else:
|
||||
logger.info("env WEB_VNC cannot be found, VNC_WEB is not started!")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("app", type=click.Choice([app.value for app in Application.App]))
|
||||
def start(app):
|
||||
selected_app = str(app).lower()
|
||||
if selected_app == Application.App.APPIUM.value.lower():
|
||||
start_appium()
|
||||
elif selected_app == Application.App.DEVICE.value.lower():
|
||||
start_device()
|
||||
elif selected_app == Application.App.DISPLAY_SCREEN.value.lower():
|
||||
start_display_screen()
|
||||
elif selected_app == Application.App.DISPLAY_WM.value.lower():
|
||||
start_display_wm()
|
||||
elif selected_app == Application.App.PORT_FORWARDER.value.lower():
|
||||
start_port_forwarder()
|
||||
elif selected_app == Application.App.VNC_SERVER.value.lower():
|
||||
start_vnc_server()
|
||||
elif selected_app == Application.App.VNC_WEB.value.lower():
|
||||
start_vnc_web()
|
||||
else:
|
||||
logger.error(f"application '{selected_app}' is not supported!")
|
||||
|
||||
|
||||
class SharedComponent(Enum):
|
||||
LOG = "log"
|
||||
|
||||
|
||||
def shared_log() -> None:
|
||||
if convert_str_to_bool(os.getenv(ENV.WEB_LOG)):
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
|
||||
log_path = get_env_value_or_raise(ENV.LOG_PATH)
|
||||
log_port = int(get_env_value_or_raise(ENV.WEB_LOG_PORT))
|
||||
logger.info(f"Shared log is enabled! all logs can be found on port '{log_port}'")
|
||||
|
||||
class LogSharedHandler(BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
# root path
|
||||
if self.path == "/":
|
||||
html = "<html><body>"
|
||||
for f in os.listdir(log_path):
|
||||
html += f"<p><a href=\"{f}\">{f}</a></p>"
|
||||
html += "</body></html>"
|
||||
|
||||
self.send_response(200)
|
||||
self.send_header("Content-type", "text/html")
|
||||
self.end_headers()
|
||||
self.wfile.write(html.encode())
|
||||
# open each selected log file
|
||||
else:
|
||||
p = log_path + self.path
|
||||
try:
|
||||
with open(p, "rb") as file:
|
||||
self.send_response(200)
|
||||
self.send_header("Content-type", "text/plain")
|
||||
self.end_headers()
|
||||
self.wfile.write(file.read())
|
||||
except FileNotFoundError:
|
||||
self.send_error(404, "File not found")
|
||||
|
||||
httpd = HTTPServer(('0.0.0.0', log_port), LogSharedHandler)
|
||||
httpd.serve_forever()
|
||||
else:
|
||||
logger.info(f"Shared log is disabled! nothing to do!")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("component", type=click.Choice([component.value for component in SharedComponent]))
|
||||
def share(component):
|
||||
selected_component = str(component).lower()
|
||||
if selected_component == SharedComponent.LOG.value.lower():
|
||||
shared_log()
|
||||
else:
|
||||
logger.error(f"component '{component}' is not supported!")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
cli()
|
||||
@@ -0,0 +1,35 @@
|
||||
import logging
|
||||
import subprocess
|
||||
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class Application:
|
||||
class App(Enum):
|
||||
APPIUM = "appium"
|
||||
DEVICE = "device"
|
||||
DISPLAY_SCREEN = "display_screen"
|
||||
DISPLAY_WM = "display_wm"
|
||||
PORT_FORWARDER = "port_forwarder"
|
||||
VNC_SERVER = "vnc_server"
|
||||
VNC_WEB = "vnc_web"
|
||||
|
||||
def __init__(self, name: str, command: str, additional_args: str = "", ui: bool = False) -> None:
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
self.name = name
|
||||
self.command = command
|
||||
self.additional_args = additional_args
|
||||
self.ui = ui
|
||||
|
||||
def start(self) -> None:
|
||||
if self.ui:
|
||||
self.logger.info(f"{self.name} will be started with ui!")
|
||||
subprocess.check_call(f"/usr/bin/xterm -T {self.name} -n {self.name} "
|
||||
f"-e '{self.command} {self.additional_args}'", shell=True)
|
||||
else:
|
||||
self.logger.info(f"{self.name} will be started without ui!")
|
||||
subprocess.check_call(f"{self.command} {self.additional_args}", shell=True)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "Application(name={n}, command={c}, args={args}, ui={ui})".format(
|
||||
n=self.name, c=self.command, args=self.additional_args, ui=self.ui)
|
||||
@@ -0,0 +1,6 @@
|
||||
# Status
|
||||
STATUS_CREATING = "CREATING"
|
||||
STATUS_STARTING = "STARTING"
|
||||
STATUS_BOOTING = "BOOTING"
|
||||
STATUS_RECONFIGURING = "RECONFIGURING"
|
||||
STATUS_READY = "READY"
|
||||
@@ -0,0 +1,45 @@
|
||||
# General
|
||||
DOCKER_ANDROID_VERSION = "DOCKER_ANDROID_VERSION"
|
||||
USER_BEHAVIOR_ANALYTICS = "USER_BEHAVIOR_ANALYTICS"
|
||||
APPIUM = "APPIUM"
|
||||
APPIUM_ADDITIONAL_ARGS = "APPIUM_ADDITIONAL_ARGS"
|
||||
DISPLAY = "DISPLAY"
|
||||
SCREEN_DEPTH = "SCREEN_DEPTH"
|
||||
SCREEN_HEIGHT = "SCREEN_HEIGHT"
|
||||
SCREEN_NUMBER = "SCREEN_NUMBER"
|
||||
SCREEN_WIDTH = "SCREEN_WIDTH"
|
||||
VNC_PASSWORD = "VNC_PASSWORD"
|
||||
VNC_PORT = "VNC_PORT"
|
||||
WEB_VNC_PORT = "WEB_VNC_PORT"
|
||||
WEB_VNC = "WEB_VNC"
|
||||
WORK_PATH = "WORK_PATH"
|
||||
LOG_PATH = "LOG_PATH"
|
||||
WEB_LOG_PORT = "WEB_LOG_PORT"
|
||||
WEB_LOG = "WEB_LOG"
|
||||
|
||||
# Device
|
||||
DEVICE_INTERVAL_WAITING = "DEVICE_INTERVAL_WAITING"
|
||||
DEVICE_TYPE = "DEVICE_TYPE"
|
||||
|
||||
# Device (Emulator)
|
||||
EMULATOR_ADDITIONAL_ARGS = "EMULATOR_ADDITIONAL_ARGS"
|
||||
EMULATOR_ANDROID_VERSION = "EMULATOR_ANDROID_VERSION"
|
||||
EMULATOR_DATA_PARTITION = "EMULATOR_DATA_PARTITION"
|
||||
EMULATOR_DEVICE = "EMULATOR_DEVICE"
|
||||
EMULATOR_IMG_TYPE = "EMULATOR_IMG_TYPE"
|
||||
EMULATOR_NAME = "EMULATOR_NAME"
|
||||
EMULATOR_NO_SKIN = "EMULATOR_NO_SKIN"
|
||||
EMULATOR_SYS_IMG = "EMULATOR_SYS_IMG"
|
||||
|
||||
# Device (Genymotion - General)
|
||||
GENYMOTION_TEMPLATE_PATH = "GENYMOTION_TEMPLATE_PATH"
|
||||
|
||||
# Device (Geny_SAAS)
|
||||
GENY_SAAS_USER = "GENY_SAAS_USER"
|
||||
GENY_SAAS_PASS = "GENY_SAAS_PASS"
|
||||
GENY_SAAS_TEMPLATE_FILE_NAME = "saas.json"
|
||||
|
||||
# Device (Geny_AWS)
|
||||
AWS_ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID"
|
||||
AWS_SECRET_ACCESS_KEY = "AWS_SECRET_ACCESS_KEY"
|
||||
GENY_AWS_TEMPLATE_FILE_NAME = "aws.json"
|
||||
@@ -0,0 +1 @@
|
||||
UTF8 = "utf-8"
|
||||
@@ -0,0 +1,172 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import platform
|
||||
import requests
|
||||
import signal
|
||||
import time
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from enum import Enum
|
||||
|
||||
from src.helper import convert_str_to_bool, get_env_value_or_raise
|
||||
from src.constants import DEVICE, ENV
|
||||
|
||||
|
||||
class DeviceType(Enum):
|
||||
EMULATOR = "emulator"
|
||||
GENY_SAAS = "geny_saas"
|
||||
GENY_AWS = "geny_aws"
|
||||
|
||||
|
||||
class Device(ABC):
|
||||
FORM_ID = "1FAIpQLSdrKWQdMh6Nt8v8NQdYvTIntohebAgqWCpXT3T9NofAoxcpkw"
|
||||
FORM_USER = "user"
|
||||
FORM_CITY = "city"
|
||||
FORM_REGION = "region"
|
||||
FORM_COUNTRY = "country"
|
||||
FORM_APP_VERSION = "app_version"
|
||||
FORM_APPIUM = "appium"
|
||||
FORM_APPIUM_ADDITIONAL_ARGS = "appium_additional_args"
|
||||
FORM_WEB_LOG = "web_log"
|
||||
FORM_WEB_VNC = "web_vnc"
|
||||
FORM_SCREEN_RESOLUTION = "screen_resolution"
|
||||
FORM_DEVICE_TYPE = "device_type"
|
||||
FORM_EMU_DEVICE = "emu_device"
|
||||
FORM_EMU_ANDROID_VERSION = "emu_android_version"
|
||||
FORM_EMU_NO_SKIN = "emu_no_skin"
|
||||
FORM_EMU_DATA_PARTITION = "emu_data_partition"
|
||||
FORM_EMU_ADDITIONAL_ARGS = "emu_additional_args"
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
self.device_type = None
|
||||
self.interval_waiting = int(os.getenv(ENV.DEVICE_INTERVAL_WAITING, 2))
|
||||
self.user_behavior_analytics = convert_str_to_bool(os.getenv(ENV.USER_BEHAVIOR_ANALYTICS, "true"))
|
||||
self.form_field = {
|
||||
Device.FORM_USER: "entry.108751316",
|
||||
Device.FORM_CITY: "entry.2083022547",
|
||||
Device.FORM_REGION: "entry.1083141079",
|
||||
Device.FORM_COUNTRY: "entry.1946159560",
|
||||
Device.FORM_APP_VERSION: "entry.818050927",
|
||||
Device.FORM_APPIUM: "entry.181610571",
|
||||
Device.FORM_APPIUM_ADDITIONAL_ARGS: "entry.727759656",
|
||||
Device.FORM_WEB_LOG: "entry.1225589007",
|
||||
Device.FORM_WEB_VNC: "entry.2055392048",
|
||||
Device.FORM_SCREEN_RESOLUTION: "entry.709976626",
|
||||
Device.FORM_DEVICE_TYPE: "entry.207096546",
|
||||
Device.FORM_EMU_DEVICE: "entry.1960740382",
|
||||
Device.FORM_EMU_ANDROID_VERSION: "entry.671872491",
|
||||
Device.FORM_EMU_NO_SKIN: "entry.403556951",
|
||||
Device.FORM_EMU_DATA_PARTITION: "entry.1052258875",
|
||||
Device.FORM_EMU_ADDITIONAL_ARGS: "entry.57529972"
|
||||
}
|
||||
self.form_data = {}
|
||||
signal.signal(signal.SIGTERM, self.tear_down)
|
||||
|
||||
def set_status(self, current_status) -> None:
|
||||
bashrc_file = f"{os.getenv(ENV.WORK_PATH)}/device_status"
|
||||
with open(bashrc_file, "w+") as bf:
|
||||
bf.write(current_status)
|
||||
# It won't work using docker exec
|
||||
# os.environ[constants.ENV_DEVICE_STATUS] = current_status
|
||||
|
||||
def _prepare_analytics_payload(self) -> None:
|
||||
self.form_data.update({
|
||||
self.form_field[Device.FORM_USER]: f"{platform.platform()}_{platform.version().replace(' ', '_')}",
|
||||
self.form_field[Device.FORM_APP_VERSION]: os.getenv(ENV.DOCKER_ANDROID_VERSION),
|
||||
self.form_field[Device.FORM_DEVICE_TYPE]: self.device_type,
|
||||
self.form_field[Device.FORM_WEB_VNC]: convert_str_to_bool(os.getenv(ENV.WEB_VNC)),
|
||||
self.form_field[Device.FORM_WEB_LOG]: convert_str_to_bool(os.getenv(ENV.WEB_LOG)),
|
||||
self.form_field[Device.FORM_APPIUM]: convert_str_to_bool(os.getenv(ENV.APPIUM))
|
||||
})
|
||||
|
||||
try:
|
||||
res = requests.get("https://ipinfo.io")
|
||||
if res.ok:
|
||||
json_res = res.json()
|
||||
self.form_data.update({
|
||||
self.form_field[Device.FORM_CITY]: json_res[Device.FORM_CITY],
|
||||
self.form_field[Device.FORM_REGION]: json_res[Device.FORM_REGION],
|
||||
self.form_field[Device.FORM_COUNTRY]: json_res[Device.FORM_COUNTRY]
|
||||
})
|
||||
except requests.exceptions.RequestException as rer:
|
||||
self.logger.warning(rer)
|
||||
pass
|
||||
except KeyError as ke:
|
||||
self.logger.warning(ke)
|
||||
pass
|
||||
|
||||
def create(self) -> None:
|
||||
if self.user_behavior_analytics:
|
||||
self.logger.info("Sending user behavior analytics to improve the tool")
|
||||
try:
|
||||
form_url = f"https://docs.google.com/forms/d/e/{Device.FORM_ID}/formResponse"
|
||||
self._prepare_analytics_payload()
|
||||
requests.post(url=form_url, data=self.form_data)
|
||||
except requests.exceptions.RequestException as rer:
|
||||
self.logger.warning(rer)
|
||||
pass
|
||||
self.set_status(DEVICE.STATUS_CREATING)
|
||||
|
||||
def start(self) -> None:
|
||||
self.set_status(DEVICE.STATUS_STARTING)
|
||||
|
||||
def wait_until_ready(self) -> None:
|
||||
self.set_status(DEVICE.STATUS_BOOTING)
|
||||
|
||||
def reconfigure(self) -> None:
|
||||
self.set_status(DEVICE.STATUS_RECONFIGURING)
|
||||
|
||||
def keep_alive(self) -> None:
|
||||
self.set_status(DEVICE.STATUS_READY)
|
||||
self.logger.warning(f"{self.device_type} process will be kept alive to be able to get sigterm signal...")
|
||||
while True:
|
||||
time.sleep(2)
|
||||
|
||||
@abstractmethod
|
||||
def tear_down(self, *args) -> None:
|
||||
pass
|
||||
|
||||
|
||||
class Genymotion(Device):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
|
||||
def get_data_from_template(self, filename: str) -> dict:
|
||||
path_template_json = os.path.join(get_env_value_or_raise(ENV.GENYMOTION_TEMPLATE_PATH), filename)
|
||||
data = {}
|
||||
if os.path.isfile(path_template_json):
|
||||
try:
|
||||
self.logger.info(path_template_json)
|
||||
with open(path_template_json, "r") as f:
|
||||
data = json.load(f)
|
||||
except FileNotFoundError as fnf:
|
||||
self.shutdown_and_logout()
|
||||
self.logger.error(f"File cannot be found: {fnf}")
|
||||
except json.JSONDecodeError as jde:
|
||||
self.shutdown_and_logout()
|
||||
self.logger.error(f"Error Decoding Json: {jde}")
|
||||
except Exception as e:
|
||||
self.shutdown_and_logout()
|
||||
self.logger.error(e)
|
||||
else:
|
||||
self.shutdown_and_logout()
|
||||
raise RuntimeError(f"'{path_template_json}' cannot be found!")
|
||||
return data
|
||||
|
||||
@abstractmethod
|
||||
def login(self) -> None:
|
||||
pass
|
||||
|
||||
def create(self) -> None:
|
||||
super().create()
|
||||
self.login()
|
||||
|
||||
@abstractmethod
|
||||
def shutdown_and_logout(self) -> None:
|
||||
pass
|
||||
|
||||
def tear_down(self, *args) -> None:
|
||||
self.shutdown_and_logout()
|
||||
@@ -0,0 +1,237 @@
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
from enum import Enum
|
||||
|
||||
from src.device import Device, DeviceType
|
||||
from src.helper import convert_str_to_bool, get_env_value_or_raise, symlink_force
|
||||
from src.constants import ENV, UTF8
|
||||
|
||||
|
||||
class Emulator(Device):
|
||||
DEVICE = (
|
||||
"Nexus 4",
|
||||
"Nexus 5",
|
||||
"Nexus 7",
|
||||
"Nexus One",
|
||||
"Nexus S",
|
||||
"Samsung Galaxy S6",
|
||||
"Samsung Galaxy S7",
|
||||
"Samsung Galaxy S7 Edge",
|
||||
"Samsung Galaxy S8",
|
||||
"Samsung Galaxy S9",
|
||||
"Samsung Galaxy S10"
|
||||
)
|
||||
|
||||
API_LEVEL = {
|
||||
"9.0": "28",
|
||||
"10.0": "29",
|
||||
"11.0": "30",
|
||||
"12.0": "32",
|
||||
"13.0": "33"
|
||||
}
|
||||
|
||||
adb_name_id = 5554
|
||||
|
||||
class ReadinessCheck(Enum):
|
||||
BOOTED = "booted"
|
||||
RUN_STATE = "in running state"
|
||||
WELCOME_SCREEN = "in welcome screen"
|
||||
POP_UP_WINDOW = "pop up window"
|
||||
|
||||
def __init__(self, name: str, device: str, android_version: str, data_partition: str,
|
||||
additional_args: str, img_type: str, sys_img: str) -> None:
|
||||
super().__init__()
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
self.adb_name = f"emulator-{Emulator.adb_name_id}"
|
||||
self.device_type = DeviceType.EMULATOR.value
|
||||
self.name = name
|
||||
if device in self.DEVICE:
|
||||
self.device = device
|
||||
else:
|
||||
raise RuntimeError(f"device '{device}' is not supported!")
|
||||
if android_version in self.API_LEVEL.keys():
|
||||
self.android_version = android_version
|
||||
else:
|
||||
raise RuntimeError(f"android version '{android_version}' is not supported!")
|
||||
self.api_level = self.API_LEVEL[self.android_version]
|
||||
self.data_partition = data_partition
|
||||
self.additional_args = additional_args
|
||||
self.img_type = img_type
|
||||
self.sys_img = sys_img
|
||||
workdir = get_env_value_or_raise(ENV.WORK_PATH)
|
||||
self.path_device_profile_target = os.path.join(workdir, ".android", "devices.xml")
|
||||
self.path_emulator = os.path.join(workdir, "emulator")
|
||||
self.path_emulator_config = os.path.join(workdir, "emulator", "config.ini")
|
||||
self.path_emulator_profiles = os.path.join(workdir, "docker-android", "mixins",
|
||||
"configs", "devices", "profiles")
|
||||
self.path_emulator_skins = os.path.join(workdir, "docker-android", "mixins",
|
||||
"configs", "devices", "skins")
|
||||
self.file_name = self.device.replace(" ", "_").lower()
|
||||
self.no_skin = convert_str_to_bool(os.getenv(ENV.EMULATOR_NO_SKIN))
|
||||
self.interval_after_booting = 15
|
||||
Emulator.adb_name_id += 2
|
||||
self.form_data.update({
|
||||
self.form_field[Device.FORM_SCREEN_RESOLUTION]: f"{os.getenv(ENV.SCREEN_WIDTH)}x"
|
||||
f"{os.getenv(ENV.SCREEN_HEIGHT)}x"
|
||||
f"{os.getenv(ENV.SCREEN_DEPTH)}",
|
||||
self.form_field[Device.FORM_EMU_DEVICE]: self.device,
|
||||
self.form_field[Device.FORM_EMU_ANDROID_VERSION]: self.android_version,
|
||||
self.form_field[Device.FORM_EMU_NO_SKIN]: self.no_skin,
|
||||
self.form_field[Device.FORM_EMU_DATA_PARTITION]: self.data_partition,
|
||||
self.form_field[Device.FORM_EMU_ADDITIONAL_ARGS]: self.additional_args
|
||||
})
|
||||
|
||||
def is_initialized(self) -> bool:
|
||||
import re
|
||||
if os.path.exists(self.path_emulator_config):
|
||||
self.logger.info("Config file exists")
|
||||
with open(self.path_emulator_config, 'r') as f:
|
||||
if any(re.match(r'hw\.device\.name ?= ?{}'.format(self.device), line) for line in f):
|
||||
self.logger.info("Selected device is already created")
|
||||
return True
|
||||
else:
|
||||
self.logger.info("Selected device is not created")
|
||||
return False
|
||||
|
||||
self.logger.info("Config file does not exist")
|
||||
return False
|
||||
|
||||
def _add_profile(self) -> None:
|
||||
if "samsung" in self.device.lower():
|
||||
path_device_profile_source = os.path.join(self.path_emulator_profiles,
|
||||
"{fn}.xml".format(fn=self.file_name))
|
||||
symlink_force(path_device_profile_source, self.path_device_profile_target)
|
||||
self.logger.info("Samsung device profile is linked")
|
||||
|
||||
def _add_skin(self) -> None:
|
||||
device_skin_path = os.path.join(
|
||||
self.path_emulator_skins, "{fn}".format(fn=self.file_name))
|
||||
with open(self.path_emulator_config, "a") as cf:
|
||||
cf.write("hw.keyboard=yes\n")
|
||||
cf.write("disk.dataPartition.size={dp}\n".format(dp=self.data_partition))
|
||||
cf.write("skin.path={sp}\n".format(
|
||||
sp="_no_skin" if self.no_skin else device_skin_path))
|
||||
self.logger.info(f"Skin is added in: '{self.path_emulator_config}'")
|
||||
|
||||
def create(self) -> None:
|
||||
super().create()
|
||||
first_run = not self.is_initialized()
|
||||
if first_run:
|
||||
self.logger.info(f"Creating the {self.device_type}...")
|
||||
self._add_profile()
|
||||
creation_cmd = "avdmanager create avd -f -n {n} -b {it}/{si} " \
|
||||
"-k 'system-images;android-{al};{it};{si}' " \
|
||||
"-d {d} -p {pe}".format(n=self.name, it=self.img_type, si=self.sys_img,
|
||||
al=self.api_level, d=self.device.replace(" ", "\ "),
|
||||
pe=self.path_emulator)
|
||||
self.logger.info(f"Command to create emulator: '{creation_cmd}'")
|
||||
subprocess.check_call(creation_cmd, shell=True)
|
||||
self._add_skin()
|
||||
self.logger.info(f"{self.device_type} is created!")
|
||||
|
||||
def change_permission(self) -> None:
|
||||
kvm_path = "/dev/kvm"
|
||||
if os.path.exists(kvm_path):
|
||||
cmds = (f"sudo chown 1300:1301 {kvm_path}",
|
||||
"sudo sed -i '1d' /etc/passwd")
|
||||
for c in cmds:
|
||||
subprocess.check_call(c, shell=True)
|
||||
self.logger.info("KVM permission is granted!")
|
||||
else:
|
||||
raise RuntimeError("/dev/kvm cannot be found!")
|
||||
|
||||
def deploy(self):
|
||||
self.logger.info(f"Deploying the {self.device_type}")
|
||||
|
||||
basic_cmd = "emulator @{n}".format(n=self.name)
|
||||
basic_args = "-gpu swiftshader_indirect -accel on -writable-system -verbose"
|
||||
wipe_arg = "-wipe-data" if not self.is_initialized() else ""
|
||||
|
||||
start_cmd = f"{basic_cmd} {basic_args} {wipe_arg} {self.additional_args}"
|
||||
self.logger.info(f"Command to run {self.device_type}: '{start_cmd}'")
|
||||
subprocess.Popen(start_cmd.split())
|
||||
|
||||
def start(self) -> None:
|
||||
super().start()
|
||||
self.change_permission()
|
||||
self.deploy()
|
||||
|
||||
def check_adb_command(self, readiness_check_type: ReadinessCheck, bash_command: str,
|
||||
expected_keyword: str, max_attempts: int, interval_waiting_time: int,
|
||||
adb_action: str = None) -> None:
|
||||
success = False
|
||||
for _ in range(1, max_attempts):
|
||||
if success:
|
||||
break
|
||||
else:
|
||||
try:
|
||||
output = subprocess.check_output(
|
||||
bash_command.split()).decode(UTF8)
|
||||
if expected_keyword in str(output).lower():
|
||||
if readiness_check_type is self.ReadinessCheck.POP_UP_WINDOW:
|
||||
subprocess.check_call(adb_action, shell=True)
|
||||
else:
|
||||
self.logger.info(
|
||||
f"{self.device_type} is {readiness_check_type.value}!")
|
||||
success = True
|
||||
else:
|
||||
self.logger.info(f"[attempt: {_}] {self.device_type} is not {readiness_check_type.value}! "
|
||||
f"will check again in {interval_waiting_time} seconds")
|
||||
time.sleep(interval_waiting_time)
|
||||
except subprocess.CalledProcessError:
|
||||
self.logger.warning("command cannot be executed! will continue...")
|
||||
time.sleep(2)
|
||||
continue
|
||||
else:
|
||||
if readiness_check_type is self.ReadinessCheck.POP_UP_WINDOW:
|
||||
self.logger.info(f"Pop up windows '{expected_keyword}' is not found!")
|
||||
else:
|
||||
raise RuntimeError(
|
||||
f"{readiness_check_type.value} is checked {_} times!")
|
||||
|
||||
def wait_until_ready(self) -> None:
|
||||
super().wait_until_ready()
|
||||
booting_cmd = f"adb -s {self.adb_name} wait-for-device shell getprop sys.boot_completed"
|
||||
focus_cmd = f"adb -s {self.adb_name} shell dumpsys window | grep -i mCurrentFocus"
|
||||
self.check_adb_command(self.ReadinessCheck.BOOTED,
|
||||
booting_cmd, "1", 60, self.interval_waiting)
|
||||
time.sleep(self.interval_after_booting)
|
||||
|
||||
interval_pop_up = 0
|
||||
max_attempt_pop_up = 3
|
||||
pop_up_system_ui = "Not Responding: com.android.systemui"
|
||||
system_ui_cmd = f"adb shell su root 'kill $(pidof com.android.systemui)'"
|
||||
pop_up_key_enter = {
|
||||
"Not Responding: com.google.android.gms",
|
||||
"Not Responding: system",
|
||||
"ConversationListActivity"
|
||||
}
|
||||
key_enter_cmd = "adb shell input keyevent KEYCODE_ENTER"
|
||||
self.check_adb_command(self.ReadinessCheck.POP_UP_WINDOW, focus_cmd, pop_up_system_ui,
|
||||
max_attempt_pop_up, interval_pop_up, system_ui_cmd)
|
||||
for pe in pop_up_key_enter:
|
||||
self.check_adb_command(self.ReadinessCheck.POP_UP_WINDOW, focus_cmd, pe, max_attempt_pop_up,
|
||||
interval_pop_up, key_enter_cmd)
|
||||
|
||||
self.check_adb_command(self.ReadinessCheck.WELCOME_SCREEN,
|
||||
focus_cmd, "launcheractivity", 60, self.interval_waiting)
|
||||
self.logger.info(f"{self.device_type} is ready to use")
|
||||
|
||||
def tear_down(self, *args) -> None:
|
||||
self.logger.warning("Sigterm is detected! Nothing to do!")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
try:
|
||||
return "Emulator(name={n}, device={d}, adb_name={an}, android_version={av}, api_level={al}, " \
|
||||
"data_partition={dp}, additional_args={aa}, img_type={it}, sys_img={si}, " \
|
||||
"path_device_profile_target={pdpt}, path_emulator={pe}, path_emulator_config={pec}, " \
|
||||
"file={f})".format(n=self.name, d=self.device, an=self.adb_name, av=self.android_version,
|
||||
al=self.api_level, dp=self.data_partition, aa=self.additional_args,
|
||||
it=self.img_type, si=self.sys_img, pdpt=self.path_device_profile_target,
|
||||
pe=self.path_emulator, pec=self.path_emulator_config, f=self.file_name)
|
||||
except AttributeError as ae:
|
||||
self.logger.error(ae)
|
||||
return ""
|
||||
@@ -0,0 +1,223 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
from src.device import Genymotion, DeviceType
|
||||
from src.helper import get_env_value_or_raise
|
||||
from src.constants import ENV, UTF8
|
||||
|
||||
|
||||
class GenyAWS(Genymotion):
|
||||
port = 5555
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
self.device_type = DeviceType.GENY_AWS.value
|
||||
self.workdir = get_env_value_or_raise(ENV.WORK_PATH)
|
||||
self.aws_credentials_path = os.path.join(self.workdir, ".aws")
|
||||
self.remove_cred_at_the_end = False # for logout
|
||||
self.geny_aws_template_path = os.path.join(self.workdir, "docker-android", "mixins",
|
||||
"templates", "genymotion", "aws")
|
||||
self.created_devices = {}
|
||||
|
||||
def login(self) -> None:
|
||||
aws_credentials_file = os.path.join(self.aws_credentials_path, "credentials")
|
||||
if os.path.exists(self.aws_credentials_path):
|
||||
self.logger.info(".aws is found! It will be used as credentials")
|
||||
else:
|
||||
self.logger.info(".aws cannot be found! the template will be used!")
|
||||
self.remove_cred_at_the_end = True
|
||||
aws_credentials_template_path = os.path.join(self.geny_aws_template_path, ".aws")
|
||||
shutil.move(aws_credentials_template_path, self.aws_credentials_path)
|
||||
|
||||
aws_key_id = get_env_value_or_raise(ENV.AWS_ACCESS_KEY_ID)
|
||||
aws_secret_key = get_env_value_or_raise(ENV.AWS_SECRET_ACCESS_KEY)
|
||||
replacements_cred = {
|
||||
f"<{ENV.AWS_ACCESS_KEY_ID.lower()}>": aws_key_id,
|
||||
f"<{ENV.AWS_SECRET_ACCESS_KEY.lower()}>": aws_secret_key
|
||||
}
|
||||
with open(aws_credentials_file, 'r+') as cred_file:
|
||||
cred_file_contents = cred_file.read()
|
||||
for old_str, new_str in replacements_cred.items():
|
||||
cred_file_contents = cred_file_contents.replace(old_str, new_str)
|
||||
cred_file.seek(0)
|
||||
cred_file.write(cred_file_contents)
|
||||
cred_file.truncate()
|
||||
self.logger.info("aws credentials is set!")
|
||||
|
||||
def create_ssh_key(self) -> None:
|
||||
subprocess.check_call('ssh-keygen -t rsa -b 4096 -f ~/.ssh/id_rsa -N ""', shell=True)
|
||||
self.logger.info("ssh key is created!")
|
||||
|
||||
def create_tf_files(self) -> None:
|
||||
try:
|
||||
for item in self.get_data_from_template(ENV.GENY_AWS_TEMPLATE_FILE_NAME):
|
||||
name = item["name"]
|
||||
region = item["region"]
|
||||
ami = item["ami"]
|
||||
instance_type = item["instance_type"]
|
||||
if "security_group" in item:
|
||||
sg = item["security_group"]
|
||||
tf_content = f'''
|
||||
provider "aws" {{
|
||||
alias = "provider_{name}"
|
||||
region = "{region}"
|
||||
}}
|
||||
|
||||
resource "aws_key_pair" "geny_key_{name}" {{
|
||||
provider = aws.provider_{name}
|
||||
public_key = file("~/.ssh/id_rsa.pub")
|
||||
}}
|
||||
|
||||
resource "aws_instance" "geny_aws_{name}" {{
|
||||
provider = aws.provider_{name}
|
||||
ami = "{ami}"
|
||||
instance_type = "{instance_type}"
|
||||
vpc_security_group_ids = ["{sg}"]
|
||||
key_name = aws_key_pair.geny_key_{name}.key_name
|
||||
tags = {{
|
||||
Name = "DockerAndroid-GenyAWS-{ami}.id}}"
|
||||
}}
|
||||
|
||||
provisioner "remote-exec" {{
|
||||
connection {{
|
||||
type = "ssh"
|
||||
user = "shell"
|
||||
host = self.public_ip
|
||||
private_key = file("~/.ssh/id_rsa")
|
||||
}}
|
||||
script = "/home/androidusr/docker-android/mixins/scripts/genymotion/aws/enable_adb.sh"
|
||||
}}
|
||||
}}
|
||||
|
||||
output "public_dns_{name}" {{
|
||||
value = aws_instance.geny_aws_{name}.public_dns
|
||||
}}
|
||||
'''
|
||||
else:
|
||||
ingress_rules = json.dumps(item["ingress_rules"])
|
||||
egress_rules = json.dumps(item["egress_rules"])
|
||||
tf_content = f'''
|
||||
locals {{
|
||||
ingress_rules = {ingress_rules}
|
||||
egress_rules = {egress_rules}
|
||||
}}
|
||||
|
||||
provider "aws" {{
|
||||
alias = "provider_{name}"
|
||||
region = "{region}"
|
||||
}}
|
||||
|
||||
resource "aws_security_group" "geny_sg_{name}" {{
|
||||
provider = aws.provider_{name}
|
||||
dynamic "ingress" {{
|
||||
for_each = local.ingress_rules
|
||||
content {{
|
||||
from_port = ingress.value.from_port
|
||||
to_port = ingress.value.to_port
|
||||
protocol = ingress.value.protocol
|
||||
cidr_blocks = ingress.value.cidr_blocks
|
||||
}}
|
||||
}}
|
||||
|
||||
dynamic "egress" {{
|
||||
for_each = local.egress_rules
|
||||
content {{
|
||||
from_port = egress.value.from_port
|
||||
to_port = egress.value.to_port
|
||||
protocol = egress.value.protocol
|
||||
cidr_blocks = egress.value.cidr_blocks
|
||||
}}
|
||||
}}
|
||||
}}
|
||||
|
||||
resource "aws_key_pair" "geny_key_{name}" {{
|
||||
provider = aws.provider_{name}
|
||||
public_key = file("~/.ssh/id_rsa.pub")
|
||||
}}
|
||||
|
||||
resource "aws_instance" "geny_aws_{name}" {{
|
||||
provider = aws.provider_{name}
|
||||
ami = "{ami}"
|
||||
instance_type = "{instance_type}"
|
||||
vpc_security_group_ids = [aws_security_group.geny_sg_{name}.name]
|
||||
key_name = aws_key_pair.geny_key_{name}.key_name
|
||||
tags = {{
|
||||
Name = "DockerAndroid-GenyAWS-{ami}.id}}"
|
||||
}}
|
||||
|
||||
provisioner "remote-exec" {{
|
||||
connection {{
|
||||
type = "ssh"
|
||||
user = "shell"
|
||||
host = self.public_ip
|
||||
private_key = file("~/.ssh/id_rsa")
|
||||
}}
|
||||
script = "/home/androidusr/docker-android/mixins/scripts/genymotion/aws/enable_adb.sh"
|
||||
}}
|
||||
}}
|
||||
|
||||
output "public_dns_{name}" {{
|
||||
value = aws_instance.geny_aws_{name}.public_dns
|
||||
}}
|
||||
'''
|
||||
tf_deployment_filename = f"{name}.tf"
|
||||
self.created_devices[name] = GenyAWS.port
|
||||
GenyAWS.port += 1
|
||||
with open(tf_deployment_filename, "w") as df:
|
||||
df.write(tf_content)
|
||||
self.logger.info("Terraform files are created!")
|
||||
except Exception as e:
|
||||
self.logger.error(e)
|
||||
self.shutdown_and_logout()
|
||||
|
||||
def deploy_tf(self) -> None:
|
||||
try:
|
||||
cmds = (
|
||||
"terraform init",
|
||||
"terraform plan",
|
||||
"terraform apply -auto-approve"
|
||||
)
|
||||
for c in cmds:
|
||||
subprocess.check_call(c, shell=True)
|
||||
self.logger.info("Genymotion-Device(s) are deployed on AWS")
|
||||
except subprocess.CalledProcessError as cpe:
|
||||
self.logger.error(cpe)
|
||||
self.shutdown_and_logout()
|
||||
|
||||
def connect_with_local_adb(self) -> None:
|
||||
self.logger.info(f"created devices: {self.created_devices}")
|
||||
try:
|
||||
for d, p in self.created_devices.items():
|
||||
dns_cmd = f"terraform output public_dns_{d}"
|
||||
dns_ip = subprocess.check_output(dns_cmd.split()).decode(UTF8).replace('"', '')
|
||||
tunnel_cmd = f"ssh -i ~/.ssh/id_rsa -o ServerAliveInterval=60 -o StrictHostKeyChecking=no -q -NL " \
|
||||
f"{p}:localhost:5555 shell@{dns_ip}"
|
||||
subprocess.Popen(tunnel_cmd.split())
|
||||
time.sleep(10)
|
||||
subprocess.check_call(f"adb connect localhost:{p} >/dev/null 2>&1", shell=True)
|
||||
except Exception as e:
|
||||
self.logger.error(e)
|
||||
self.shutdown_and_logout()
|
||||
|
||||
def create(self) -> None:
|
||||
super().create()
|
||||
self.create_ssh_key()
|
||||
self.create_tf_files()
|
||||
self.deploy_tf()
|
||||
self.connect_with_local_adb()
|
||||
|
||||
def shutdown_and_logout(self) -> None:
|
||||
try:
|
||||
subprocess.check_call("terraform destroy -auto-approve -lock=false", shell=True)
|
||||
self.logger.info("device(s) is successfully removed!")
|
||||
except subprocess.CalledProcessError as cpe:
|
||||
self.logger.error(cpe)
|
||||
finally:
|
||||
if self.remove_cred_at_the_end:
|
||||
subprocess.check_call(f"rm -rf {self.aws_credentials_path}", shell=True)
|
||||
self.logger.info("successfully logged out!")
|
||||
@@ -0,0 +1,72 @@
|
||||
import logging
|
||||
import subprocess
|
||||
|
||||
from src.device import Genymotion, DeviceType
|
||||
from src.helper import get_env_value_or_raise
|
||||
from src.constants import ENV, UTF8
|
||||
|
||||
|
||||
class GenySAAS(Genymotion):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
self.device_type = DeviceType.GENY_SAAS.value
|
||||
self.created_devices = []
|
||||
|
||||
def login(self) -> None:
|
||||
user = get_env_value_or_raise(ENV.GENY_SAAS_USER)
|
||||
password = get_env_value_or_raise(ENV.GENY_SAAS_PASS)
|
||||
subprocess.check_call(f"gmsaas auth login {user} {password} > /dev/null 2>&1", shell=True)
|
||||
self.logger.info("successfully logged in!")
|
||||
|
||||
def create(self) -> None:
|
||||
super().create()
|
||||
for item in self.get_data_from_template(ENV.GENY_SAAS_TEMPLATE_FILE_NAME):
|
||||
name = ""
|
||||
template = ""
|
||||
local_port = ""
|
||||
|
||||
# implement like this because local_port param is not a must
|
||||
for k, v in item.items():
|
||||
if k.lower() == "name":
|
||||
name = v
|
||||
elif k.lower() == "template":
|
||||
template = v
|
||||
elif k.lower() == "local_port":
|
||||
local_port = v
|
||||
else:
|
||||
self.logger.warning(f"'{k}' is not supported! Please check the documentation!")
|
||||
|
||||
if not name:
|
||||
import uuid
|
||||
name = str(uuid.uuid4())
|
||||
|
||||
if not template:
|
||||
self.shutdown_and_logout()
|
||||
raise RuntimeError(f"'template' is a must parameter and not given!")
|
||||
else:
|
||||
self.logger.info(f"name: {name}, template: {template}")
|
||||
creation_cmd = f"gmsaas instances start {template} {name}"
|
||||
try:
|
||||
instance_id = subprocess.check_output(creation_cmd.split()).decode(UTF8).replace("\n", "")
|
||||
created_device = {f"{name}": {instance_id}}
|
||||
self.created_devices.append(created_device)
|
||||
additional_args = ""
|
||||
if local_port:
|
||||
additional_args = f"--adb-serial-port {local_port}"
|
||||
connect_cmd = f"gmsaas instances adbconnect {instance_id} {additional_args}"
|
||||
subprocess.check_call(f"{connect_cmd}", shell=True)
|
||||
except Exception as e:
|
||||
self.shutdown_and_logout()
|
||||
self.logger.error(e)
|
||||
exit(1)
|
||||
|
||||
def shutdown_and_logout(self) -> None:
|
||||
if bool(self.created_devices):
|
||||
self.logger.info("Created device(s) will be removed!")
|
||||
for d in self.created_devices:
|
||||
for n, i in d.items():
|
||||
subprocess.check_call(f"gmsaas instances stop {i}", shell=True)
|
||||
self.logger.info(f"device '{n}' is successfully removed!")
|
||||
subprocess.check_call("gmsaas auth logout", shell=True)
|
||||
self.logger.info("successfully logged out!")
|
||||
@@ -0,0 +1,56 @@
|
||||
import logging
|
||||
import os
|
||||
|
||||
logger = logging.getLogger("helper")
|
||||
|
||||
|
||||
def convert_str_to_bool(given_str: str) -> bool:
|
||||
"""
|
||||
Convert String to Boolean value
|
||||
|
||||
:param given_str: given string
|
||||
:return: converted string in Boolean
|
||||
"""
|
||||
if given_str:
|
||||
if type(given_str) is str:
|
||||
return given_str.lower() in ("yes", "true", "t", "1")
|
||||
else:
|
||||
raise AttributeError
|
||||
else:
|
||||
logger.info(f"'{given_str}' is empty!")
|
||||
return False
|
||||
|
||||
|
||||
def get_env_value_or_raise(env_key: str) -> str:
|
||||
"""
|
||||
Get value of necessary environment variable.
|
||||
|
||||
:param env_key: given environment variable
|
||||
:return: env_value in String
|
||||
"""
|
||||
try:
|
||||
env_value = os.getenv(env_key)
|
||||
if not env_value:
|
||||
raise RuntimeError(f"'{env_key}' is missing.")
|
||||
elif env_value.isspace():
|
||||
raise RuntimeError(f"'{env_key}' contains only white space.")
|
||||
return env_value
|
||||
except TypeError as t_err:
|
||||
logger.error(t_err)
|
||||
|
||||
|
||||
def symlink_force(source: str, target: str) -> None:
|
||||
"""
|
||||
Create Symbolic link
|
||||
|
||||
:param source: source file
|
||||
:param target: target file
|
||||
:return: None
|
||||
"""
|
||||
try:
|
||||
os.symlink(source, target)
|
||||
except FileNotFoundError as ffe_err:
|
||||
logger.error(ffe_err)
|
||||
except FileExistsError:
|
||||
os.remove(target)
|
||||
os.symlink(source, target)
|
||||
@@ -0,0 +1,3 @@
|
||||
import os
|
||||
|
||||
LOGGING_FILE = os.path.join(os.path.dirname(__file__), 'logging.conf')
|
||||
@@ -0,0 +1,7 @@
|
||||
import logging.config
|
||||
|
||||
from src.logger import LOGGING_FILE
|
||||
|
||||
|
||||
def init():
|
||||
logging.config.fileConfig(LOGGING_FILE)
|
||||
@@ -0,0 +1,21 @@
|
||||
[loggers]
|
||||
keys=root
|
||||
|
||||
[handlers]
|
||||
keys=console
|
||||
|
||||
[formatters]
|
||||
keys=formatter
|
||||
|
||||
[logger_root]
|
||||
level=INFO
|
||||
handlers=console
|
||||
|
||||
[handler_console]
|
||||
class=StreamHandler
|
||||
formatter=formatter
|
||||
args=(sys.stdout,)
|
||||
|
||||
[formatter_formatter]
|
||||
format=%(asctime)s %(levelname)s %(name)s - %(message)s
|
||||
datefmt=%Y-%m-%d %H:%M:%S
|
||||
@@ -0,0 +1,9 @@
|
||||
from unittest import TestCase
|
||||
|
||||
|
||||
class BaseTest(TestCase):
|
||||
def setUp(self) -> None:
|
||||
pass
|
||||
|
||||
def tearDown(self) -> None:
|
||||
pass
|
||||
@@ -0,0 +1,21 @@
|
||||
import os
|
||||
|
||||
from src.constants import ENV
|
||||
from src.tests import BaseTest
|
||||
|
||||
|
||||
class BaseDeviceTest(BaseTest):
|
||||
DEVICE_ENVS = {
|
||||
ENV.WORK_PATH: "/home/androidusr",
|
||||
ENV.USER_BEHAVIOR_ANALYTICS: str(False),
|
||||
ENV.EMULATOR_NO_SKIN: str(False)
|
||||
}
|
||||
|
||||
def setUp(self) -> None:
|
||||
for k, v in self.DEVICE_ENVS.items():
|
||||
os.environ[k] = v
|
||||
|
||||
def tearDown(self) -> None:
|
||||
for k in self.DEVICE_ENVS.keys():
|
||||
if os.environ[k]:
|
||||
del os.environ[k]
|
||||
@@ -0,0 +1,8 @@
|
||||
from src.device import Device
|
||||
from src.tests.device import BaseDeviceTest
|
||||
|
||||
|
||||
class TestDevice(BaseDeviceTest):
|
||||
def test_create_device(self):
|
||||
with self.assertRaises(TypeError):
|
||||
Device()
|
||||
@@ -0,0 +1,87 @@
|
||||
import mock
|
||||
|
||||
from src.device.emulator import Emulator
|
||||
from src.tests.device import BaseDeviceTest
|
||||
|
||||
|
||||
class TestEmulator(BaseDeviceTest):
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
self.name = "my_emu"
|
||||
self.device = "Nexus 4"
|
||||
self.a_version = "10.0"
|
||||
self.d_partition = "550m"
|
||||
self.additional_args = ""
|
||||
self.i_type = "google_apis"
|
||||
self.s_img = "x86"
|
||||
self.emu = Emulator(self.name, self.device, self.a_version, self.d_partition,
|
||||
self.additional_args, self.i_type, self.s_img)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
super().tearDown()
|
||||
|
||||
def test_adb_name(self):
|
||||
my_emu = Emulator("my_other_emu", self.device, self.a_version, self.d_partition,
|
||||
self.additional_args, self.i_type, self.s_img)
|
||||
self.assertNotEqual(self.emu.adb_name, my_emu.adb_name)
|
||||
|
||||
def test_invalid_device(self):
|
||||
with self.assertRaises(RuntimeError):
|
||||
Emulator("my_other_emu", "unknown device", self.a_version, self.d_partition,
|
||||
self.additional_args, self.i_type, self.s_img)
|
||||
with self.assertRaises(RuntimeError):
|
||||
Emulator("my_other_emu", "NEXUS 5", self.a_version, self.d_partition,
|
||||
self.additional_args, self.i_type, self.s_img)
|
||||
|
||||
def test_invalid_android_version(self):
|
||||
with self.assertRaises(RuntimeError):
|
||||
Emulator("my_other_emu", self.device, "0.0", self.d_partition,
|
||||
self.additional_args, self.i_type, self.s_img)
|
||||
|
||||
@mock.patch("os.path.exists", mock.MagicMock(return_value=False))
|
||||
def test_initialisation_config_not_exist(self):
|
||||
self.assertEqual(self.emu.is_initialized(), False)
|
||||
|
||||
@mock.patch("os.path.exists", mock.MagicMock(return_value=True))
|
||||
@mock.patch("builtins.open", mock.mock_open(read_data=""))
|
||||
def test_initialisation_device_not_exist(self):
|
||||
self.assertEqual(self.emu.is_initialized(), False)
|
||||
|
||||
@mock.patch("os.path.exists", mock.MagicMock(return_value=True))
|
||||
@mock.patch("builtins.open", mock.mock_open(read_data="hw.device.name=Nexus 4\n"))
|
||||
def test_initialisation_device_exists(self):
|
||||
self.assertEqual(self.emu.is_initialized(), True)
|
||||
|
||||
@mock.patch("src.device.Device.set_status")
|
||||
@mock.patch("src.device.emulator.Emulator._add_profile")
|
||||
@mock.patch("subprocess.check_call")
|
||||
@mock.patch("src.device.emulator.Emulator._add_skin")
|
||||
@mock.patch("src.device.emulator.Emulator.is_initialized", mock.MagicMock(return_value=False))
|
||||
def test_create_device_not_exist(self, mocked_status, mocked_profile, mocked_subprocess, mocked_skin):
|
||||
self.emu.create()
|
||||
self.assertEqual(mocked_status.called, True)
|
||||
self.assertEqual(mocked_profile.called, True)
|
||||
self.assertEqual(mocked_subprocess.called, True)
|
||||
self.assertEqual(mocked_skin.called, True)
|
||||
|
||||
@mock.patch("src.device.Device.set_status")
|
||||
@mock.patch("src.device.emulator.Emulator._add_profile")
|
||||
@mock.patch("subprocess.check_call")
|
||||
@mock.patch("src.device.emulator.Emulator._add_skin")
|
||||
@mock.patch("src.device.emulator.Emulator.is_initialized", mock.MagicMock(return_value=True))
|
||||
def test_create_device_exists(self, mocked_status, mocked_profile, mocked_subprocess, mocked_skin):
|
||||
self.emu.create()
|
||||
self.assertEqual(mocked_status.called, False)
|
||||
self.assertEqual(mocked_profile.called, False)
|
||||
self.assertEqual(mocked_subprocess.called, False)
|
||||
|
||||
def test_check_adb_command(self):
|
||||
with mock.patch("subprocess.check_output", mock.MagicMock(return_value="1".encode("utf-8"))):
|
||||
self.emu.check_adb_command(
|
||||
self.emu.ReadinessCheck.BOOTED, "mocked_command", "1", 3, 0)
|
||||
|
||||
def test_check_adb_command_out_of_attempts(self):
|
||||
with mock.patch("subprocess.check_output", mock.MagicMock(return_value=" ".encode("utf-8"))):
|
||||
with self.assertRaises(RuntimeError):
|
||||
self.emu.check_adb_command(
|
||||
self.emu.ReadinessCheck.BOOTED, "mocked_command", "1", 3, 0)
|
||||
@@ -0,0 +1,73 @@
|
||||
import os
|
||||
import mock
|
||||
|
||||
from src.helper import convert_str_to_bool, get_env_value_or_raise, symlink_force
|
||||
from src.tests import BaseTest
|
||||
|
||||
|
||||
class TestHelperMethods(BaseTest):
|
||||
def test_boolean_converter_with_valid_str(self):
|
||||
self.assertEqual(convert_str_to_bool("TRUE"), True)
|
||||
self.assertEqual(convert_str_to_bool("true"), True)
|
||||
self.assertEqual(convert_str_to_bool("T"), True)
|
||||
self.assertEqual(convert_str_to_bool("Yes"), True)
|
||||
self.assertEqual(convert_str_to_bool("1"), True)
|
||||
self.assertEqual(convert_str_to_bool("False"), False)
|
||||
self.assertEqual(convert_str_to_bool("f"), False)
|
||||
self.assertEqual(convert_str_to_bool("0"), False)
|
||||
|
||||
def test_boolean_converter_with_empty(self):
|
||||
self.assertEqual(convert_str_to_bool(None), False)
|
||||
self.assertEqual(convert_str_to_bool(""), False)
|
||||
|
||||
def test_boolean_converter_with_invalid_str(self):
|
||||
self.assertEqual(convert_str_to_bool(" "), False)
|
||||
self.assertEqual(convert_str_to_bool("test"), False)
|
||||
|
||||
def test_boolean_converter_with_invalid_format(self):
|
||||
with self.assertRaises(AttributeError):
|
||||
convert_str_to_bool(True)
|
||||
|
||||
def test_get_env_value_from_valid_key(self):
|
||||
env_key = "env_key01"
|
||||
os.environ[env_key] = "env_value01"
|
||||
get_env_value_or_raise(env_key)
|
||||
del os.environ[env_key]
|
||||
|
||||
def test_get_env_value_with_empty_string(self):
|
||||
with self.assertRaises(RuntimeError):
|
||||
env_key = "env_key01"
|
||||
os.environ[env_key] = " "
|
||||
get_env_value_or_raise(env_key)
|
||||
del os.environ[env_key]
|
||||
|
||||
def test_get_env_value_from_invalid_key(self):
|
||||
with self.assertRaises(RuntimeError):
|
||||
get_env_value_or_raise("env_key02")
|
||||
|
||||
def test_get_env_value_with_invalid_format(self):
|
||||
with mock.patch("src.logger"):
|
||||
get_env_value_or_raise(True)
|
||||
|
||||
def test_symlink(self):
|
||||
s = os.path.join("source.txt")
|
||||
t = os.path.join("target_file.txt")
|
||||
open(s, "a").close()
|
||||
symlink_force(s, t)
|
||||
os.remove(s)
|
||||
os.remove(t)
|
||||
|
||||
def test_symlink_already_exist(self):
|
||||
s = os.path.join("source.txt")
|
||||
t = os.path.join("target_file.txt")
|
||||
open(s, "a").close()
|
||||
open(t, "a").close()
|
||||
symlink_force(s, t)
|
||||
os.remove(s)
|
||||
os.remove(t)
|
||||
|
||||
def test_symlink_file_not_exists(self):
|
||||
s = os.path.join("source.txt")
|
||||
t = os.path.join("target_file.txt")
|
||||
symlink_force(s, t)
|
||||
os.remove(t)
|
||||
Reference in New Issue
Block a user