2023-08-10 06:58:50 +00:00
|
|
|
import logging
|
|
|
|
import time
|
2023-08-04 18:27:03 +00:00
|
|
|
import typing
|
|
|
|
|
|
|
|
import configs.testpath
|
|
|
|
from scripts.utils import local_system
|
|
|
|
|
|
|
|
_PROCESS_NAME = '_squishserver'
|
|
|
|
|
2023-08-10 06:58:50 +00:00
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
2023-08-04 18:27:03 +00:00
|
|
|
|
|
|
|
class SquishServer:
|
|
|
|
|
|
|
|
def __init__(
|
|
|
|
self,
|
|
|
|
host: str = '127.0.0.1',
|
|
|
|
port: int = 4322
|
|
|
|
):
|
|
|
|
self.path = configs.testpath.SQUISH_DIR / 'bin' / 'squishserver'
|
|
|
|
self.config = configs.testpath.ROOT / 'squish_server.ini'
|
|
|
|
self.host = host
|
|
|
|
self.port = port
|
|
|
|
|
|
|
|
def start(self):
|
|
|
|
cmd = [
|
|
|
|
f'"{self.path}"',
|
|
|
|
'--configfile', str(self.config),
|
|
|
|
f'--host={self.host}',
|
|
|
|
f'--port={self.port}',
|
|
|
|
]
|
|
|
|
local_system.execute(cmd)
|
|
|
|
try:
|
|
|
|
local_system.wait_for_started(_PROCESS_NAME)
|
2023-08-10 06:58:50 +00:00
|
|
|
except AssertionError as err:
|
|
|
|
_logger.info(err)
|
2023-08-04 18:27:03 +00:00
|
|
|
local_system.execute(cmd, check=True)
|
|
|
|
|
2023-08-10 06:58:50 +00:00
|
|
|
def stop(self, attempt: int = 2):
|
|
|
|
local_system.run([self.path, '--stop'])
|
|
|
|
try:
|
|
|
|
local_system.wait_for_close(_PROCESS_NAME, 2)
|
|
|
|
except AssertionError as err:
|
|
|
|
_logger.debug(err)
|
|
|
|
if attempt:
|
|
|
|
time.sleep(1)
|
|
|
|
self.stop(attempt-1)
|
|
|
|
else:
|
|
|
|
raise err
|
2023-08-04 18:27:03 +00:00
|
|
|
|
|
|
|
# https://doc-snapshots.qt.io/squish/cli-squishserver.html
|
|
|
|
def configuring(self, action: str, options: typing.Union[int, str, list]):
|
|
|
|
local_system.run(
|
|
|
|
[f'"{self.path}"', '--configfile', str(self.config), '--config', action, ' '.join(options)])
|
|
|
|
|
|
|
|
def add_executable_aut(self, aut_id, app_dir):
|
|
|
|
self.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}'])
|
|
|
|
|
|
|
|
def set_aut_timeout(self, value: int = configs.timeouts.PROCESS_TIMEOUT_SEC):
|
|
|
|
self.configuring('setAUTTimeout', [str(value)])
|