Fix AttachError because of parallel execution (#197)

* Fix AttachError because of parallel execution

* Fix X connection failed

test/e2e/configs/squish.py

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

test/e2e/conftest.py

@@ -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(),

test/e2e/driver/aut.py

@@ -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()

test/e2e/driver/context.py

@@ -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:

test/e2e/driver/server.py

@@ -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):
+    def __new__(cls):
-        self.path = configs.testpath.SQUISH_DIR / 'bin' / 'squishserver'
+        if not SquishServer.__instance:
-        self.config = configs.testpath.ROOT / 'squish_server.ini'
+            SquishServer.__instance = super(SquishServer, cls).__new__(cls)
-        self.host = '127.0.0.1'
+        return SquishServer.__instance
-        self.port = None
-        self.pid = None
 
-    def start(self):
+    @classmethod
-        self.port = local_system.find_free_port(configs.squish.SERVET_PORT, 100)
+    def start(cls):
+        cls.port = local_system.find_free_port(configs.squish.SERVER_PORT, 100)
         cmd = [
-            f'"{self.path}"',
+            f'"{cls.path}"',
-            '--configfile', str(self.config),
+            '--configfile', str(cls.config),
-            f'--host={self.host}',
+            f'--host={cls.host}',
-            f'--port={self.port}',
+            f'--port={cls.port}',
         ]
-        self.pid = local_system.execute(cmd)
+        cls.pid = local_system.execute(cmd)
 
-    def stop(self):
+    @classmethod
-        if self.pid is not None:
+    def stop(cls):
-            local_system.kill_process(self.pid, verify=True)
+        if cls.pid is not None:
-            self.pid = None
+            local_system.kill_process(cls.pid, verify=True)
-        self.port = None
+            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):
+    @classmethod
-        self.configuring('addAUT', [aut_id, f'"{app_dir}"'])
+    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):
+    @classmethod
-        self.configuring('addAttachableAUT', [aut_id, f'localhost:{port}'])
+    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):
+    @classmethod
-        self.configuring('setAUTTimeout', [str(value)])
+    def set_aut_timeout(cls, value: int = configs.timeouts.PROCESS_TIMEOUT_SEC):
+        cls.configuring('setAUTTimeout', [str(value)])