Fix AttachError because of parallel execution (#197)

* Fix AttachError because of parallel execution

* Fix X connection failed
This commit is contained in:
Vladimir Druzhinin 2023-10-23 13:53:11 +02:00 committed by GitHub
parent 3e3929b8a0
commit 608a201a22
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 46 additions and 31 deletions

View File

@ -1,2 +1,4 @@
AUT_PORT = 61500
SERVET_PORT = 4322
import os
AUT_PORT = 61500 + int(os.getenv('BUILD_NUMBER', 0))
SERVER_PORT = 4322 + int(os.getenv('BUILD_NUMBER', 0))

View File

@ -7,6 +7,7 @@ from PIL import ImageGrab
import configs
import driver
from configs.system import IS_LIN
from fixtures.path import generate_test_info
from scripts.utils import local_system
from scripts.utils.system_path import SystemPath
@ -57,7 +58,7 @@ def pytest_exception_interact(node):
screenshot = node_dir / 'screenshot.png'
if screenshot.exists():
screenshot = node_dir / f'screenshot_{datetime.now():%H%M%S}.png'
ImageGrab.grab().save(screenshot)
ImageGrab.grab(xdisplay=":0" if IS_LIN else None).save(screenshot)
allure.attach(
name='Screenshot on fail',
body=screenshot.read_bytes(),

View File

@ -7,6 +7,7 @@ from PIL import ImageGrab
import configs
import driver
from configs.system import IS_LIN
from driver import context
from driver.server import SquishServer
from scripts.utils import system_path, local_system
@ -43,7 +44,7 @@ class AUT:
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)
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()

View File

@ -5,6 +5,7 @@ import allure
import squish
import configs
from driver.server import SquishServer
_logger = logging.getLogger(__name__)
@ -15,7 +16,7 @@ def attach(aut_id: str, timeout_sec: int = configs.timeouts.PROCESS_TIMEOUT_SEC)
_logger.debug(f'Attaching to {aut_id}')
while True:
try:
context = squish.attachToApplication(aut_id)
context = squish.attachToApplication(aut_id, SquishServer().host, SquishServer().port)
_logger.info(f'AUT: {aut_id} attached')
return context
except RuntimeError as err:

View File

@ -10,40 +10,50 @@ _logger = logging.getLogger(__name__)
class SquishServer:
__instance = None
path = configs.testpath.SQUISH_DIR / 'bin' / 'squishserver'
config = configs.testpath.ROOT / 'squish_server.ini'
host = '127.0.0.1'
port = None
pid = None
def __init__(self):
self.path = configs.testpath.SQUISH_DIR / 'bin' / 'squishserver'
self.config = configs.testpath.ROOT / 'squish_server.ini'
self.host = '127.0.0.1'
self.port = None
self.pid = None
def __new__(cls):
if not SquishServer.__instance:
SquishServer.__instance = super(SquishServer, cls).__new__(cls)
return SquishServer.__instance
def start(self):
self.port = local_system.find_free_port(configs.squish.SERVET_PORT, 100)
@classmethod
def start(cls):
cls.port = local_system.find_free_port(configs.squish.SERVER_PORT, 100)
cmd = [
f'"{self.path}"',
'--configfile', str(self.config),
f'--host={self.host}',
f'--port={self.port}',
f'"{cls.path}"',
'--configfile', str(cls.config),
f'--host={cls.host}',
f'--port={cls.port}',
]
self.pid = local_system.execute(cmd)
cls.pid = local_system.execute(cmd)
def stop(self):
if self.pid is not None:
local_system.kill_process(self.pid, verify=True)
self.pid = None
self.port = None
@classmethod
def stop(cls):
if cls.pid is not None:
local_system.kill_process(cls.pid, verify=True)
cls.pid = None
cls.port = None
# https://doc-snapshots.qt.io/squish/cli-squishserver.html
def configuring(self, action: str, options: typing.Union[int, str, list]):
@classmethod
def configuring(cls, action: str, options: typing.Union[int, str, list]):
local_system.run(
[f'"{self.path}"', '--configfile', str(self.config), '--config', action, ' '.join(options)])
[f'"{cls.path}"', '--configfile', str(cls.config), '--config', action, ' '.join(options)])
def add_executable_aut(self, aut_id, app_dir):
self.configuring('addAUT', [aut_id, f'"{app_dir}"'])
@classmethod
def add_executable_aut(cls, aut_id, app_dir):
cls.configuring('addAUT', [aut_id, f'"{app_dir}"'])
def add_attachable_aut(self, aut_id: str, port: int):
self.configuring('addAttachableAUT', [aut_id, f'localhost:{port}'])
@classmethod
def add_attachable_aut(cls, aut_id: str, port: int):
cls.configuring('addAttachableAUT', [aut_id, f'localhost:{port}'])
def set_aut_timeout(self, value: int = configs.timeouts.PROCESS_TIMEOUT_SEC):
self.configuring('setAUTTimeout', [str(value)])
@classmethod
def set_aut_timeout(cls, value: int = configs.timeouts.PROCESS_TIMEOUT_SEC):
cls.configuring('setAUTTimeout', [str(value)])