LibrePCB

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import platform
import pytest
import re
import shutil
import slint_testing
import time
from copy import deepcopy
from typing import Iterable


ROOT_DIR = os.path.dirname(__file__)
TESTS_DIR = os.path.dirname(ROOT_DIR)
REPO_DIR = os.path.dirname(TESTS_DIR)
DATA_DIR = os.path.join(TESTS_DIR, "data")


def pytest_addoption(parser):
    parser.addoption(
        "--librepcb-executable",
        action="store",
        help="Path to librepcb executable to test",
    )


def pytest_report_header(config):
    return f"librepcb: {os.path.abspath(config.getoption('--librepcb-executable'))}"


# Avoid errors due to too long file path on Windows :-|
def _long_path(path):
    if os.name == "nt":
        return "\\\\?\\" + path.replace("/", "\\")
    else:
        return path


def _flatten(items):
    for x in items:
        if isinstance(x, Iterable) and not isinstance(x, (str, bytes)):
            for sub_x in _flatten(x):
                yield sub_x
        else:
            yield x


def _query_children(element, query):
    segment = query[0].split("[")
    segment_name = segment[0]
    if segment_name.startswith("#"):
        segment_name = segment_name[1:]
        children = element.query_descendants().match_type_name(segment_name).find_all()
    elif "{" in segment_name:
        children = []
        start = segment_name.find("{")
        end = segment_name.find("}")
        for value in segment_name[start + 1 : end].split(","):
            name = segment_name[:start] + value + segment_name[end + 1 :]
            children += element.query_descendants().match_id(name).find_all()
    else:
        children = element.query_descendants().match_id(segment_name).find_all()
    if len(segment) > 1:
        index = int(segment[1][:-1])
        children = [children[index]]
    if len(query) > 1:
        children = [_query_children(child, query[1:]) for child in children]
    return children


class Rect:
    """
    Position & size of a UI element
    """

    def __init__(self, pos, size):
        self.pos = pos
        self.size = size

    def position_abs(self, rel_x, rel_y):
        """
        Convert a relative, normalized position to an absolute position

        For example (0, 0) to get the top left, (1, 1) to get the bottom
        right position of the UI element.
        """
        return (self.pos[0] + self.size[0] * rel_x, self.pos[1] + self.size[1] * rel_y)

    @property
    def center_abs(self):
        """
        Get the absolute center position of the UI element
        """
        return self.position_abs(0.5, 0.5)


class Element:
    """
    Wrapper around slint_testing.Element

    Basically this wrapper only provides shorter names for the properties and
    functions, to keep test code as readable as possible.
    """

    def __init__(self, element):
        self._element = element

    @property
    def valid(self):
        return self._element.is_valid

    @property
    def position_abs(self):
        return self._element.absolute_position

    @property
    def size(self):
        return self._element.size

    @property
    def rect(self):
        return Rect(self.position_abs, self.size)

    @property
    def label(self):
        return self._element.accessible_label

    @property
    def placeholder(self):
        return self._element.accessible_placeholder_text

    @property
    def value(self):
        return self._element.accessible_value

    @property
    def checked(self):
        return self._element.accessible_checked

    @property
    def enabled(self):
        return self._element.accessible_enabled

    @property
    def readonly(self):
        return self._element.accessible_read_only

    def trigger(self):
        self._element.invoke_accessible_default_action()

    def click(self):
        self._element.single_click(slint_testing.PointerEventButton.Left)

    def mclick(self):
        self._element.double_click(slint_testing.PointerEventButton.Middle)

    def dclick(self):
        self._element.double_click(slint_testing.PointerEventButton.Left)

    def set_value(self, value):
        self._element.accessible_value = value

    def set_checked(self, checked):
        if self._element.accessible_checked != checked:
            self._element.invoke_accessible_default_action()


class ElementQuery:
    """
    Helper to query and validate UI elements

    The `get()` method allows to pass a query string in this form:

        LibrariesPanel::local-libs #LibraryListViewItem *

    Where:

    * Each nesting level is separated by a space. Queries are executed
      recursively, so not every level needs to be supplied.
    * Each string segment matches Slint elements by their ID.
    * Segments starting with "#" match Slint elements by type name rather than
      by ID.
    * String segments to match the ID may contain placeholder like `foo-{a,b}`
      to match *any* of those substrings (i.e. `foo-a` or `foo-b`).
    * Optionally, a last segment can be supplied to specify the expected number
      of Slint elements matching the query. A star "*" means any number of
      result is allowed (0..n). A question mark "?" means any number of result
      is allowed (0..n), and the query shall not be executed yet (must be done
      by calling `wait()` manually). Omitting the suffix means exactly one
      result (==1) is expected. The method raises an exception if there
      are != 1 elements matching the query.

    The suffix also impacts the API of this class. All functions returning
    a value will return a single value if no suffix is provided. If "*" or "?"
    was provided, all methods return a list of values, according to the
    number of elements matching the query.
    """

    def __init__(self, app, window, query=None):
        self._app = app
        self._window = window
        self._query = query or []
        self._results = None
        self._multi_result = False
        suffix = self._query[-1] if len(self._query) else ""
        if suffix == "?":
            self._query = self._query[:-1]
        elif suffix == "*":
            self._query = self._query[:-1]
            self._multi_result = True
            self.refresh()
        else:
            self.wait(1)

    def __len__(self):
        """
        Get the number of results
        """
        return len(self._results)

    def __getitem__(self, index):
        """
        Get a particular result by index (an `Element` object)
        """
        return self._results[index]

    @property
    def window(self):
        return self._app.windows[self._window]

    def get(self, path):
        """
        Query children of this query's results

        See class documentation for details.
        """
        return ElementQuery(self._app, self._window, self._query + path.split(" "))

    def refresh(self):
        """
        Re-execute the query once (e.g. after the UI has changed)
        """
        root = self.window.root_element
        if not self._query:
            self._results = [root]
        else:
            self._results = list(_flatten(_query_children(root, self._query)))
        self._results = [Element(x) for x in self._results]

    def wait(self, min=1, max=-1, timeout=10.0):
        """
        Re-execute the query in a loop until a specific number of results are
        matching the query

        The arguments `min` and `max` are used to specify the number of
        results to wait for. A `max` value of `-1` means `max==min`, i.e.
        a specific count is expected rather than a range. If `max` is `None`,
        any number of results `>=min` are considered as valid.

        The function aborts after the provided timeout in seconds.
        """
        if max == -1:
            max = min
        start = time.time()
        while True:
            self.refresh()
            count = len(self._results)
            if (count >= min) and ((max is None) or (count <= max)):
                return
            if time.time() - start > timeout:
                self.save_screenshot()
                raise TimeoutError(
                    f"Expecting {min}..{max} results, got {count}: {self._query} "
                    + "-- Screenshot saved."
                )
            time.sleep(0.01)

    def wait_for_item(self, test, timeout=10.0):
        start = time.time()
        while True:
            self.refresh()
            for result in self._results:
                if test(result):
                    return result
            if time.time() - start > timeout:
                self.save_screenshot()
                raise TimeoutError(
                    f"Waiting for item timed out: {self._query} "
                    + "-- Screenshot saved."
                )
            time.sleep(0.01)

    def read(self, keys, indices=None):
        """
        Read several element properties
        """
        if indices is None:
            filtered_results = self._results
        else:
            filtered_results = [self._results[i] for i in indices]
        props = dict()
        for key in keys:
            match key:
                case "valid":
                    props[key] = [r.valid for r in filtered_results]
                case "label":
                    props[key] = [r.label for r in filtered_results]
                case "placeholder":
                    props[key] = [r.placeholder for r in filtered_results]
                case "value":
                    props[key] = [r.value for r in filtered_results]
                case "checked":
                    props[key] = [r.checked for r in filtered_results]
                case "enabled":
                    props[key] = [r.enabled for r in filtered_results]
                case "readonly":
                    props[key] = [r.readonly for r in filtered_results]
                case _:
                    raise ValueError(f"Unknown property: '{key}'")
        return props

    def wait_for(self, timeout=10.0, indices=None, **properties):
        """
        Wait for all query's results to have specific properties set
        """

        def _compare(actual, expected):
            if isinstance(expected, list):
                return actual == expected
            else:
                return all([a == expected for a in actual])

        if ("valid" in properties) and (len(properties) > 1):
            raise ValueError(
                f"Cannot check for other properties in addition to 'valid': {properties}"
            )

        start = time.time()
        while True:
            # In some cases, UI elements temporarily disappear (for example
            # due to model resets). This would invalidate element handles,
            # so we regularly have to re-evaluate the query.
            self.refresh()
            actual_props = self.read(properties.keys(), indices)
            if all([_compare(v, properties[k]) for k, v in actual_props.items()]):
                return
            if time.time() > (start + timeout):
                self.save_screenshot()
                raise TimeoutError(
                    f"Timeout while waiting for {properties}, actual {actual_props}: "
                    + f"{self._query} -- Screenshot saved."
                )
            time.sleep(0.01)

    @property
    def valid(self):
        return self._flatten([x.valid for x in self._results])

    @property
    def rect(self):
        return self._flatten([x.rect for x in self._results])

    @property
    def label(self):
        return self._flatten([x.label for x in self._results])

    @property
    def placeholder(self):
        return self._flatten([x.placeholder for x in self._results])

    @property
    def value(self):
        return self._flatten([x.value for x in self._results])

    @property
    def checked(self):
        return self._flatten([x.checked for x in self._results])

    @property
    def enabled(self):
        return self._flatten([x.enabled for x in self._results])

    def trigger(self):
        for x in self._results:
            x.trigger()

    def click(self):
        for x in self._results:
            x.click()

    def mclick(self):
        for x in self._results:
            x.mclick()

    def dclick(self):
        for x in self._results:
            x.dclick()

    def set_value(self, value):
        for x in self._results:
            x.set_value(value)

    def set_checked(self, checked):
        for x in self._results:
            x.set_checked(checked)

    def save_screenshot(self, fname="screenshot.png"):
        path = os.path.join(ROOT_DIR, fname)
        with open(path, "wb") as f:
            f.write(self.window.grab_window_as_png())

    def _flatten(self, result):
        if self._multi_result or len(result) != 1:
            return result
        else:
            return result[0]


class Application(slint_testing.Application):
    def root(self, window=0):
        return ElementQuery(self, window)

    def get(self, path, window=0):
        return self.root(window=window).get(path)

    def trigger_shortcut(self, keys, window=0):
        win = self.windows[window]
        for key in keys:
            win.dispatch_event(slint_testing.KeyPressedEvent(key))
        for key in keys:
            win.dispatch_event(slint_testing.KeyReleasedEvent(key))

    def drag_and_drop(
        self,
        start,
        end,
        button=slint_testing.PointerEventButton.Left,
        steps=3,
        window=0,
    ):
        self.windows[window].drag_and_drop(
            slint_testing.LogicalPosition(*start),
            slint_testing.LogicalPosition(*end),
            button,
            steps,
        )


class LibrePcbFixture(object):
    def __init__(self, config, tmpdir):
        super(LibrePcbFixture, self).__init__()
        self.executable = os.path.abspath(config.getoption("--librepcb-executable"))
        if not os.path.exists(self.executable):
            raise Exception(
                "Executable '{}' not found. Please pass it with "
                "'--librepcb-executable'.".format(self.executable)
            )
        self.tmpdir = tmpdir
        # Copy test data to temporary directory to avoid modifications in original data
        shutil.copytree(
            _long_path(os.path.join(DATA_DIR, "workspaces", "Empty Workspace")),
            _long_path(os.path.join(self.tmpdir, "Empty Workspace")),
        )
        # Init members to default values
        self.workspace_path = os.path.join(self.tmpdir, "Empty Workspace")
        self.project_path = None

        # Set environment variables
        self.env = deepcopy(os.environ)
        # Make GUI independent from the system's language
        self.env["LC_ALL"] = "C"
        # Override configuration location to make tests independent of existing configs
        self.env["LIBREPCB_CONFIG_DIR"] = os.path.join(self.tmpdir, "config")
        # Override cache location to make each test indepotent
        self.env["LIBREPCB_CACHE_DIR"] = os.path.join(self.tmpdir, "cache")
        # Use a neutral username
        self.env["USERNAME"] = "testuser"
        # Force large enough window size because UI elements are only visible
        # if there is enough space for them.
        self.env["LIBREPCB_WINDOW_SIZE"] = "1600x900"
        # Force LibrePCB to use Qt-style file dialogs because native dialogs don't work
        self.env["LIBREPCB_DISABLE_NATIVE_DIALOGS"] = "1"
        # Disable warning about unstable file format, since tests are run also
        # on the (unstable) master branch
        self.env["LIBREPCB_DISABLE_UNSTABLE_WARNING"] = "1"
        # Disable desktop integration by default, since the notification
        # disturbes some tests.
        self.env["LIBREPCB_ENABLE_DESKTOP_INTEGRATION"] = "0"

    def abspath(self, relpath):
        return os.path.join(self.tmpdir, relpath)

    def get_data_path(self, relpath):
        return os.path.join(DATA_DIR, relpath)

    def set_workspace(self, path):
        if not os.path.isabs(path):
            path = self.abspath(path)
        self.workspace_path = path

    def set_library_update_mode(self, mode):
        assert mode in ["disabled", "check", "notify", "install"]
        old_value = "(library_updates check)"
        fp = self.get_workspace_path("data/settings.lp")
        with open(fp, "r") as f:
            content = f.read()
        assert old_value in content
        content = content.replace(old_value, f"(library_updates {mode})")
        with open(fp, "w") as f:
            f.write(content)

    def set_keyboard_shortcut(self, id, shortcut):
        fp = self.get_workspace_path("data/settings.lp")
        with open(fp, "r") as f:
            lines = f.read().splitlines()
        lines.insert(1, f' (keyboard_shortcuts (shortcut {id} "{shortcut}"))')
        with open(fp, "w") as f:
            f.write("\n".join(lines))

    def add_project(self, project, as_lppz=False):
        src = os.path.join(DATA_DIR, "projects", project)
        dst = os.path.join(self.tmpdir, project)
        if as_lppz:
            shutil.make_archive(dst, "zip", src)
            shutil.move(dst + ".zip", dst + ".lppz")
        else:
            shutil.copytree(_long_path(src), _long_path(dst))

    def set_project(self, path):
        if not os.path.isabs(path):
            path = self.abspath(path)
        self.project_path = path

    def get_workspace_path(self, relpath):
        return os.path.join(self.workspace_path, relpath)

    def get_workspace_libraries_path(self, subdir=""):
        return os.path.join(self.workspace_path, "data", "libraries", subdir)

    def add_local_library_to_workspace(self, path):
        if not os.path.isabs(path):
            path = os.path.join(DATA_DIR, path)
        dest = self.get_workspace_libraries_path("local")
        dest = os.path.join(dest, os.path.basename(path))
        shutil.copytree(_long_path(path), _long_path(dest))

    def add_remote_library_to_workspace(self, zip_path, mark_outdated=False):
        if not os.path.isabs(zip_path):
            zip_path = os.path.join(DATA_DIR, zip_path)
        outdir = self.get_workspace_libraries_path("remote")
        shutil.unpack_archive(_long_path(zip_path), _long_path(outdir))
        dest = os.path.join(outdir, os.path.basename(zip_path).replace("zip", "lplib"))
        with os.scandir(outdir) as it:
            for entry in it:
                if entry.is_dir() and not entry.name.endswith(".lplib"):
                    os.rename(entry.path, dest)
        if mark_outdated:
            lp_file = os.path.join(dest, "library.lp")
            with open(lp_file, "r") as f:
                content = f.read()
            content = re.sub(
                r" \(version \"(\d+\.?)+\"\)", ' (version "0.0.0.1")', content
            )
            with open(lp_file, "w") as f:
                f.write(content)

    def add_project_to_workspace(self, project):
        src = os.path.join(DATA_DIR, "projects", project)
        dst = os.path.join(self.workspace_path, "projects", project)
        shutil.copytree(_long_path(src), _long_path(dst))

    def open(self):
        self._create_application_config_file()
        return Application([self.executable] + self._args(), env=self.env)

    def _create_application_config_file(self):
        org_dir = "LibrePCB.org" if platform.system() == "Darwin" else "LibrePCB"
        config_dir = os.path.join(self.tmpdir, "config", org_dir)
        config_ini = os.path.join(config_dir, "LibrePCB.ini")
        if not os.path.exists(config_dir):
            os.makedirs(config_dir)
        # Only create config file once per test, so tests can check if settings
        # are stored permanently.
        if not os.path.exists(config_ini):
            with open(config_ini, "w") as f:
                if self.workspace_path:
                    f.write("[workspaces]\n")
                    f.write(
                        'most_recently_used="{}"\n'.format(
                            self.workspace_path.replace("\\", "/")
                        )
                    )

    def _args(self):
        args = []
        if self.project_path:
            args.append(self.project_path)
        return args


class Helpers(object):
    @staticmethod
    def wait_for_file_exists(path, timeout=10.0):
        start = time.time()
        while True:
            if os.path.exists(path):
                return
            if time.time() > (start + timeout):
                raise TimeoutError(f"File does not exist: {path}")
            time.sleep(0.02)

    @staticmethod
    def wait_for_directories(parent, directories, timeout=10.0):
        start = time.time()
        directories = set(directories)
        dirs = None
        while True:
            if os.path.exists(parent):
                dirs = set(os.listdir(parent))
                if dirs == directories:
                    return
            if time.time() > (start + timeout):
                raise TimeoutError(
                    f"Expected directories {directories}, but found {dirs}."
                )
            time.sleep(0.02)

    @staticmethod
    def wait_for_notification(app, title, visible=True, timeout=10.0):
        start = time.time()
        while True:
            items = app.get("#NotificationsPopup #NotificationItem *")
            for item in items:
                if visible and (item.label == title):
                    return item
            if not visible:
                return None
            if time.time() > (start + timeout):
                raise TimeoutError(f"Notification not emitted: {title}")
            time.sleep(0.02)


@pytest.fixture(scope="session")
def librepcb_server():
    """
    Fixture which provides a HTTP server at localhost:8080

    All tests should use this server instead of the official LibrePCB API server
    or GitHub for downloading libraries.
    """
    import time
    import threading
    import socket
    import socketserver
    import http.server

    class Handler(http.server.SimpleHTTPRequestHandler, object):
        def translate_path(self, path):
            path = super(Handler, self).translate_path(path)
            relpath = os.path.relpath(path, os.curdir)
            return os.path.join(DATA_DIR, "server", relpath)

    # Set SO_REUSEADDR option to avoid "port already in use" errors
    httpd = socketserver.TCPServer(("", 50080), Handler, bind_and_activate=False)
    httpd.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    httpd.server_bind()
    httpd.server_activate()
    thread = threading.Thread(target=httpd.serve_forever)
    thread.daemon = True
    thread.start()
    time.sleep(0.2)  # wait a bit to make sure the server is ready


@pytest.fixture(scope="session")
def check_resolution(request, tmp_path_factory):
    """
    Fixture which just fails the complete session if the screen resolution is
    not sufficient (some tests require a large enough screen)
    """
    path = tmp_path_factory.mktemp("check_resolution")
    with LibrePcbFixture(request.config, path).open() as app:
        min_size = (1024, 720)
        win_size = app.first_window.size
        if (win_size[0] < min_size[0]) or (win_size[1] < min_size[1]):
            pytest.exit(
                f"Screen resolution is {win_size}, requires minimum {min_size}.",
                returncode=1,
            )


@pytest.fixture
def create_librepcb(request, tmpdir, librepcb_server, check_resolution):
    """
    Fixture allowing to create multiple application instances
    """

    def _create():
        return LibrePcbFixture(request.config, str(tmpdir))

    return _create


@pytest.fixture
def librepcb(create_librepcb):
    """
    Fixture allowing to create one application instance
    """
    yield create_librepcb()


@pytest.fixture(scope="session")
def helpers():
    """
    Fixture providing some helper functions
    """
    return Helpers()