135 lines
4.8 KiB
Python
135 lines
4.8 KiB
Python
import logging
|
|
from datetime import datetime
|
|
|
|
import allure
|
|
import cv2
|
|
import numpy as np
|
|
import squish
|
|
from PIL import ImageGrab
|
|
|
|
import configs
|
|
import driver
|
|
from configs.system import IS_LIN
|
|
from driver import context
|
|
from driver.server import SquishServer
|
|
from gui.objects_map import statusDesktop_mainWindow
|
|
from scripts.utils import system_path, local_system
|
|
from scripts.utils.system_path import SystemPath
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AUT:
|
|
def __init__(
|
|
self,
|
|
app_path: system_path.SystemPath = configs.APP_DIR,
|
|
user_data: SystemPath = None
|
|
):
|
|
super(AUT, self).__init__()
|
|
self.path = app_path
|
|
self.ctx = None
|
|
self.pid = None
|
|
self.port = None
|
|
self.aut_id = f'AUT_{datetime.now():%H%M%S}'
|
|
self.app_data = configs.testpath.STATUS_DATA / f'app_{datetime.now():%H%M%S_%f}'
|
|
if user_data is not None:
|
|
user_data.copy_to(self.app_data / 'data')
|
|
self.options = ''
|
|
driver.testSettings.setWrappersForApplication(self.aut_id, ['Qt'])
|
|
|
|
def __str__(self):
|
|
return type(self).__qualname__
|
|
|
|
def __enter__(self):
|
|
return self.launch()
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
if exc_type:
|
|
try:
|
|
self.attach()
|
|
driver.waitForObjectExists(statusDesktop_mainWindow).setVisible(True)
|
|
configs.testpath.TEST.mkdir(parents=True, exist_ok=True)
|
|
screenshot = configs.testpath.TEST / f'{self.aut_id}.png'
|
|
|
|
rect = driver.object.globalBounds(driver.waitForObject(statusDesktop_mainWindow))
|
|
img = ImageGrab.grab(
|
|
bbox=(rect.x, rect.y, rect.x + rect.width, rect.y + rect.height),
|
|
xdisplay=configs.system.DISPLAY if IS_LIN else None)
|
|
view = cv2.cvtColor(np.array(img), cv2.COLOR_BGR2RGB)
|
|
cv2.imwrite(str(screenshot), view)
|
|
|
|
allure.attach(
|
|
name=f'Screenshot on fail: {self.aut_id}',
|
|
body=screenshot.read_bytes(),
|
|
attachment_type=allure.attachment_type.PNG)
|
|
except Exception as err:
|
|
_logger.info(err)
|
|
|
|
self.stop()
|
|
|
|
@allure.step('Attach Squish to Test Application')
|
|
def attach(self, timeout_sec: int = configs.timeouts.PROCESS_TIMEOUT_SEC):
|
|
if self.ctx is None:
|
|
self.ctx = context.attach(self.aut_id, timeout_sec)
|
|
squish.setApplicationContext(self.ctx)
|
|
|
|
@allure.step('Close application')
|
|
def stop(self):
|
|
if self.ctx is not None:
|
|
squish.currentApplicationContext().detach()
|
|
self.ctx = None
|
|
|
|
if self.port is not None:
|
|
pid_list = local_system.find_process_by_port(self.port)
|
|
if pid_list:
|
|
for pid in pid_list:
|
|
local_system.kill_process(pid, verify=True)
|
|
assert driver.waitFor(
|
|
lambda: not local_system.find_process_by_port(self.port), configs.timeouts.PROCESS_TIMEOUT_SEC), \
|
|
f'Port {self.port} still in use by process: {local_system.find_process_by_port(self.port)}'
|
|
self.port = None
|
|
if self.pid in pid_list:
|
|
self.pid = None
|
|
|
|
if self.pid is not None:
|
|
local_system.kill_process(self.pid, verify=True)
|
|
self.pid = None
|
|
|
|
@allure.step("Start application")
|
|
def launch(self, options='', attempt: int = 2) -> 'AUT':
|
|
self.options = options
|
|
try:
|
|
self.port = local_system.find_free_port(configs.squish.AUT_PORT, 1000)
|
|
if configs.ATTACH_MODE:
|
|
SquishServer().add_attachable_aut(self.aut_id, self.port)
|
|
command = [
|
|
configs.testpath.SQUISH_DIR / 'bin' / 'startaut',
|
|
f'--port={self.port}',
|
|
f'"{self.path}"',
|
|
f'-d={self.app_data}',
|
|
options
|
|
]
|
|
self.pid = local_system.execute(command)
|
|
else:
|
|
SquishServer().add_executable_aut(self.aut_id, self.path.parent)
|
|
command = [self.aut_id, f'-d={self.app_data}']
|
|
self.ctx = squish.startApplication(' '.join(command), configs.timeouts.PROCESS_TIMEOUT_SEC)
|
|
self.pid = self.ctx.pid
|
|
|
|
self.attach()
|
|
assert squish.waitFor(lambda: self.ctx.isRunning, configs.timeouts.PROCESS_TIMEOUT_SEC)
|
|
return self
|
|
except (AssertionError, TypeError) as err:
|
|
_logger.debug(err)
|
|
self.stop()
|
|
if attempt:
|
|
return self.launch(options, attempt - 1)
|
|
else:
|
|
raise err
|
|
|
|
|
|
@allure.step('Restart application')
|
|
def restart(self):
|
|
self.stop()
|
|
self.launch()
|