chore: Improve AUT methods (#223)

- Close and Detach methods are merged
- Saving pid in Aut changed to close after AttachError
- Collect all pid that use port
This commit is contained in:
Vladimir Druzhinin 2023-10-27 13:44:45 +02:00 committed by GitHub
parent 176481564c
commit ee95fbc150
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 37 additions and 36 deletions

View File

@ -47,43 +47,40 @@ class AUT:
ImageGrab.grab(xdisplay=":0" if IS_LIN else None).save(screenshot)
allure.attach(
name='Screenshot on fail', body=screenshot.read_bytes(), attachment_type=allure.attachment_type.PNG)
self.detach().stop()
self.stop()
@allure.step('Attach Squish to Test Application')
def attach(self, timeout_sec: int = configs.timeouts.PROCESS_TIMEOUT_SEC, attempt: int = 2):
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)
try:
squish.setApplicationContext(self.ctx)
except TypeError as err:
if attempt:
return self.attach(timeout_sec, attempt - 1)
else:
raise err
@allure.step('Detach Squish and Application')
def detach(self):
if self.ctx is not None:
squish.currentApplicationContext().detach()
self.ctx = None
return self
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
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")
def launch(self, attempt: int = 2) -> 'AUT':
try:
self.port = local_system.find_free_port(configs.squish.AUT_PORT, 1000)
SquishServer().set_aut_timeout()
if configs.ATTACH_MODE:
SquishServer().add_attachable_aut(self.aut_id, self.port)
command = [
@ -92,19 +89,19 @@ class AUT:
f'"{self.path}"',
f'-d={self.app_data}'
]
local_system.execute(command)
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()
self.pid = self.ctx.pid
assert squish.waitFor(lambda: self.ctx.isRunning, configs.timeouts.PROCESS_TIMEOUT_SEC)
return self
except AssertionError as err:
except (AssertionError, TypeError) as err:
_logger.debug(err)
self.detach().stop()
self.stop()
if attempt:
return self.launch(attempt-1)
else:
@ -113,5 +110,5 @@ class AUT:
@allure.step('Restart application')
def restart(self):
self.detach().stop()
self.stop()
self.launch()

View File

@ -20,9 +20,8 @@ def attach(aut_id: str, timeout_sec: int = configs.timeouts.PROCESS_TIMEOUT_SEC)
_logger.info(f'AUT: {aut_id} attached')
return context
except RuntimeError as err:
_logger.debug(err)
time.sleep(1)
assert time.monotonic() - started_at < timeout_sec, f'Attach error: {aut_id}'
assert time.monotonic() - started_at < timeout_sec, str(err)
@allure.step('Detaching')

View File

@ -13,10 +13,11 @@ from scripts.utils.system_path import SystemPath
@pytest.fixture
def application_logs():
yield
for app_data in configs.testpath.STATUS_DATA.iterdir():
for log in (app_data / 'logs').iterdir():
allure.attach.file(log, name=str(log.name), attachment_type=allure.attachment_type.TEXT)
log.unlink()
if configs.testpath.STATUS_DATA.exists():
for app_data in configs.testpath.STATUS_DATA.iterdir():
for log in (app_data / 'logs').iterdir():
allure.attach.file(log, name=str(log.name), attachment_type=allure.attachment_type.TEXT)
log.unlink()
@pytest.fixture
@ -49,7 +50,7 @@ def multiple_instance():
def main_window(aut: AUT, user_data):
aut.launch()
yield MainWindow().wait_until_appears().prepare()
aut.detach().stop()
aut.stop()
@pytest.fixture

View File

@ -13,6 +13,7 @@ def start_squish_server():
try:
squish_server.start()
squish_server.set_cursor_animation()
squish_server.set_aut_timeout()
break
except AssertionError as err:
attempt -= 1

View File

@ -3,6 +3,7 @@ import os
import signal
import subprocess
import time
import typing
import allure
import psutil
@ -13,14 +14,16 @@ from configs.system import IS_WIN
_logger = logging.getLogger(__name__)
def find_process_by_port(port: int) -> int:
def find_process_by_port(port: int) -> typing.List[int]:
pid_list = []
for proc in psutil.process_iter():
try:
for conns in proc.connections(kind='inet'):
if conns.laddr.port == port:
return proc.pid
pid_list.append(proc.pid)
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
pass
return pid_list
def find_free_port(start: int, step: int):