tdesktop/Telegram/ThirdParty/xdg-desktop-portal/tests/conftest.py

# SPDX-License-Identifier: LGPL-2.1-or-later
#
# This file is formatted with Python Black

from typing import Any, Dict, Iterator, Optional
from types import ModuleType

import pytest
import dbus
import dbusmock
import os
import sys
import tempfile
import subprocess
import time
import signal
from pathlib import Path
from contextlib import chdir

import gi

gi.require_version("UMockdev", "1.0")
from gi.repository import UMockdev  # noqa E402


def pytest_configure() -> None:
    ensure_environment_set()
    ensure_umockdev_loaded()


def pytest_sessionfinish(session, exitstatus):
    # Meson and ginsttest-runner expect tests to exit with status 77 if all
    # tests were skipped
    if exitstatus == pytest.ExitCode.NO_TESTS_COLLECTED:
        session.exitstatus = 77


def ensure_environment_set() -> None:
    env_vars = [
        "XDG_DESKTOP_PORTAL_PATH",
        "XDG_PERMISSION_STORE_PATH",
        "XDG_DOCUMENT_PORTAL_PATH",
    ]

    if not os.getenv("XDP_VALIDATE_AUTO"):
        env_vars += [
            "XDP_VALIDATE_ICON",
            "XDP_VALIDATE_SOUND",
        ]
    else:
        os.environ.pop("XDP_VALIDATE_ICON", None)
        os.environ.pop("XDP_VALIDATE_SOUND", None)

    for env_var in env_vars:
        if not os.getenv(env_var):
            raise Exception(f"{env_var} must be set")


def ensure_umockdev_loaded() -> None:
    umockdev_preload = "libumockdev-preload.so"
    preload = os.environ.get("LD_PRELOAD", "")
    if umockdev_preload not in preload:
        os.environ["LD_PRELOAD"] = f"{umockdev_preload}:{preload}"
        os.execv(sys.executable, [sys.executable] + sys.argv)


def test_dir() -> Path:
    return Path(__file__).resolve().parent


@pytest.fixture
def xdg_desktop_portal_path() -> Path:
    return Path(os.environ["XDG_DESKTOP_PORTAL_PATH"])


@pytest.fixture
def xdg_permission_store_path() -> Path:
    return Path(os.environ["XDG_PERMISSION_STORE_PATH"])


@pytest.fixture
def xdg_document_portal_path() -> Path:
    return Path(os.environ["XDG_DOCUMENT_PORTAL_PATH"])


@pytest.fixture(autouse=True)
def create_test_dirs(umockdev: Optional[UMockdev.Testbed]) -> Iterator[None]:
    # The umockdev argument is to make sure the testbed
    # is created before we create the tmpdir
    env_dirs = [
        "HOME",
        "TMPDIR",
        "XDG_CACHE_HOME",
        "XDG_CONFIG_HOME",
        "XDG_DATA_HOME",
        "XDG_RUNTIME_DIR",
        "XDG_DESKTOP_PORTAL_DIR",
    ]

    test_root = tempfile.TemporaryDirectory(
        prefix="xdp-testroot-", ignore_cleanup_errors=True
    )

    for env_dir in env_dirs:
        directory = Path(test_root.name) / env_dir.lower()
        directory.mkdir(mode=0o700, parents=True)
        os.environ[env_dir] = directory.absolute().as_posix()

    yield

    test_root.cleanup()


@pytest.fixture
def xdg_data_home_files() -> Dict[str, bytes]:
    """
    Default fixture which can be used to create files in the temporary
    XDG_DATA_HOME directory of the test.
    """
    return {}


@pytest.fixture(autouse=True)
def ensure_xdg_data_home(
    create_test_dirs: Any, xdg_data_home_files: Dict[str, bytes]
) -> None:
    files = xdg_data_home_files
    for name, content in files.items():
        file_path = Path(os.environ["XDG_DATA_HOME"]) / name
        file_path.parent.mkdir(parents=True, exist_ok=True)
        with open(file_path.absolute().as_posix(), "wb") as f:
            f.write(content)


@pytest.fixture
def xdg_desktop_portal_dir_files() -> Dict[str, bytes]:
    """
    Default fixture which can be used to create files in the temporary
    XDG_DESKTOP_PORTAL_DIR directory of the test.
    """
    return {}


@pytest.fixture
def xdg_desktop_portal_dir_default_files() -> Dict[str, bytes]:
    files = {}

    portals = [
        "org.freedesktop.impl.portal.Access",
        "org.freedesktop.impl.portal.Account",
        "org.freedesktop.impl.portal.AppChooser",
        "org.freedesktop.impl.portal.Background",
        "org.freedesktop.impl.portal.Clipboard",
        "org.freedesktop.impl.portal.DynamicLauncher",
        "org.freedesktop.impl.portal.Email",
        "org.freedesktop.impl.portal.FileChooser",
        "org.freedesktop.impl.portal.GlobalShortcuts",
        "org.freedesktop.impl.portal.Inhibit",
        "org.freedesktop.impl.portal.InputCapture",
        "org.freedesktop.impl.portal.Lockdown",
        "org.freedesktop.impl.portal.Notification",
        "org.freedesktop.impl.portal.Print",
        "org.freedesktop.impl.portal.RemoteDesktop",
        "org.freedesktop.impl.portal.Screenshot",
        "org.freedesktop.impl.portal.Settings",
        "org.freedesktop.impl.portal.Usb",
        "org.freedesktop.impl.portal.Wallpaper",
    ]

    files["test-portals.conf"] = b"""
[preferred]
default=test;
"""

    files["test.portal"] = """
[portal]
DBusName=org.freedesktop.impl.portal.Test
Interfaces={}
""".format(";".join(portals)).encode("utf-8")

    return files


@pytest.fixture(autouse=True)
def ensure_xdg_desktop_portal_dir(
    create_test_dirs: Any,
    xdg_desktop_portal_dir_files: Dict[str, bytes],
    xdg_desktop_portal_dir_default_files: Dict[str, bytes],
) -> None:
    files = xdg_desktop_portal_dir_default_files | xdg_desktop_portal_dir_files
    for name, content in files.items():
        file_path = Path(os.environ["XDG_DESKTOP_PORTAL_DIR"]) / name
        file_path.parent.mkdir(parents=True, exist_ok=True)
        with open(file_path.absolute().as_posix(), "wb") as f:
            f.write(content)


@pytest.fixture(autouse=True)
def create_test_dbus() -> Iterator[dbusmock.DBusTestCase]:
    bus = dbusmock.DBusTestCase()
    bus.setUp()
    bus.start_session_bus()
    bus.start_system_bus()

    yield bus

    bus.tearDown()
    bus.tearDownClass()


@pytest.fixture(autouse=True)
def create_dbus_monitor(create_test_dbus) -> Iterator[Optional[subprocess.Popen]]:
    if not os.getenv("XDP_DBUS_MONITOR"):
        yield None
        return

    dbus_monitor = subprocess.Popen(["dbus-monitor", "--session"])

    yield dbus_monitor

    dbus_monitor.terminate()
    dbus_monitor.wait()


def _get_server_for_module(
    busses: dict[dbusmock.BusType, dict[str, dbusmock.SpawnedMock]],
    module: ModuleType,
    bustype: dbusmock.BusType,
) -> dbusmock.SpawnedMock:
    assert bustype in dbusmock.BusType

    try:
        return busses[bustype][module.BUS_NAME]
    except KeyError:
        server = dbusmock.SpawnedMock.spawn_for_name(
            module.BUS_NAME,
            "/dbusmock",
            dbusmock.OBJECT_MANAGER_IFACE,
            bustype,
            stdout=None,
            stderr=None,
        )

        busses[bustype][module.BUS_NAME] = server
        return server


def _get_main_obj_for_module(
    server: dbusmock.SpawnedMock, module: ModuleType, bustype: dbusmock.BusType
) -> dbusmock.DBusMockObject:
    try:
        server.obj.AddObject(
            module.MAIN_OBJ,
            "com.example.EmptyInterface",
            {},
            [],
            dbus_interface=dbusmock.MOCK_IFACE,
        )
    except Exception:
        pass

    bustype.wait_for_bus_object(module.BUS_NAME, module.MAIN_OBJ)
    bus = bustype.get_connection()
    return bus.get_object(module.BUS_NAME, module.MAIN_OBJ)


def _terminate_mock_p(process: subprocess.Popen) -> None:
    process.terminate()
    process.wait()


def _terminate_servers(
    busses: dict[dbusmock.BusType, dict[str, dbusmock.SpawnedMock]],
) -> None:
    for server in busses[dbusmock.BusType.SYSTEM].values():
        _terminate_mock_p(server.process)
    for server in busses[dbusmock.BusType.SESSION].values():
        _terminate_mock_p(server.process)


def _start_template(
    busses: dict[dbusmock.BusType, dict[str, dbusmock.SpawnedMock]],
    template: str,
    bus_name: Optional[str],
    params: Dict[str, Any] = {},
) -> None:
    """
    Start the template and potentially start a server for it
    """
    module_path_dir = (test_dir()).parent.absolute().as_posix()
    template_path = test_dir() / f"templates/{template.lower()}.py"
    template = template_path.absolute().as_posix()

    # we cd to the parent dir of the test_dir so that the module search path for
    # the templates is the same as opening the modules from here
    with chdir(module_path_dir):
        module = dbusmock.mockobject.load_module(template)
        bustype = (
            dbusmock.BusType.SYSTEM if module.SYSTEM_BUS else dbusmock.BusType.SESSION
        )

        if bus_name:
            module.BUS_NAME = bus_name

        server = _get_server_for_module(busses, module, bustype)
        main_obj = _get_main_obj_for_module(server, module, bustype)

        main_obj.AddTemplate(
            template,
            dbus.Dictionary(params, signature="sv"),
            dbus_interface=dbusmock.MOCK_IFACE,
        )


@pytest.fixture
def template_params() -> dict[str, dict[str, Any]]:
    """
    Default fixture for overriding the parameters which should be passed to the
    mocking templates. Use required_templates to specify the default parameters
    and override it for specific test cases via

        @pytest.mark.parametrize("template_params", ({"Template": {"foo": "bar"}},))

    """
    return {}


@pytest.fixture
def required_templates() -> dict[str, dict[str, Any]]:
    """
    Default fixture for enumerating the mocking templates the test case requires
    to be started. This is a map from a template spec to the parameters which
    should be passed to the template. The template spec is the name of a
    template in the templates directory and an optional dbus bus name to start
    the template at, separated by a colon.

    This starts the `settings` and `email` templates on the bus names specified
    with BUS_NAME in the templates and passes parameters to settings.

        {
            "settings": {
                "some_param": true,
            },
            "email": {},
        }

    This starts two instances of the settings template with their own parameters
    once on the bus name specified in BUS_NAME in the template, and once on the
    bus name `org.freedesktop.impl.portal.OtherImpl`.

        {
            "settings": {
                "some_param": true,
            },
            "settings:org.freedesktop.impl.portal.OtherImpl": {
                "some_param": false,
            },
        }

    """
    return {}


@pytest.fixture
def templates(
    required_templates: dict[str, dict[str, Any]],
    template_params: dict[str, dict[str, Any]],
) -> Iterator[None]:
    """
    Fixture which starts the required templates with their parameters. Usually
    the `portals` fixture is what you're looking for because it also starts
    the portal frontend and the permission store.
    """
    busses: dict[dbusmock.BusType, dict[str, dbusmock.SpawnedMock]] = {
        dbusmock.BusType.SYSTEM: {},
        dbusmock.BusType.SESSION: {},
    }
    for template_data, params in required_templates.items():
        template, bus_name = (template_data.split(":") + [None])[:2]
        assert template
        params = template_params.get(template, params)
        _start_template(busses, template, bus_name, params)
    yield
    _terminate_servers(busses)


@pytest.fixture
def xdp_overwrite_env() -> dict[str, str]:
    """
    Default fixture which can be used to override the environment that gets
    passed to xdg-desktop-portal, xdg-document-portal and xdg-permission-store.
    """
    return {}


@pytest.fixture
def app_id() -> str:
    """
    Default fixture which can be used to override the app id that the portal
    frontend will discover for incoming connections.
    """
    return "org.example.Test"


@pytest.fixture
def xdp_env(
    xdp_overwrite_env: dict[str, str],
    app_id: str,
    usb_queries: Optional[str],
    umockdev: Optional[UMockdev.Testbed],
) -> dict[str, str]:
    env = os.environ.copy()
    env["G_DEBUG"] = "fatal-criticals"
    env["XDG_CURRENT_DESKTOP"] = "test"

    if app_id:
        env["XDG_DESKTOP_PORTAL_TEST_APP_ID"] = app_id

    if usb_queries:
        env["XDG_DESKTOP_PORTAL_TEST_USB_QUERIES"] = usb_queries

    if umockdev:
        env["UMOCKDEV_DIR"] = umockdev.get_root_dir()

    asan_suppression = test_dir() / "asan.suppression"
    if not asan_suppression.exists():
        raise FileNotFoundError(f"{asan_suppression} does not exist")
    env["LSAN_OPTIONS"] = f"suppressions={asan_suppression}"

    for key, val in xdp_overwrite_env.items():
        env[key] = val

    return env


def _maybe_add_asan_preload(executable: Path, env: dict[str, str]) -> None:
    # ASAN really wants to be the first library to get loaded but we also
    # LD_PRELOAD umockdev and LD_PRELOAD gets loaded before any "normally"
    # linked libraries. This uses ldd to find the version of libasan.so that
    # should be loaded and puts it in front of LD_PRELOAD.
    # This way, LD_PRELOAD and ASAN can be used at the same time.
    ldd = subprocess.check_output(["ldd", executable]).decode("utf-8")
    libs = [line.split()[0] for line in ldd.splitlines()]
    try:
        libasan = next(filter(lambda lib: lib.startswith("libasan"), libs))
    except StopIteration:
        return

    preload = env.get("LD_PRELOAD", "")
    env["LD_PRELOAD"] = f"{libasan}:{preload}"


@pytest.fixture
def xdg_desktop_portal(
    dbus_con: dbus.Bus, xdg_desktop_portal_path: Path, xdp_env: dict[str, str]
) -> Iterator[subprocess.Popen]:
    """
    Fixture which starts and eventually stops xdg-desktop-portal
    """
    if not xdg_desktop_portal_path.exists():
        raise FileNotFoundError(f"{xdg_desktop_portal_path} does not exist")

    env = xdp_env.copy()
    _maybe_add_asan_preload(xdg_desktop_portal_path, env)

    xdg_desktop_portal = subprocess.Popen([xdg_desktop_portal_path], env=env)

    while not dbus_con.name_has_owner("org.freedesktop.portal.Desktop"):
        returncode = xdg_desktop_portal.poll()
        if returncode is not None:
            raise subprocess.SubprocessError(
                f"xdg-desktop-portal exited with {returncode}"
            )
        time.sleep(0.1)

    yield xdg_desktop_portal

    xdg_desktop_portal.send_signal(signal.SIGHUP)
    returncode = xdg_desktop_portal.wait()
    assert returncode == 0


@pytest.fixture
def xdg_permission_store(
    dbus_con: dbus.Bus, xdg_permission_store_path: Path, xdp_env: dict[str, str]
) -> Iterator[subprocess.Popen]:
    """
    Fixture which starts and eventually stops xdg-permission-store
    """
    if not xdg_permission_store_path.exists():
        raise FileNotFoundError(f"{xdg_permission_store_path} does not exist")

    env = xdp_env.copy()
    _maybe_add_asan_preload(xdg_permission_store_path, env)

    permission_store = subprocess.Popen([xdg_permission_store_path], env=env)

    while not dbus_con.name_has_owner("org.freedesktop.impl.portal.PermissionStore"):
        returncode = permission_store.poll()
        if returncode is not None:
            raise subprocess.SubprocessError(
                f"xdg-permission-store exited with {returncode}"
            )
        time.sleep(0.1)

    yield permission_store

    permission_store.send_signal(signal.SIGHUP)
    permission_store.wait()
    # The permission store does not shut down cleanly currently
    # returncode = permission_store.wait()
    # assert returncode == 0


@pytest.fixture
def xdg_document_portal(
    dbus_con: dbus.Bus, xdg_document_portal_path: Path, xdp_env: dict[str, str]
) -> Iterator[subprocess.Popen]:
    """
    Fixture which starts and eventually stops xdg-document-portal
    """
    if not xdg_document_portal_path.exists():
        raise FileNotFoundError(f"{xdg_document_portal_path} does not exist")

    # FUSE and LD_PRELOAD don't like each other. Not sure what exactly is going
    # wrong but it usually just results in a weird hang that needs SIGKILL
    env = xdp_env.copy()
    env.pop("LD_PRELOAD", None)

    document_portal = subprocess.Popen([xdg_document_portal_path], env=env)

    while not dbus_con.name_has_owner("org.freedesktop.portal.Documents"):
        returncode = document_portal.poll()
        if returncode is not None:
            raise subprocess.SubprocessError(
                f"xdg-document-portal exited with {returncode}"
            )
        time.sleep(0.1)

    yield document_portal

    document_portal.send_signal(signal.SIGHUP)
    returncode = document_portal.wait()
    assert returncode == 0


@pytest.fixture
def portals(templates: Any, xdg_desktop_portal: Any, xdg_permission_store: Any) -> None:
    """
    Fixture which starts the required templates, xdg-desktop-portal,
    xdg-document-portal and xdg-permission-store. Most tests require this.
    """
    return None


@pytest.fixture
def usb_queries() -> Optional[str]:
    """
    Default fixture providing the usb queries the connecting process can
    enumerate
    """
    return None


@pytest.fixture
def umockdev() -> Optional[UMockdev.Testbed]:
    """
    Default fixture providing a umockdev testbed
    """
    return None


@pytest.fixture
def dbus_con(create_test_dbus: dbusmock.DBusTestCase) -> dbus.Bus:
    """
    Default fixture which provides the python-dbus session bus of the test.
    """
    con = create_test_dbus.get_dbus(system_bus=False)
    assert con
    return con


@pytest.fixture
def dbus_con_sys(create_test_dbus: dbusmock.DBusTestCase) -> dbus.Bus:
    """
    Default fixture which provides the python-dbus system bus of the test.
    """
    con_sys = create_test_dbus.get_dbus(system_bus=True)
    assert con_sys
    return con_sys