Fix AUT AttachError (#180)

* Fix AUT AttachError

#174

* Test close by port

* Method to add a screenshot on fail added for multiple instances AUT

#177
This commit is contained in:
Vladimir Druzhinin 2023-10-19 14:26:05 +02:00 committed by GitHub
parent 733ccdfd92
commit 3343bc675e
4 changed files with 51 additions and 40 deletions

View File

@ -1,3 +1,2 @@
AUT_PORT = 61500 AUT_PORT = 61500
MAX_INSTANCES = 3
SERVET_PORT = 4322 SERVET_PORT = 4322

View File

@ -62,12 +62,5 @@ def pytest_exception_interact(node):
name='Screenshot on fail', name='Screenshot on fail',
body=screenshot.read_bytes(), body=screenshot.read_bytes(),
attachment_type=allure.attachment_type.PNG) attachment_type=allure.attachment_type.PNG)
driver.context.detach()
# The maximum count of multiple instances in tests now is 3
max_instances = configs.squish.MAX_INSTANCES
for port in range(configs.squish.AUT_PORT, configs.squish.AUT_PORT + (max_instances * 1000), 1000):
pid = local_system.find_process_by_port(port)
if pid is not None:
local_system.kill_process(pid, verify=True)
except Exception as ex: except Exception as ex:
_logger.debug(ex) _logger.debug(ex)

View File

@ -3,6 +3,7 @@ from datetime import datetime
import allure import allure
import squish import squish
from PIL import ImageGrab
import configs import configs
import driver import driver
@ -18,16 +19,13 @@ class AUT:
def __init__( def __init__(
self, self,
app_path: system_path.SystemPath = configs.APP_DIR, app_path: system_path.SystemPath = configs.APP_DIR,
host: str = '127.0.0.1',
port: int = local_system.find_free_port(configs.squish.AUT_PORT, 1000),
user_data: SystemPath = None user_data: SystemPath = None
): ):
super(AUT, self).__init__() super(AUT, self).__init__()
self.path = app_path self.path = app_path
self.host = host
self.port = int(port)
self.ctx = None self.ctx = None
self.pid = None self.pid = None
self.port = None
self.aut_id = f'AUT_{datetime.now():%H%M%S}' 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}' self.app_data = configs.testpath.STATUS_DATA / f'app_{datetime.now():%H%M%S_%f}'
if user_data is not None: if user_data is not None:
@ -40,7 +38,14 @@ class AUT:
def __enter__(self): def __enter__(self):
return self.launch() return self.launch()
def __exit__(self, *args): def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
screenshot = configs.testpath.RUN / 'screenshot.png'
if screenshot.exists():
screenshot = configs.testpath.RUN / f'screenshot_{datetime.now():%H%M%S}.png'
ImageGrab.grab().save(screenshot)
allure.attach(
name='Screenshot on fail', body=screenshot.read_bytes(), attachment_type=allure.attachment_type.PNG)
self.detach().stop() self.detach().stop()
@allure.step('Attach Squish to Test Application') @allure.step('Attach Squish to Test Application')
@ -64,12 +69,20 @@ class AUT:
@allure.step('Close application') @allure.step('Close application')
def stop(self): def stop(self):
if self.pid is not None:
local_system.kill_process(self.pid, verify=True) local_system.kill_process(self.pid, verify=True)
self.pid = None
if self.port is not None:
assert driver.waitFor(lambda: local_system.find_process_by_port(self.port) is None,
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
@allure.step("Start application") @allure.step("Start application")
def launch(self, ) -> 'AUT': def launch(self, attempt: int = 2) -> 'AUT':
try:
self.port = local_system.find_free_port(configs.squish.AUT_PORT, 1000)
SquishServer().set_aut_timeout() SquishServer().set_aut_timeout()
if configs.ATTACH_MODE: if configs.ATTACH_MODE:
SquishServer().add_attachable_aut(self.aut_id, self.port) SquishServer().add_attachable_aut(self.aut_id, self.port)
command = [ command = [
@ -79,15 +92,23 @@ class AUT:
f'-d={self.app_data}' f'-d={self.app_data}'
] ]
local_system.execute(command) local_system.execute(command)
self.attach()
else: else:
SquishServer().add_executable_aut(self.aut_id, self.path.parent) SquishServer().add_executable_aut(self.aut_id, self.path.parent)
command = [self.aut_id, f'-d={self.app_data}'] command = [self.aut_id, f'-d={self.app_data}']
self.ctx = squish.startApplication(' '.join(command), configs.timeouts.PROCESS_TIMEOUT_SEC) self.ctx = squish.startApplication(' '.join(command), configs.timeouts.PROCESS_TIMEOUT_SEC)
self.attach()
self.pid = self.ctx.pid self.pid = self.ctx.pid
assert squish.waitFor(lambda: self.ctx.isRunning, configs.timeouts.PROCESS_TIMEOUT_SEC) assert squish.waitFor(lambda: self.ctx.isRunning, configs.timeouts.PROCESS_TIMEOUT_SEC)
return self return self
except AssertionError as err:
_logger.debug(err)
if attempt:
self.detach().stop()
return self.launch(attempt-1)
else:
raise err
@allure.step('Restart application') @allure.step('Restart application')
def restart(self): def restart(self):

View File

@ -11,18 +11,15 @@ _logger = logging.getLogger(__name__)
class SquishServer: class SquishServer:
def __init__( def __init__(self):
self,
host: str = '127.0.0.1',
port: int = local_system.find_free_port(configs.squish.SERVET_PORT, 100)
):
self.path = configs.testpath.SQUISH_DIR / 'bin' / 'squishserver' self.path = configs.testpath.SQUISH_DIR / 'bin' / 'squishserver'
self.config = configs.testpath.ROOT / 'squish_server.ini' self.config = configs.testpath.ROOT / 'squish_server.ini'
self.host = host self.host = '127.0.0.1'
self.port = port self.port = None
self.pid = None self.pid = None
def start(self): def start(self):
self.port = local_system.find_free_port(configs.squish.SERVET_PORT, 100)
cmd = [ cmd = [
f'"{self.path}"', f'"{self.path}"',
'--configfile', str(self.config), '--configfile', str(self.config),
@ -35,6 +32,7 @@ class SquishServer:
if self.pid is not None: if self.pid is not None:
local_system.kill_process(self.pid, verify=True) local_system.kill_process(self.pid, verify=True)
self.pid = None self.pid = None
self.port = None
# https://doc-snapshots.qt.io/squish/cli-squishserver.html # https://doc-snapshots.qt.io/squish/cli-squishserver.html
def configuring(self, action: str, options: typing.Union[int, str, list]): def configuring(self, action: str, options: typing.Union[int, str, list]):