feat: Added EGS integration #247
No files matched your search
@@ -6,7 +6,7 @@
|
||||
<img src="build-aux/share/icons/hicolor/scalable/apps/ru.linux_gaming.PortProtonQt.svg" width="64">
|
||||
<h1 align="center">PortProtonQt</h1>
|
||||
<a href="https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt/releases"><img src="https://img.shields.io/badge/Downloads-6382-green?style=flat-square" alt="Total Downloads"></a>
|
||||
<p align="center">A modern and convenient interface for managing and launching games from PortProton, Steam, and GOG. It brings libraries together in one place and simplifies launching Windows games on Linux.</p>
|
||||
<p align="center">A modern and convenient interface for managing and launching games from PortProton, Steam, Epic Games Store, and GOG. It brings libraries together in one place and simplifies launching Windows games on Linux.</p>
|
||||
</div>
|
||||
|
||||
### Installation (devel)
|
||||
|
||||
+1
-1
@@ -6,7 +6,7 @@
|
||||
<img src="build-aux/share/icons/hicolor/scalable/apps/ru.linux_gaming.PortProtonQt.svg" width="64">
|
||||
<h1 align="center">PortProtonQt</h1>
|
||||
<a href="https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt/releases"><img src="https://img.shields.io/badge/%D0%97%D0%B0%D0%B3%D1%80%D1%83%D0%B7%D0%BA%D0%B8-6382-green?style=flat-square" alt="Всего загрузок"></a>
|
||||
<p align="center">Современный и удобный интерфейс для управления и запуска игр из PortProton, Steam и GOG. Объединяет библиотеки в одном месте и упрощает запуск Windows-игр на Linux.</p>
|
||||
<p align="center">Современный и удобный интерфейс для управления и запуска игр из PortProton, Steam, Epic Games Store и GOG. Объединяет библиотеки в одном месте и упрощает запуск Windows-игр на Linux.</p>
|
||||
</div>
|
||||
|
||||
### Установка (devel)
|
||||
|
||||
@@ -342,6 +342,10 @@ def main():
|
||||
if app_id.isdigit():
|
||||
gog_launch_uri = args.file_or_url
|
||||
ipc_message = f"gog:{app_id}"
|
||||
elif args.file_or_url and args.file_or_url.startswith("egs://launch/"):
|
||||
app_id = args.file_or_url.removeprefix("egs://launch/")
|
||||
if app_id and "/" not in app_id:
|
||||
ipc_message = f"egs:{app_id}"
|
||||
elif args.file_or_url:
|
||||
theme_store_id = parse_portprotonqt_theme_url(args.file_or_url)
|
||||
if theme_store_id is not None:
|
||||
@@ -582,6 +586,12 @@ def main():
|
||||
window.toggleGame(f"gog://launch/{app_id}")
|
||||
else:
|
||||
logger.warning("Invalid GOG app id via IPC: %s", app_id)
|
||||
elif msg.startswith("egs:"):
|
||||
app_id = msg[4:].strip()
|
||||
if app_id and "/" not in app_id:
|
||||
window.toggleGame(f"egs://launch/{app_id}")
|
||||
else:
|
||||
logger.warning("Invalid Epic app id via IPC: %s", app_id)
|
||||
elif msg.startswith("backup:"):
|
||||
parts = msg.split(":", 2)
|
||||
if len(parts) == 3:
|
||||
|
||||
@@ -471,12 +471,12 @@ class ContextMenuManager:
|
||||
return
|
||||
|
||||
# Normal menu for games with valid exe or from Steam
|
||||
is_uninstalled_gog = (
|
||||
game_card.game_source == "gog"
|
||||
and game_card.exec_line.startswith("gog://install/")
|
||||
is_uninstalled_store_game = (
|
||||
game_card.game_source in ("gog", "egs")
|
||||
and game_card.exec_line.startswith(("gog://install/", "egs://install/"))
|
||||
)
|
||||
is_running = self._is_game_running(game_card)
|
||||
if not is_uninstalled_gog:
|
||||
if not is_uninstalled_store_game:
|
||||
action_text = _("Stop Game") if is_running else _("Launch Game")
|
||||
action_icon = "stop" if is_running else "play"
|
||||
launch_action = menu.addAction(self._get_safe_icon(action_icon), action_text)
|
||||
@@ -545,6 +545,44 @@ class ContextMenuManager:
|
||||
lambda: self.parent._import_gog_game(game)
|
||||
)
|
||||
|
||||
if game_card.game_source == "egs":
|
||||
app_id = str(game_card.appid)
|
||||
if game_card.exec_line.startswith("egs://launch/"):
|
||||
update_action = menu.addAction(
|
||||
self._get_safe_icon("update"), _("Update")
|
||||
)
|
||||
update_action.triggered.connect(
|
||||
lambda: self.parent._update_egs_game(app_id)
|
||||
)
|
||||
repair_action = menu.addAction(
|
||||
self._get_safe_icon("update"), _("Repair")
|
||||
)
|
||||
repair_action.triggered.connect(
|
||||
lambda: self.parent._repair_egs_game(app_id)
|
||||
)
|
||||
eos_enabled = self.parent._is_egs_overlay_enabled(app_id)
|
||||
eos_action = menu.addAction(
|
||||
self._get_safe_icon("update"),
|
||||
_("Disable EOS Overlay")
|
||||
if eos_enabled else _("Enable EOS Overlay"),
|
||||
)
|
||||
eos_action.triggered.connect(
|
||||
lambda: self.parent._enable_egs_overlay(app_id)
|
||||
)
|
||||
delete_action = menu.addAction(
|
||||
self._get_safe_icon("delete"), _("Delete")
|
||||
)
|
||||
delete_action.triggered.connect(
|
||||
lambda: self.parent._delete_egs_game(app_id)
|
||||
)
|
||||
else:
|
||||
import_action = menu.addAction(
|
||||
self._get_safe_icon("folder"), _("Import")
|
||||
)
|
||||
import_action.triggered.connect(
|
||||
lambda: self.parent._import_egs_game(app_id)
|
||||
)
|
||||
|
||||
if game_card.game_source == "steam":
|
||||
desktop_dir = QStandardPaths.writableLocation(
|
||||
QStandardPaths.StandardLocation.DesktopLocation
|
||||
|
||||
@@ -86,6 +86,9 @@ class DetailPageManager:
|
||||
self._gog_size_requests: dict[
|
||||
str, tuple[QProcess, list[list[Callable[[], QLabel | None]]]]
|
||||
] = {}
|
||||
self._egs_size_requests: dict[
|
||||
str, tuple[QProcess, list[list[Callable[[], QLabel | None]]]]
|
||||
] = {}
|
||||
|
||||
def openGameDetailPage(self, game_data: dict) -> None:
|
||||
"""Open detailed game information page."""
|
||||
@@ -310,6 +313,7 @@ class DetailPageManager:
|
||||
game_info_layout.addLayout(first_row)
|
||||
|
||||
self._setup_gog_size_data(game_data, game_info_layout)
|
||||
self._setup_egs_size_data(game_data, game_info_layout)
|
||||
|
||||
hltb_layout = QHBoxLayout()
|
||||
hltb_layout.setSpacing(10)
|
||||
@@ -365,6 +369,75 @@ class DetailPageManager:
|
||||
)
|
||||
process.start()
|
||||
|
||||
def _setup_egs_size_data(
|
||||
self, game_data: dict, game_info_layout: QVBoxLayout
|
||||
) -> None:
|
||||
exec_line = str(game_data.get("exec_line", ""))
|
||||
if str(game_data.get("game_source", "")).lower() != "egs":
|
||||
return
|
||||
if not exec_line.startswith("egs://install/"):
|
||||
return
|
||||
row = QHBoxLayout()
|
||||
values = []
|
||||
for title_text in (_("Download"), _("Size")):
|
||||
title = QLabel(title_text.upper())
|
||||
title.setStyleSheet(self.main_window.theme.LAST_LAUNCH_TITLE_STYLE)
|
||||
value = QLabel("—")
|
||||
value.setStyleSheet(self.main_window.theme.LAST_LAUNCH_VALUE_STYLE)
|
||||
row.addWidget(title)
|
||||
row.addWidget(value)
|
||||
row.addSpacing(30)
|
||||
values.append(weakref.ref(value))
|
||||
game_info_layout.addLayout(row)
|
||||
app_id = exec_line.rsplit("/", 1)[-1]
|
||||
api = self.main_window.egs_api
|
||||
cached = api.get_cached_download_sizes(app_id)
|
||||
if cached is not None:
|
||||
self._set_gog_size_values(values, cached)
|
||||
return
|
||||
request = self._egs_size_requests.get(app_id)
|
||||
if request is not None:
|
||||
request[1].append(values)
|
||||
return
|
||||
try:
|
||||
command = api.build_command(["info", app_id, "--json", "--platform", "Windows"])
|
||||
except (FileNotFoundError, OSError) as error:
|
||||
logger.warning("Failed to get EGS download size: %s", error)
|
||||
return
|
||||
process = QProcess(self.main_window)
|
||||
process.setProgram(command[0])
|
||||
process.setArguments(command[1:])
|
||||
environment = QProcessEnvironment.systemEnvironment()
|
||||
environment.insert("LEGENDARY_CONFIG_PATH", str(api.config_dir))
|
||||
process.setProcessEnvironment(environment)
|
||||
self._egs_size_requests[app_id] = (process, [values])
|
||||
process.finished.connect(
|
||||
lambda code, _status: self._finish_egs_size_request(app_id, code)
|
||||
)
|
||||
process.start()
|
||||
|
||||
def _finish_egs_size_request(self, app_id: str, code: int) -> None:
|
||||
request = self._egs_size_requests.pop(app_id, None)
|
||||
if request is None:
|
||||
return
|
||||
process, value_groups = request
|
||||
if code != 0:
|
||||
logger.warning("legendary info exited with code %s", code)
|
||||
process.deleteLater()
|
||||
return
|
||||
try:
|
||||
sizes = self.main_window.egs_api.parse_download_sizes(
|
||||
bytes(process.readAllStandardOutput().data())
|
||||
)
|
||||
except (ValueError, TypeError) as error:
|
||||
logger.warning("Failed to parse EGS download size: %s", error)
|
||||
process.deleteLater()
|
||||
return
|
||||
self.main_window.egs_api.save_download_sizes(app_id, sizes)
|
||||
for value_refs in value_groups:
|
||||
self._set_gog_size_values(value_refs, sizes)
|
||||
process.deleteLater()
|
||||
|
||||
def _finish_gog_size_request(self, app_id: str, code: int) -> None:
|
||||
request = self._gog_size_requests.pop(app_id, None)
|
||||
if request is None:
|
||||
@@ -531,7 +604,9 @@ class DetailPageManager:
|
||||
|
||||
current_exe = self._get_current_exe(exec_line)
|
||||
play_button = self._create_play_button(exec_line, current_exe)
|
||||
if str(game_source).lower() == "gog" and exec_line.startswith("gog://install/"):
|
||||
source = str(game_source).lower()
|
||||
install_uri = exec_line.startswith(f"{source}://install/")
|
||||
if source in {"gog", "egs"} and install_uri:
|
||||
play_button.setText(_("Install"))
|
||||
play_button.setIcon(
|
||||
self.main_window.theme_manager.get_icon("update", as_path=True)
|
||||
@@ -570,7 +645,7 @@ class DetailPageManager:
|
||||
)
|
||||
buttons_layout.addWidget(open_folder_button)
|
||||
elif (
|
||||
str(game_source).lower() != "gog"
|
||||
str(game_source).lower() not in {"gog", "egs"}
|
||||
and not self._has_game_shortcut(game_name)
|
||||
):
|
||||
add_button = self._make_action_button(
|
||||
@@ -588,21 +663,26 @@ class DetailPageManager:
|
||||
if str(game_source).lower() == "portproton":
|
||||
self._add_portproton_buttons(buttons_layout, exec_line)
|
||||
elif (
|
||||
str(game_source).lower() == "gog"
|
||||
source in {"gog", "egs"}
|
||||
and appid
|
||||
and exec_line.startswith("gog://launch/")
|
||||
and exec_line.startswith(f"{source}://launch/")
|
||||
):
|
||||
gog_exe = self.main_window.gog_api.get_launch_target(str(appid))
|
||||
if not gog_exe:
|
||||
store_api = (
|
||||
self.main_window.gog_api if source == "gog"
|
||||
else self.main_window.egs_api
|
||||
)
|
||||
store_exe = store_api.get_launch_target(str(appid))
|
||||
if not store_exe:
|
||||
return buttons_layout
|
||||
self.main_window.gog_api.ensure_launch_parameters(str(appid))
|
||||
if source == "gog":
|
||||
store_api.ensure_launch_parameters(str(appid))
|
||||
settings_button = self._make_action_button(
|
||||
_("Settings"),
|
||||
self.main_window.theme_manager.get_icon("settings", as_path=True),
|
||||
)
|
||||
settings_button.clicked.connect(
|
||||
lambda: self.main_window.open_exe_settings(
|
||||
gog_exe, game_source="gog",
|
||||
store_exe, game_source=source,
|
||||
)
|
||||
)
|
||||
buttons_layout.addWidget(settings_button)
|
||||
@@ -610,7 +690,9 @@ class DetailPageManager:
|
||||
_("Create Log"),
|
||||
self.main_window.theme_manager.get_icon("edit", as_path=True),
|
||||
)
|
||||
log_button.clicked.connect(lambda: self.toggleDebugLog(gog_exe, log_button))
|
||||
log_button.clicked.connect(
|
||||
lambda: self.toggleDebugLog(store_exe, log_button)
|
||||
)
|
||||
buttons_layout.addWidget(log_button)
|
||||
self._debug_log_button = log_button
|
||||
folder_button = self._make_action_button(
|
||||
@@ -618,7 +700,7 @@ class DetailPageManager:
|
||||
self.main_window.theme_manager.get_icon("search", as_path=True),
|
||||
)
|
||||
folder_button.clicked.connect(
|
||||
lambda: self._open_executable_folder(gog_exe)
|
||||
lambda: self._open_executable_folder(store_exe)
|
||||
)
|
||||
buttons_layout.addWidget(folder_button)
|
||||
elif str(game_source).lower() == "steam" and appid:
|
||||
|
||||
@@ -140,7 +140,7 @@ def _add_source_corner(
|
||||
) -> None:
|
||||
"""Add source corner badge to cover frame."""
|
||||
source = game_source.lower()
|
||||
if source not in ("steam", "gog", "portproton") or theme_manager is None:
|
||||
if source not in ("steam", "gog", "egs", "portproton") or theme_manager is None:
|
||||
return
|
||||
|
||||
icon_name = f"badge_{source}"
|
||||
|
||||
@@ -0,0 +1,438 @@
|
||||
"""Epic Games Store integration through Legendary."""
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import orjson
|
||||
import requests
|
||||
|
||||
from portprotonqt.localization import get_store_content_languages
|
||||
from portprotonqt.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
EGS_LOGIN_URL = "https://legendary.gl/epiclogin"
|
||||
LEGENDARY_RELEASE_URL = (
|
||||
"https://api.github.com/repos/Heroic-Games-Launcher/legendary/releases/latest"
|
||||
)
|
||||
LEGENDARY_TIMEOUT = 120
|
||||
EGS_METADATA_WORKERS = 4
|
||||
EGS_METADATA_TIMEOUT = 15
|
||||
EGS_DESCRIPTION_CACHE_VERSION = 2
|
||||
LEGENDARY_UPDATE_INTERVAL = 30 * 24 * 60 * 60
|
||||
EGS_API_USER_AGENT = (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
"AppleWebKit/537.36 EpicGamesLauncher"
|
||||
)
|
||||
|
||||
|
||||
class EGSAPI:
|
||||
"""Provide EGS authentication, library and installation helpers."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
data_home = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local/share"))
|
||||
self.data_dir = data_home / "PortProtonQt" / "egs"
|
||||
self.config_dir = self.data_dir / "legendary"
|
||||
self.user_path = self.config_dir / "user.json"
|
||||
self.library_path = self.data_dir / "library.json"
|
||||
self.sizes_path = self.data_dir / "sizes.json"
|
||||
self.version_path = self.data_dir / "bin/legendary.version"
|
||||
self.games_dir = Path.home() / "Games"
|
||||
|
||||
def get_legendary_path(self) -> str | None:
|
||||
"""Return an available Legendary executable."""
|
||||
bundled = self.data_dir / "bin/legendary"
|
||||
if bundled.is_file() and os.access(bundled, os.X_OK):
|
||||
return str(bundled)
|
||||
return shutil.which("legendary")
|
||||
|
||||
def ensure_legendary(self) -> str:
|
||||
"""Download the current official Heroic Legendary binary if needed."""
|
||||
existing = self.get_legendary_path()
|
||||
if existing:
|
||||
return existing
|
||||
asset, release_tag = self._get_latest_legendary_release()
|
||||
return self._download_legendary(asset, release_tag)
|
||||
|
||||
def update_legendary(self) -> str:
|
||||
"""Update the bundled Legendary binary when a newer release exists."""
|
||||
bundled = self.data_dir / "bin/legendary"
|
||||
if bundled.is_file() and os.access(bundled, os.X_OK):
|
||||
try:
|
||||
last_check = self.version_path.stat().st_mtime
|
||||
except OSError:
|
||||
last_check = 0
|
||||
if time.time() - last_check < LEGENDARY_UPDATE_INTERVAL:
|
||||
return str(bundled)
|
||||
asset, release_tag = self._get_latest_legendary_release()
|
||||
if (
|
||||
bundled.is_file()
|
||||
and self._get_legendary_release_tag() == release_tag
|
||||
):
|
||||
self.version_path.touch()
|
||||
return str(bundled)
|
||||
return self._download_legendary(asset, release_tag)
|
||||
|
||||
def _get_latest_legendary_release(self) -> tuple[dict, str]:
|
||||
"""Return the matching Legendary asset and release tag."""
|
||||
architecture = {"x86_64": "x86_64", "aarch64": "arm64"}.get(
|
||||
platform.machine()
|
||||
)
|
||||
if not architecture:
|
||||
raise OSError(f"Unsupported Legendary architecture: {platform.machine()}")
|
||||
response = requests.get(LEGENDARY_RELEASE_URL, timeout=15)
|
||||
response.raise_for_status()
|
||||
release = response.json()
|
||||
asset_name = f"legendary_linux_{architecture}"
|
||||
asset = next(
|
||||
(item for item in release.get("assets", []) if item.get("name") == asset_name),
|
||||
None,
|
||||
)
|
||||
if not asset:
|
||||
raise OSError(f"Legendary release asset not found: {asset_name}")
|
||||
release_tag = str(release.get("tag_name", ""))
|
||||
if not release_tag:
|
||||
raise OSError("Legendary release tag not found")
|
||||
return asset, release_tag
|
||||
|
||||
def _get_legendary_release_tag(self) -> str:
|
||||
try:
|
||||
return self.version_path.read_text(encoding="utf-8").strip()
|
||||
except OSError:
|
||||
return ""
|
||||
|
||||
def _download_legendary(self, asset: dict, release_tag: str) -> str:
|
||||
target = self.data_dir / "bin/legendary"
|
||||
temporary = target.with_suffix(".part")
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
digest = hashlib.sha256()
|
||||
try:
|
||||
with requests.get(asset["browser_download_url"], stream=True, timeout=30) as response:
|
||||
response.raise_for_status()
|
||||
with open(temporary, "wb") as output:
|
||||
for chunk in response.iter_content(chunk_size=1024 * 1024):
|
||||
if chunk:
|
||||
output.write(chunk)
|
||||
digest.update(chunk)
|
||||
expected = str(asset.get("digest", "")).removeprefix("sha256:")
|
||||
if not expected or digest.hexdigest() != expected:
|
||||
raise OSError("Legendary checksum verification failed")
|
||||
temporary.chmod(0o755)
|
||||
temporary.replace(target)
|
||||
self.version_path.write_text(release_tag, encoding="utf-8")
|
||||
return str(target)
|
||||
except Exception:
|
||||
temporary.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
def build_command(self, arguments: list[str]) -> list[str]:
|
||||
"""Build a Legendary command using private configuration paths."""
|
||||
executable = self.get_legendary_path()
|
||||
if not executable:
|
||||
raise FileNotFoundError("Legendary executable not found")
|
||||
self.config_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.config_dir.chmod(0o700)
|
||||
command_arguments = list(arguments)
|
||||
if "-y" in command_arguments:
|
||||
command_arguments.remove("-y")
|
||||
command_arguments.insert(0, "-y")
|
||||
return [executable, *command_arguments]
|
||||
|
||||
def get_environment(self) -> dict[str, str]:
|
||||
"""Return the environment required by Legendary."""
|
||||
environment = os.environ.copy()
|
||||
environment["LEGENDARY_CONFIG_PATH"] = str(self.config_dir)
|
||||
return environment
|
||||
|
||||
def authenticate(self, code: str) -> tuple[bool, str]:
|
||||
"""Authenticate an Epic account with an authorization code."""
|
||||
self.ensure_legendary()
|
||||
result = subprocess.run(
|
||||
self.build_command(["auth", "--code", code]),
|
||||
env=self.get_environment(), capture_output=True, check=False,
|
||||
timeout=LEGENDARY_TIMEOUT,
|
||||
)
|
||||
if result.returncode == 0 and self.user_path.is_file():
|
||||
self.user_path.chmod(0o600)
|
||||
return True, ""
|
||||
error = result.stderr.decode(errors="replace").strip()
|
||||
logger.error("Legendary authentication failed: %s", error)
|
||||
return False, error or f"Legendary exited with code {result.returncode}"
|
||||
|
||||
@staticmethod
|
||||
def extract_auth_code(text: str) -> str:
|
||||
"""Extract an Epic authorization code from copied text or JSON."""
|
||||
value = text.strip()
|
||||
try:
|
||||
data = orjson.loads(value)
|
||||
except orjson.JSONDecodeError:
|
||||
return value if "://" not in value else ""
|
||||
if not isinstance(data, dict):
|
||||
return ""
|
||||
return str(data.get("authorizationCode", "")).strip()
|
||||
|
||||
def logout(self) -> bool:
|
||||
"""Remove the Epic login and cached library."""
|
||||
try:
|
||||
if self.get_legendary_path() and self.user_path.is_file():
|
||||
subprocess.run(
|
||||
self.build_command(["auth", "--delete"]),
|
||||
env=self.get_environment(), capture_output=True, check=False,
|
||||
timeout=LEGENDARY_TIMEOUT,
|
||||
)
|
||||
self.user_path.unlink(missing_ok=True)
|
||||
self.library_path.unlink(missing_ok=True)
|
||||
return True
|
||||
except (OSError, subprocess.SubprocessError) as error:
|
||||
logger.error("Failed to disconnect Epic account: %s", error)
|
||||
return False
|
||||
|
||||
def get_account_name(self) -> str:
|
||||
"""Return the Epic display name from Legendary state."""
|
||||
user = self._load_json(self.user_path, {})
|
||||
return str(user.get("displayName", "")) if isinstance(user, dict) else ""
|
||||
|
||||
def refresh_library(self) -> list[dict]:
|
||||
"""Refresh and cache the Epic library through Legendary."""
|
||||
self.update_legendary()
|
||||
result = subprocess.run(
|
||||
self.build_command(["list", "--json", "--third-party"]),
|
||||
env=self.get_environment(), capture_output=True, check=False,
|
||||
timeout=LEGENDARY_TIMEOUT,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
error = result.stderr.decode(errors="replace").strip()
|
||||
raise OSError(error or "Failed to refresh Epic library")
|
||||
raw_games = orjson.loads(result.stdout)
|
||||
if not isinstance(raw_games, list):
|
||||
raise OSError("Legendary returned an invalid Epic library")
|
||||
games = [self._normalize_game(game) for game in raw_games]
|
||||
games = [game for game in games if game.get("app_id")]
|
||||
self._enrich_descriptions(games)
|
||||
self._save_json(self.library_path, games)
|
||||
return games
|
||||
|
||||
def load_library(self) -> list[dict]:
|
||||
"""Load the cached Epic library."""
|
||||
data = self._load_json(self.library_path, [])
|
||||
return data if isinstance(data, list) else []
|
||||
|
||||
def clear_library_cache(self) -> None:
|
||||
"""Remove cached Epic library metadata."""
|
||||
try:
|
||||
self.library_path.unlink(missing_ok=True)
|
||||
except OSError as error:
|
||||
logger.warning("Failed to clear Epic library cache: %s", error)
|
||||
|
||||
def load_installed(self) -> dict[str, dict]:
|
||||
"""Load Legendary installed game records."""
|
||||
data = self._load_json(self.config_dir / "installed.json", {})
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
def is_game_installed(self, app_id: str) -> bool:
|
||||
"""Return whether Legendary records a usable installation."""
|
||||
return self.get_launch_target(app_id) is not None
|
||||
|
||||
@staticmethod
|
||||
def is_eos_overlay_enabled(prefix_path: Path) -> bool:
|
||||
"""Return whether the EOS overlay is enabled in a Wine prefix."""
|
||||
try:
|
||||
registry = (prefix_path / "user.reg").read_text(
|
||||
encoding="utf-8", errors="replace"
|
||||
)
|
||||
except OSError:
|
||||
return False
|
||||
section_start = registry.find(r"[Software\\Epic Games\\EOS]")
|
||||
if section_start < 0:
|
||||
return False
|
||||
section_end = registry.find("\n[", section_start + 1)
|
||||
section = registry[section_start:section_end if section_end >= 0 else None]
|
||||
return '"OverlayPath"=' in section
|
||||
|
||||
def get_launch_target(self, app_id: str) -> str | None:
|
||||
"""Return the installed game's Windows executable."""
|
||||
installed = self.load_installed().get(app_id, {})
|
||||
install_path = Path(str(installed.get("install_path", "")))
|
||||
executable = str(installed.get("executable", "")).replace("\\", "/")
|
||||
target = install_path.joinpath(*executable.split("/"))
|
||||
return str(target) if executable and target.is_file() else None
|
||||
|
||||
def get_cached_download_sizes(self, app_id: str) -> tuple[int, int] | None:
|
||||
"""Return cached EGS download and disk sizes."""
|
||||
cache = self._load_json(self.sizes_path, {})
|
||||
value = cache.get(app_id) if isinstance(cache, dict) else None
|
||||
if not isinstance(value, list) or len(value) != 2:
|
||||
return None
|
||||
sizes = int(value[0]), int(value[1])
|
||||
return None if sizes[0] > 0 and sizes[1] <= 0 else sizes
|
||||
|
||||
def save_download_sizes(self, app_id: str, sizes: tuple[int, int]) -> None:
|
||||
"""Cache EGS download and disk sizes."""
|
||||
cache = self._load_json(self.sizes_path, {})
|
||||
if not isinstance(cache, dict):
|
||||
cache = {}
|
||||
cache[app_id] = list(sizes)
|
||||
self._save_json(self.sizes_path, cache)
|
||||
|
||||
@staticmethod
|
||||
def parse_download_sizes(output: bytes) -> tuple[int, int]:
|
||||
"""Extract download and install sizes from Legendary JSON."""
|
||||
data = orjson.loads(output)
|
||||
values: dict[str, int] = {}
|
||||
|
||||
def visit(value: Any) -> None:
|
||||
if isinstance(value, dict):
|
||||
for key, item in value.items():
|
||||
if key in {"download_size", "install_size", "disk_size"}:
|
||||
try:
|
||||
values[key] = max(values.get(key, 0), int(item))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
visit(item)
|
||||
elif isinstance(value, list):
|
||||
for item in value:
|
||||
visit(item)
|
||||
|
||||
visit(data)
|
||||
install_size = values.get("install_size") or values.get("disk_size", 0)
|
||||
return values.get("download_size", 0), install_size
|
||||
|
||||
def _enrich_descriptions(self, games: list[dict]) -> None:
|
||||
cached = {str(game.get("app_id")): game for game in self.load_library()}
|
||||
pending = []
|
||||
preferred_locale = get_store_content_languages()[0]
|
||||
for game in games:
|
||||
old_game = cached.get(game["app_id"], {})
|
||||
old_description = str(old_game.get("description", ""))
|
||||
if (
|
||||
old_description
|
||||
and old_description != game["title"]
|
||||
and old_game.get("description_preference") == preferred_locale
|
||||
and old_game.get("description_cache_version")
|
||||
== EGS_DESCRIPTION_CACHE_VERSION
|
||||
):
|
||||
game["description"] = old_description
|
||||
game["description_locale"] = str(old_game.get("description_locale", ""))
|
||||
game["description_preference"] = preferred_locale
|
||||
game["description_cache_version"] = EGS_DESCRIPTION_CACHE_VERSION
|
||||
else:
|
||||
pending.append(game)
|
||||
if not pending:
|
||||
return
|
||||
worker_count = min(EGS_METADATA_WORKERS, os.cpu_count() or 1)
|
||||
with ThreadPoolExecutor(max_workers=worker_count) as executor:
|
||||
descriptions = executor.map(self._get_store_description, pending)
|
||||
for game, result in zip(pending, descriptions, strict=True):
|
||||
description, description_locale = result
|
||||
if description:
|
||||
game["description"] = description
|
||||
game["description_locale"] = description_locale
|
||||
game["description_preference"] = preferred_locale
|
||||
game["description_cache_version"] = EGS_DESCRIPTION_CACHE_VERSION
|
||||
|
||||
def _get_store_description(self, game: dict) -> tuple[str, str]:
|
||||
slug = self._get_product_slug(str(game.get("namespace", "")), game["title"])
|
||||
for language in get_store_content_languages():
|
||||
try:
|
||||
response = requests.get(
|
||||
"https://store-content.ak.epicgames.com/api/"
|
||||
f"{language}/content/products/{slug}",
|
||||
headers={"User-Agent": EGS_API_USER_AGENT},
|
||||
timeout=EGS_METADATA_TIMEOUT,
|
||||
)
|
||||
if response.status_code == requests.codes.not_found:
|
||||
continue
|
||||
response.raise_for_status()
|
||||
description = self._parse_store_description(response.json())
|
||||
if description:
|
||||
return description, language
|
||||
except requests.RequestException as error:
|
||||
logger.debug(
|
||||
"Epic description unavailable for %s: %s", game["app_id"], error
|
||||
)
|
||||
return "", ""
|
||||
return "", ""
|
||||
|
||||
@staticmethod
|
||||
def _parse_store_description(data: Any) -> str:
|
||||
if not isinstance(data, dict) or not isinstance(data.get("pages"), list):
|
||||
return ""
|
||||
page = next(
|
||||
(item for item in data["pages"]
|
||||
if isinstance(item, dict) and item.get("type") == "productHome"),
|
||||
{},
|
||||
)
|
||||
about = page.get("data", {}).get("about", {})
|
||||
short_description = str(about.get("shortDescription") or "").strip()
|
||||
if short_description:
|
||||
return short_description
|
||||
description = str(about.get("description") or "").strip()
|
||||
return description.split("\n\n", 1)[0].strip()
|
||||
|
||||
def _get_product_slug(self, namespace: str, title: str) -> str:
|
||||
fallback = "-".join("".join(
|
||||
character if character.isascii() and character.isalnum() else " "
|
||||
for character in title.lower()
|
||||
).split())
|
||||
if not namespace:
|
||||
return fallback
|
||||
query = {
|
||||
"query": "query ProductSlug($namespace: String!) { Catalog { "
|
||||
"catalogNs(namespace: $namespace) { mappings(pageType: "
|
||||
"\"productHome\") { pageSlug pageType } } } }",
|
||||
"variables": {"namespace": namespace},
|
||||
}
|
||||
try:
|
||||
response = requests.post(
|
||||
"https://launcher.store.epicgames.com/graphql",
|
||||
json=query,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": EGS_API_USER_AGENT,
|
||||
},
|
||||
timeout=EGS_METADATA_TIMEOUT,
|
||||
)
|
||||
response.raise_for_status()
|
||||
mappings = response.json()["data"]["Catalog"]["catalogNs"]["mappings"]
|
||||
return str(mappings[0].get("pageSlug") or fallback) if mappings else fallback
|
||||
except (KeyError, IndexError, TypeError, requests.RequestException) as error:
|
||||
logger.debug("Epic product slug unavailable for %s: %s", namespace, error)
|
||||
return fallback
|
||||
|
||||
@staticmethod
|
||||
def _normalize_game(game: dict) -> dict:
|
||||
metadata = game.get("metadata", {})
|
||||
images = metadata.get("keyImages", []) if isinstance(metadata, dict) else []
|
||||
cover = next(
|
||||
(str(image.get("url", "")) for image in images
|
||||
if image.get("type") in ("DieselGameBoxTall", "OfferImageTall")),
|
||||
"",
|
||||
)
|
||||
return {
|
||||
"app_id": str(game.get("app_name") or game.get("appName") or ""),
|
||||
"title": str(game.get("app_title") or metadata.get("title") or ""),
|
||||
"description": str(metadata.get("description", "")),
|
||||
"cover": cover,
|
||||
"namespace": str(metadata.get("namespace", "")),
|
||||
"description_locale": "",
|
||||
"description_preference": "",
|
||||
"description_cache_version": 0,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _load_json(path: Path, default: Any) -> Any:
|
||||
try:
|
||||
return orjson.loads(path.read_bytes())
|
||||
except (OSError, orjson.JSONDecodeError):
|
||||
return default
|
||||
|
||||
@staticmethod
|
||||
def _save_json(path: Path, data: Any) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(orjson.dumps(data, option=orjson.OPT_INDENT_2))
|
||||
@@ -273,6 +273,7 @@ class GameCard(QFrame):
|
||||
|
||||
self.steam_visible = str(game_source).lower() == "steam"
|
||||
self.gog_visible = str(game_source).lower() == "gog"
|
||||
self.egs_visible = str(game_source).lower() == "egs"
|
||||
self.portproton_visible = str(game_source).lower() == "portproton"
|
||||
self.ppdb_visible = bool(self.ppdb_id) and not self.economy_mode
|
||||
|
||||
@@ -321,8 +322,8 @@ class GameCard(QFrame):
|
||||
self.coverLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self.coverLabel.setStyleSheet(self.theme.COVER_LABEL_STYLE)
|
||||
if self.missing_executable_path or (
|
||||
str(self.game_source).lower() == "gog"
|
||||
and self.exec_line.startswith("gog://install/")
|
||||
str(self.game_source).lower() in ("gog", "egs")
|
||||
and self.exec_line.startswith(("gog://install/", "egs://install/"))
|
||||
):
|
||||
self.coverOpacity = QGraphicsOpacityEffect(self.coverLabel)
|
||||
self.coverOpacity.setOpacity(self.theme.missing_exe_cover_opacity)
|
||||
@@ -370,6 +371,14 @@ class GameCard(QFrame):
|
||||
)
|
||||
self.gogLabel.setVisible(self.gog_visible)
|
||||
|
||||
egs_icon = self.theme_manager.get_icon("badge_egs", as_path=True)
|
||||
self.egsLabel = SourceCorner(
|
||||
icon=egs_icon,
|
||||
config=corner_config,
|
||||
parent=self.coverWidget,
|
||||
)
|
||||
self.egsLabel.setVisible(self.egs_visible)
|
||||
|
||||
portproton_icon = self.theme_manager.get_icon("badge_portproton", as_path=True)
|
||||
self.portprotonLabel = SourceCorner(
|
||||
icon=portproton_icon,
|
||||
@@ -478,6 +487,7 @@ class GameCard(QFrame):
|
||||
for label, icon_name in (
|
||||
(self.steamLabel, "badge_steam"),
|
||||
(self.gogLabel, "badge_gog"),
|
||||
(self.egsLabel, "badge_egs"),
|
||||
(self.portprotonLabel, "badge_portproton"),
|
||||
):
|
||||
icon = self.theme_manager.get_icon(
|
||||
@@ -657,6 +667,7 @@ class GameCard(QFrame):
|
||||
source_ribbons = [
|
||||
(self.steam_visible, self.steamLabel),
|
||||
(self.gog_visible, self.gogLabel),
|
||||
(self.egs_visible, self.egsLabel),
|
||||
(self.portproton_visible, self.portprotonLabel),
|
||||
]
|
||||
|
||||
@@ -679,6 +690,7 @@ class GameCard(QFrame):
|
||||
self.protondbLabel.raise_()
|
||||
self.portprotonLabel.raise_()
|
||||
self.gogLabel.raise_()
|
||||
self.egsLabel.raise_()
|
||||
self.steamLabel.raise_()
|
||||
except RuntimeError:
|
||||
pass
|
||||
@@ -817,6 +829,7 @@ class GameCard(QFrame):
|
||||
self.economy_mode = ui_config.get_economy_mode()
|
||||
self.steam_visible = str(self.game_source).lower() == "steam"
|
||||
self.gog_visible = str(self.game_source).lower() == "gog"
|
||||
self.egs_visible = str(self.game_source).lower() == "egs"
|
||||
self.portproton_visible = str(self.game_source).lower() == "portproton"
|
||||
self.ppdb_visible = bool(self.ppdb_id) and not self.economy_mode
|
||||
protondb_visible = is_valid_protondb_tier(self.protondb_tier) and not self.economy_mode
|
||||
@@ -827,6 +840,7 @@ class GameCard(QFrame):
|
||||
try:
|
||||
self.steamLabel.setVisible(self.steam_visible)
|
||||
self.gogLabel.setVisible(self.gog_visible)
|
||||
self.egsLabel.setVisible(self.egs_visible)
|
||||
self.portprotonLabel.setVisible(self.portproton_visible)
|
||||
self.ppdbLabel.setVisible(self.ppdb_visible and not hidden_badges)
|
||||
self.protondbLabel.setVisible(protondb_visible and not hidden_badges)
|
||||
|
||||
@@ -23,6 +23,7 @@ downloader = Downloader()
|
||||
logger = get_logger(__name__)
|
||||
COVER_IMAGE_EXTENSIONS = (".png", ".apng", ".jpg", ".jpeg", ".gif", ".webp", ".jxl", ".svg")
|
||||
DEFAULT_ANIMATION_DELAY_MS = 100
|
||||
REMOTE_COVER_DOWNLOAD_TIMEOUT = 15
|
||||
|
||||
# Global queue and thread pool for image loading
|
||||
image_load_queue = Queue()
|
||||
@@ -489,7 +490,11 @@ def load_pixmap_async(
|
||||
finish_with(pixmap)
|
||||
|
||||
finish_with(load_fallback_pixmap())
|
||||
downloader.download_async(cover, local_path, timeout=5, callback=on_downloaded)
|
||||
downloader.download_async(
|
||||
cover, local_path,
|
||||
timeout=REMOTE_COVER_DOWNLOAD_TIMEOUT,
|
||||
callback=on_downloaded,
|
||||
)
|
||||
return
|
||||
except Exception as e:
|
||||
logger.error(f"URL processing error {cover}: {e}")
|
||||
@@ -524,7 +529,11 @@ def load_pixmap_async(
|
||||
|
||||
logger.info("Downloading SGDB cover for %s -> %s", app_name or "unknown", filename)
|
||||
finish_with(load_fallback_pixmap())
|
||||
downloader.download_async(cover, local_path, timeout=5, callback=on_downloaded)
|
||||
downloader.download_async(
|
||||
cover, local_path,
|
||||
timeout=REMOTE_COVER_DOWNLOAD_TIMEOUT,
|
||||
callback=on_downloaded,
|
||||
)
|
||||
return
|
||||
|
||||
except Exception as e:
|
||||
@@ -555,7 +564,11 @@ def load_pixmap_async(
|
||||
finish_with(pixmap)
|
||||
|
||||
finish_with(load_fallback_pixmap())
|
||||
downloader.download_async(cover, local_path, timeout=5, callback=on_downloaded)
|
||||
downloader.download_async(
|
||||
cover, local_path,
|
||||
timeout=REMOTE_COVER_DOWNLOAD_TIMEOUT,
|
||||
callback=on_downloaded,
|
||||
)
|
||||
return
|
||||
except Exception as e:
|
||||
logger.error("Error processing remote image URL %s: %s", cover, str(e))
|
||||
|
||||
@@ -36,6 +36,8 @@ LOCALE_MAP = {
|
||||
'el': 'greek',
|
||||
}
|
||||
|
||||
STORE_CONTENT_LOCALES = {'es': 'es-ES', 'pt': 'pt-BR', 'zh': 'zh-CN'}
|
||||
|
||||
# Try local locale directory first, fallback to system for development
|
||||
_local_localedir = Path(__file__).parent / "locales"
|
||||
_system_prefix = "/app" if os.getenv("FLATPAK_ID") else os.getenv("SHARUN_DIR", "/usr")
|
||||
@@ -87,6 +89,23 @@ def get_metadata_language() -> str:
|
||||
|
||||
return 'en'
|
||||
|
||||
def get_store_content_languages() -> tuple[str, ...]:
|
||||
"""Return preferred store content locales with English fallback."""
|
||||
language = get_metadata_language().lower().split('_', 1)[0] or 'en'
|
||||
system_locale = get_system_locale().lower().replace('-', '_')
|
||||
if language == 'zh' and any(
|
||||
variant in system_locale for variant in ('hant', '_tw', '_hk', '_mo')
|
||||
):
|
||||
preferred = 'zh-TW'
|
||||
else:
|
||||
preferred = STORE_CONTENT_LOCALES.get(language, language)
|
||||
languages = [preferred]
|
||||
if language not in languages:
|
||||
languages.append(language)
|
||||
if 'en-US' not in languages:
|
||||
languages.append('en-US')
|
||||
return tuple(languages)
|
||||
|
||||
def read_metadata_translations(metadata_file, language_code):
|
||||
"""
|
||||
Read translations from metadata.txt for the specified language.
|
||||
|
||||
+280
-17
@@ -8,9 +8,14 @@ import subprocess
|
||||
import tempfile
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from queue import Empty, Queue
|
||||
from threading import Thread
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import psutil
|
||||
|
||||
from portprotonqt.logger import get_logger
|
||||
from portprotonqt.icon_extractor import generate_thumbnail, get_exe_icon_cache_path
|
||||
from portprotonqt.dialogs import FileExplorer, ExeSettingsDialog
|
||||
@@ -21,6 +26,7 @@ from portprotonqt.custom_widgets import ClickableLabel, AutoSizeButton, NavLabel
|
||||
from portprotonqt.detail_pages import DetailPageManager
|
||||
from portprotonqt.portproton_api import PortProtonAPI, remove_empty_custom_data_dirs
|
||||
from portprotonqt.gog_api import GOGAPI
|
||||
from portprotonqt.egs_api import EGSAPI
|
||||
from portprotonqt.debug_utils import get_prefix_name
|
||||
from portprotonqt.input_manager import InputManager, MainWindowProtocol
|
||||
from portprotonqt.context_menu_manager import ContextMenuManager
|
||||
@@ -67,7 +73,7 @@ from portprotonqt.sound_manager import SoundManager
|
||||
from portprotonqt.tabs.control_hints import MainWindowControlHintsMixin
|
||||
from portprotonqt.tabs.autoinstall_tab import MainWindowAutoInstallTabMixin
|
||||
from portprotonqt.tabs.library_tab import MainWindowLibraryTabMixin
|
||||
from portprotonqt.tabs.gog_tab import GOGLibraryWorker, MainWindowGOGTabMixin
|
||||
from portprotonqt.tabs.download_tab import GOGLibraryWorker, MainWindowDownloadTabMixin
|
||||
from portprotonqt.tabs.settings_tab import MainWindowSettingsTabMixin
|
||||
from portprotonqt.tabs.system_tab import MainWindowSystemTabMixin
|
||||
from portprotonqt.tabs.theme_tab import MainWindowThemeTabMixin
|
||||
@@ -76,12 +82,13 @@ from portprotonqt.tabs.workers import MainWindowWorkersMixin
|
||||
|
||||
from PySide6.QtWidgets import (QLineEdit, QMainWindow, QWidget, QVBoxLayout, QLabel, QHBoxLayout, QStackedWidget, QComboBox,
|
||||
QMessageBox, QApplication, QPushButton, QCheckBox)
|
||||
from PySide6.QtCore import Qt, QEvent, QEventLoop, QUrl, Signal, QTimer, Slot, QProcess, QThread, QFileSystemWatcher, QObject
|
||||
from PySide6.QtCore import Qt, QEvent, QEventLoop, QUrl, Signal, QTimer, Slot, QProcess, QProcessEnvironment, QThread, QFileSystemWatcher, QObject
|
||||
from PySide6.QtGui import QColor, QDesktopServices, QHideEvent, QShowEvent, QGuiApplication
|
||||
from typing import cast
|
||||
from collections.abc import Callable
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime
|
||||
|
||||
STORE_LAUNCH_GRACE_SECONDS = 10
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from portprotonqt.appimage_updater import AppImageUpdateWorker
|
||||
@@ -144,7 +151,7 @@ class MainWindow(
|
||||
MainWindowControlHintsMixin,
|
||||
MainWindowAutoInstallTabMixin,
|
||||
MainWindowLibraryTabMixin,
|
||||
MainWindowGOGTabMixin,
|
||||
MainWindowDownloadTabMixin,
|
||||
MainWindowSettingsTabMixin,
|
||||
MainWindowSystemTabMixin,
|
||||
MainWindowThemeTabMixin,
|
||||
@@ -375,6 +382,7 @@ class MainWindow(
|
||||
QTimer.singleShot(0, self.start_watching_directories) # Delay to ensure portproton_location is set
|
||||
self.downloader = Downloader(max_workers=4)
|
||||
self.gog_api = GOGAPI()
|
||||
self.egs_api = EGSAPI()
|
||||
self.portproton_api = PortProtonAPI(self.downloader)
|
||||
self.autoInstallCustomDataThread = None
|
||||
|
||||
@@ -961,18 +969,21 @@ class MainWindow(
|
||||
if callable(source_loader):
|
||||
source_loader(lambda games: self.games_loaded.emit(games))
|
||||
elif display_filter == "favorites":
|
||||
def on_all_games_favorites(portproton_games, steam_games, gog_games):
|
||||
games = [game for game in portproton_games + steam_games + gog_games if game[0] in favorites]
|
||||
def on_all_games_favorites():
|
||||
games = [
|
||||
game for source in results.values() for game in source
|
||||
if game[0] in favorites
|
||||
]
|
||||
self.games_loaded.emit(games)
|
||||
|
||||
# Load games from different sources in parallel to prevent blocking
|
||||
results = {'portproton': [], 'steam': [], 'gog': []}
|
||||
completed = {'portproton': False, 'steam': False, 'gog': False}
|
||||
results = {'portproton': [], 'steam': [], 'gog': [], 'egs': []}
|
||||
completed = dict.fromkeys(results, False)
|
||||
|
||||
def check_completion():
|
||||
if all(completed.values()):
|
||||
QApplication.processEvents() # Keep UI responsive
|
||||
on_all_games_favorites(results['portproton'], results['steam'], results['gog'])
|
||||
on_all_games_favorites()
|
||||
|
||||
def portproton_callback(games):
|
||||
results['portproton'] = games
|
||||
@@ -991,23 +1002,30 @@ class MainWindow(
|
||||
completed['gog'] = True
|
||||
check_completion()
|
||||
|
||||
def egs_callback(games):
|
||||
results['egs'] = games
|
||||
completed['egs'] = True
|
||||
check_completion()
|
||||
|
||||
self._load_portproton_games_async(portproton_callback)
|
||||
self._load_steam_games_async(steam_callback)
|
||||
self._load_gog_games_async(gog_callback)
|
||||
self._load_egs_games_async(egs_callback)
|
||||
else:
|
||||
# For 'all' filter - load games from different sources in parallel to prevent blocking
|
||||
results = {'portproton': [], 'steam': [], 'gog': []}
|
||||
completed = {'portproton': False, 'steam': False, 'gog': False}
|
||||
results = {'portproton': [], 'steam': [], 'gog': [], 'egs': []}
|
||||
completed = dict.fromkeys(results, False)
|
||||
|
||||
def on_all_games():
|
||||
seen = set()
|
||||
games = []
|
||||
for game in results['portproton'] + results['steam'] + results['gog']:
|
||||
for source_games in results.values():
|
||||
for game in source_games:
|
||||
# Unique key: name + exec_line
|
||||
key = (game[0], game[5])
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
games.append(game)
|
||||
key = (game[0], game[5])
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
games.append(game)
|
||||
QApplication.processEvents() # Keep UI responsive
|
||||
self.games_loaded.emit(games)
|
||||
|
||||
@@ -1033,10 +1051,16 @@ class MainWindow(
|
||||
completed['gog'] = True
|
||||
check_completion()
|
||||
|
||||
def egs_callback(games):
|
||||
results['egs'] = games
|
||||
completed['egs'] = True
|
||||
check_completion()
|
||||
|
||||
# Load all sources in parallel
|
||||
self._load_portproton_games_async(portproton_callback)
|
||||
self._load_steam_games_async(steam_callback)
|
||||
self._load_gog_games_async(gog_callback)
|
||||
self._load_egs_games_async(egs_callback)
|
||||
|
||||
# Run loading immediately to show status without delay
|
||||
start_loading()
|
||||
@@ -1104,6 +1128,54 @@ class MainWindow(
|
||||
lambda info, current_game=game: on_game_info(current_game, info),
|
||||
)
|
||||
|
||||
def _load_egs_games_async(self, callback: Callable[[list[tuple]], None]) -> None:
|
||||
library = [game for game in self.egs_api.load_library() if game.get("app_id")]
|
||||
only_installed = game_config.get_only_installed()
|
||||
games = []
|
||||
if not library:
|
||||
callback(games)
|
||||
return
|
||||
from portprotonqt.time_utils import get_statistics_path
|
||||
statistics_file = get_statistics_path()
|
||||
processed_count = 0
|
||||
|
||||
def on_game_info(game: dict, steam_info: dict) -> None:
|
||||
nonlocal processed_count
|
||||
app_id = str(game["app_id"])
|
||||
installed = self.egs_api.is_game_installed(app_id)
|
||||
if not only_installed or installed:
|
||||
target = self.egs_api.get_launch_target(app_id) if installed else None
|
||||
description = str(game.get("description", ""))
|
||||
if not description or description == str(game.get("title", "")):
|
||||
description = str(steam_info.get("description", ""))
|
||||
playtime = get_playtime_for_exe(
|
||||
statistics_file, target, exact_path=True
|
||||
) if target else 0
|
||||
playtime = playtime or 0
|
||||
games.append((
|
||||
str(game.get("title", app_id)), description,
|
||||
str(game.get("cover", "")), app_id,
|
||||
steam_info.get("controller_support", ""),
|
||||
f"egs://{'launch' if installed else 'install'}/{app_id}",
|
||||
get_last_launch(f"egs-{app_id}") if target else _("Never"),
|
||||
format_playtime(playtime), steam_info.get("protondb_tier", ""),
|
||||
steam_info.get("anticheat_status", ""),
|
||||
get_last_launch_timestamp(f"egs-{app_id}") if target else 0,
|
||||
playtime, "egs", steam_info.get("anticheat_slug", ""),
|
||||
steam_info.get("ppdb_id", ""), steam_info.get("ppdb_rating", ""),
|
||||
steam_info.get("appid", ""),
|
||||
))
|
||||
processed_count += 1
|
||||
if processed_count == len(library):
|
||||
callback(games)
|
||||
|
||||
for game in library:
|
||||
get_steam_game_info_async(
|
||||
str(game.get("title", game["app_id"])),
|
||||
f"egs://launch/{game['app_id']}",
|
||||
lambda info, current_game=game: on_game_info(current_game, info),
|
||||
)
|
||||
|
||||
def _upgrade_legacy_gog_library(
|
||||
self, api: GOGAPI, library: list[dict], callback: Callable[[list], None]
|
||||
) -> bool:
|
||||
@@ -1593,6 +1665,14 @@ class MainWindow(
|
||||
):
|
||||
game_data = dict(game_data)
|
||||
game_data["exec_line"] = f"gog://launch/{app_id}"
|
||||
if str(game_data.get("game_source", "")).lower() == "egs":
|
||||
app_id = str(game_data.get("appid", ""))
|
||||
if (
|
||||
exec_line.startswith("egs://install/")
|
||||
and self.egs_api.is_game_installed(app_id)
|
||||
):
|
||||
game_data = dict(game_data)
|
||||
game_data["exec_line"] = f"egs://launch/{app_id}"
|
||||
logger.debug(
|
||||
"Opening detail page for %s from stacked index %d",
|
||||
game_data.get("name", ""),
|
||||
@@ -2138,10 +2218,26 @@ class MainWindow(
|
||||
if exec_line.startswith("gog://launch/"):
|
||||
app_id = exec_line.rsplit("/", 1)[-1]
|
||||
return str(self.gog_api.get_launch_target(app_id) or "")
|
||||
if exec_line.startswith("egs://launch/"):
|
||||
app_id = exec_line.rsplit("/", 1)[-1]
|
||||
return str(self.egs_api.get_launch_target(app_id) or "")
|
||||
return extract_exec_target_path(exec_line) or ""
|
||||
|
||||
def _has_running_game_process(self) -> bool:
|
||||
return any(proc.poll() is None for proc in self.game_processes)
|
||||
if any(proc.poll() is None for proc in self.game_processes):
|
||||
return True
|
||||
target = str(self.target_exe or "").lower()
|
||||
if target:
|
||||
for process in psutil.process_iter(attrs=["name"]):
|
||||
try:
|
||||
if str(process.info.get("name") or "").lower() == target:
|
||||
return True
|
||||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||||
continue
|
||||
started = getattr(self, "game_start_time", None)
|
||||
if started is None:
|
||||
return False
|
||||
return (datetime.now() - started).total_seconds() < STORE_LAUNCH_GRACE_SECONDS
|
||||
|
||||
def _start_launch_output_reader(self, process: subprocess.Popen[str]) -> None:
|
||||
"""Start background reader for PortProton launch output."""
|
||||
@@ -2459,6 +2555,169 @@ class MainWindow(
|
||||
if game:
|
||||
self._install_gog_game(game)
|
||||
|
||||
def _handle_egs_game(self, exec_line: str, button=None) -> None:
|
||||
action, app_id = exec_line.removeprefix("egs://").split("/", 1)
|
||||
if action == "install" and self.egs_api.is_game_installed(app_id):
|
||||
action = "launch"
|
||||
if action == "launch":
|
||||
self._launch_egs_game(app_id, button)
|
||||
return
|
||||
self._install_egs_game(app_id)
|
||||
|
||||
def _install_egs_game(self, app_id: str) -> None:
|
||||
self._install_egs_download(app_id)
|
||||
|
||||
def _repair_egs_game(self, app_id: str) -> None:
|
||||
self._start_egs_operation(
|
||||
app_id, ["repair", app_id, "--skip-sdl", "-y"], _("Repair")
|
||||
)
|
||||
|
||||
def _update_egs_game(self, app_id: str) -> None:
|
||||
self._start_egs_operation(
|
||||
app_id, ["update", app_id, "--platform", "Windows", "--skip-sdl", "-y"],
|
||||
_("Update")
|
||||
)
|
||||
|
||||
def _delete_egs_game(self, app_id: str) -> None:
|
||||
self._start_egs_operation(
|
||||
app_id, ["uninstall", app_id, "-y"], _("Delete")
|
||||
)
|
||||
|
||||
def _import_egs_game(self, app_id: str) -> None:
|
||||
selected_paths: list[str] = []
|
||||
explorer = FileExplorer(
|
||||
self, theme=self.theme, initial_path=str(Path.home() / "Games"),
|
||||
directory_only=True,
|
||||
)
|
||||
explorer.setWindowTitle(_("Select Epic game folder"))
|
||||
explorer.file_signal.file_selected.connect(selected_paths.append)
|
||||
explorer.exec()
|
||||
if selected_paths:
|
||||
self._start_egs_operation(app_id, [
|
||||
"import", app_id, selected_paths[0], "--with-dlcs",
|
||||
"--platform", "Windows", "-y",
|
||||
], _("Import"))
|
||||
|
||||
def _start_egs_operation(
|
||||
self, app_id: str, arguments: list[str], action: str
|
||||
) -> None:
|
||||
if getattr(self, "egs_process", None) is not None:
|
||||
QMessageBox.warning(self, _("Error"), _("Another Epic operation is running"))
|
||||
return
|
||||
try:
|
||||
command = self.egs_api.build_command(arguments)
|
||||
except OSError as error:
|
||||
QMessageBox.warning(self, _("Error"), str(error))
|
||||
return
|
||||
game = next(
|
||||
(item for item in self.egs_api.load_library()
|
||||
if str(item.get("app_id", "")) == app_id),
|
||||
{"app_id": app_id, "title": app_id, "cover": ""},
|
||||
)
|
||||
self._start_egs_visible_operation(game, command, action)
|
||||
|
||||
def _egs_process_environment(self) -> QProcessEnvironment:
|
||||
environment = QProcessEnvironment.systemEnvironment()
|
||||
environment.insert("LEGENDARY_CONFIG_PATH", str(self.egs_api.config_dir))
|
||||
return environment
|
||||
|
||||
def _on_egs_operation_finished(
|
||||
self, app_id: str, action: str, exit_code: int
|
||||
) -> None:
|
||||
self._finish_egs_visible_operation(action, exit_code == 0)
|
||||
if exit_code != 0:
|
||||
logger.error(
|
||||
"Epic %s failed for %s with code %s", action, app_id, exit_code
|
||||
)
|
||||
return
|
||||
if action == "eos-install":
|
||||
self._enable_egs_overlay(app_id)
|
||||
return
|
||||
self.loadGames(force_load=True)
|
||||
|
||||
def _enable_egs_overlay(self, app_id: str) -> None:
|
||||
overlay_path = self.egs_api.data_dir / "eos_overlay"
|
||||
install_record = self.egs_api.config_dir / "overlay_install.json"
|
||||
if not install_record.is_file():
|
||||
self._start_egs_operation(app_id, [
|
||||
"eos-overlay", "install", "--path", str(overlay_path), "-y",
|
||||
], "eos-install")
|
||||
return
|
||||
target = self.egs_api.get_launch_target(app_id)
|
||||
prefix_name = get_prefix_name(target)
|
||||
if not self.portproton_location:
|
||||
QMessageBox.warning(self, _("Error"), _("PortProton directory not found"))
|
||||
return
|
||||
prefix_path = Path(self.portproton_location) / "data/prefixes" / prefix_name
|
||||
if not (prefix_path / "user.reg").is_file():
|
||||
QMessageBox.warning(self, _("Error"), _("Wine prefix not found"))
|
||||
return
|
||||
action = (
|
||||
"disable"
|
||||
if self.egs_api.is_eos_overlay_enabled(prefix_path)
|
||||
else "enable"
|
||||
)
|
||||
self._start_egs_operation(app_id, [
|
||||
"eos-overlay", action, "--prefix", str(prefix_path),
|
||||
], f"eos-{action}")
|
||||
|
||||
def _is_egs_overlay_enabled(self, app_id: str) -> bool:
|
||||
target = self.egs_api.get_launch_target(app_id)
|
||||
if not target or not self.portproton_location:
|
||||
return False
|
||||
prefix_name = get_prefix_name(target)
|
||||
prefix_path = Path(self.portproton_location) / "data/prefixes" / prefix_name
|
||||
return self.egs_api.is_eos_overlay_enabled(prefix_path)
|
||||
|
||||
def _launch_egs_game(self, app_id: str, button=None) -> None:
|
||||
target = self.egs_api.get_launch_target(app_id)
|
||||
if not target:
|
||||
QMessageBox.warning(self, _("Error"), _("Game not found."))
|
||||
return
|
||||
if not self.start_sh:
|
||||
QMessageBox.warning(self, _("Error"), _("PortProton start script not found"))
|
||||
return
|
||||
target_name = os.path.basename(target)
|
||||
if self.game_processes and self.target_exe == target_name:
|
||||
self.stop_running_game(button)
|
||||
return
|
||||
if self.game_processes:
|
||||
QMessageBox.warning(
|
||||
self, _("Error"), _("Cannot launch game while another game is running")
|
||||
)
|
||||
return
|
||||
try:
|
||||
command = self.egs_api.build_command([
|
||||
"launch", app_id, "--wine", self.start_sh[0],
|
||||
])
|
||||
process = subprocess.Popen(
|
||||
command, env=self.egs_api.get_environment(), shell=False,
|
||||
preexec_fn=os.setsid, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT, text=True, errors="replace",
|
||||
)
|
||||
except OSError as error:
|
||||
logger.error("Failed to launch Epic game %s: %s", app_id, error)
|
||||
QMessageBox.warning(self, _("Error"), str(error))
|
||||
return
|
||||
SoundManager().play("game_launch")
|
||||
self.game_processes.append(process)
|
||||
self.target_exe = target_name
|
||||
self.current_running_button = button
|
||||
launch_time = datetime.now()
|
||||
self.game_start_time = launch_time
|
||||
self.game_start_exe = target
|
||||
self.game_start_exact_path = True
|
||||
save_last_launch(f"egs-{app_id}", launch_time)
|
||||
self._update_last_launch_after_start(target, launch_time)
|
||||
self._start_launch_output_reader(process)
|
||||
self.input_manager.suspend_gamepad_polling()
|
||||
self.checkProcessTimer = QTimer(self)
|
||||
self.checkProcessTimer.timeout.connect(self.checkTargetExe)
|
||||
self.checkProcessTimer.start(500)
|
||||
if button is not None:
|
||||
button.setText(_("Stop"))
|
||||
button.setIcon(self.theme_manager.get_icon("stop", as_path=True))
|
||||
|
||||
def _launch_gog_game(
|
||||
self, app_id: str, button=None, play_sound: bool = True
|
||||
) -> None:
|
||||
@@ -2528,6 +2787,10 @@ class MainWindow(
|
||||
self._handle_gog_game(exec_line, button)
|
||||
return
|
||||
|
||||
if exec_line.startswith("egs://"):
|
||||
self._handle_egs_game(exec_line, button)
|
||||
return
|
||||
|
||||
# Handle PortProton games
|
||||
entry_exec_split = shlex.split(exec_line)
|
||||
if not entry_exec_split:
|
||||
|
||||
@@ -13,6 +13,7 @@ install_data(
|
||||
'game_card.py',
|
||||
'game_library_manager.py',
|
||||
'gog_api.py',
|
||||
'egs_api.py',
|
||||
'howlongtobeat_api.py',
|
||||
'image_utils.py',
|
||||
'icon_extractor.py',
|
||||
@@ -152,7 +153,7 @@ install_data(
|
||||
['tabs/__init__.py',
|
||||
'tabs/autoinstall_tab.py',
|
||||
'tabs/control_hints.py',
|
||||
'tabs/gog_tab.py',
|
||||
'tabs/download_tab.py',
|
||||
'tabs/library_tab.py',
|
||||
'tabs/settings_tab.py',
|
||||
'tabs/system_tab.py',
|
||||
|
||||
@@ -625,7 +625,7 @@ def get_steam_game_info_async(
|
||||
from portprotonqt.steam_api.utils import filter_candidates, remove_duplicates
|
||||
|
||||
is_autoinstall = exec_line.startswith("autoinstall:")
|
||||
is_gog_uri = exec_line.startswith("gog://")
|
||||
is_store_uri = exec_line.startswith(("gog://", "egs://"))
|
||||
|
||||
if game_exe.lower().endswith(('.bat', '.cmd')):
|
||||
if os.path.exists(game_exe):
|
||||
@@ -647,7 +647,7 @@ def get_steam_game_info_async(
|
||||
else:
|
||||
logger.error("Command file not found: %s", game_exe)
|
||||
|
||||
if is_gog_uri:
|
||||
if is_store_uri:
|
||||
meta_data = {}
|
||||
elif not game_exe.lower().endswith('.exe'):
|
||||
logger.error("Invalid executable path: %s. Expected .exe", game_exe)
|
||||
@@ -671,15 +671,15 @@ def get_steam_game_info_async(
|
||||
candidates.append(file_description)
|
||||
if desktop_name:
|
||||
candidates.append(desktop_name)
|
||||
if exe_name and not is_gog_uri:
|
||||
if exe_name and not is_store_uri:
|
||||
candidates.append(exe_name)
|
||||
if folder_name and not is_gog_uri:
|
||||
if folder_name and not is_store_uri:
|
||||
candidates.append(folder_name)
|
||||
candidates = filter_candidates(candidates)
|
||||
candidates = remove_duplicates(candidates)
|
||||
candidates_ordered = sorted(candidates, key=lambda s: len(s.split()), reverse=True)
|
||||
has_custom_data = (
|
||||
is_autoinstall or is_gog_uri or _has_custom_data_for_exe(exe_name)
|
||||
is_autoinstall or is_store_uri or _has_custom_data_for_exe(exe_name)
|
||||
)
|
||||
|
||||
def on_steam_apps_and_index(
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Shared download manager with GOG download support."""
|
||||
"""Store account actions and shared game download manager."""
|
||||
|
||||
import os
|
||||
import re
|
||||
@@ -27,6 +27,7 @@ from PySide6.QtWidgets import (
|
||||
)
|
||||
|
||||
from portprotonqt.dialogs.file_explorer import FileExplorer
|
||||
from portprotonqt.egs_api import EGSAPI, EGS_LOGIN_URL
|
||||
from portprotonqt.gog_api import GOGAPI, GOG_LOGIN_URL
|
||||
from portprotonqt.image_utils import load_pixmap_async
|
||||
from portprotonqt.localization import _
|
||||
@@ -36,6 +37,7 @@ from portprotonqt.sound_manager import SoundManager
|
||||
logger = get_logger(__name__)
|
||||
GOG_CANCEL_KILL_TIMEOUT_MS = 3000
|
||||
GOG_LOGIN_TIMEOUT_MS = 300000
|
||||
EGS_LOGIN_TIMEOUT_MS = 300000
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from PySide6.QtWidgets import QMainWindow
|
||||
@@ -64,6 +66,42 @@ class GOGLibraryWorker(QThread):
|
||||
self.failed.emit(str(error))
|
||||
|
||||
|
||||
class EGSLibraryWorker(QThread):
|
||||
"""Refresh the Epic library outside the UI thread."""
|
||||
|
||||
loaded = Signal(list)
|
||||
failed = Signal(str)
|
||||
|
||||
def __init__(self, api: EGSAPI) -> None:
|
||||
super().__init__()
|
||||
self.api = api
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
self.loaded.emit(self.api.refresh_library())
|
||||
except Exception as error:
|
||||
logger.exception("Failed to refresh Epic library")
|
||||
self.failed.emit(str(error))
|
||||
|
||||
|
||||
class EGSAuthWorker(QThread):
|
||||
"""Authenticate an Epic account outside the UI thread."""
|
||||
|
||||
authenticated = Signal(bool, str)
|
||||
|
||||
def __init__(self, api: EGSAPI, code: str) -> None:
|
||||
super().__init__()
|
||||
self.api = api
|
||||
self.code = code
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
self.authenticated.emit(*self.api.authenticate(self.code))
|
||||
except Exception as error:
|
||||
logger.exception("Failed to authenticate with Epic")
|
||||
self.authenticated.emit(False, str(error))
|
||||
|
||||
|
||||
class GOGAuthWorker(QThread):
|
||||
"""Exchange the GOG OAuth code outside the UI thread."""
|
||||
|
||||
@@ -126,7 +164,7 @@ class GOGSupportWorker(QThread):
|
||||
self.failed.emit(self.error)
|
||||
|
||||
|
||||
class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
class MainWindowDownloadTabMixin(_MainWindowTypingBase):
|
||||
"""Add account actions and the shared downloads page."""
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -134,8 +172,11 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
|
||||
def createGOGDownloadsTab(self) -> None:
|
||||
self.gog_process = None
|
||||
self.egs_process = None
|
||||
self.gog_download_queue = []
|
||||
self.gog_download_output = ""
|
||||
self.egs_download_output = ""
|
||||
self.egs_download_total = 0.0
|
||||
self.gog_support_workers = []
|
||||
self.downloadTableHeadings = {}
|
||||
page = QWidget()
|
||||
@@ -402,6 +443,124 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
def _on_gog_library_worker_finished(self) -> None:
|
||||
self.gog_library_worker = None
|
||||
|
||||
def _handle_egs_account_action(self) -> None:
|
||||
if not self.egs_api.user_path.is_file():
|
||||
self._start_egs_login()
|
||||
return
|
||||
if not self.egs_api.logout():
|
||||
self.egsAccountStatus.setText(_("Failed to log out of Epic account"))
|
||||
return
|
||||
self._update_egs_account_state()
|
||||
self.loadGames(force_load=True)
|
||||
|
||||
def _start_egs_login(self) -> None:
|
||||
if getattr(self, "egs_auth_worker", None) is not None:
|
||||
return
|
||||
clipboard = QApplication.clipboard()
|
||||
if not getattr(self, "egs_auth_clipboard_connected", False):
|
||||
clipboard.dataChanged.connect(self._check_egs_login_clipboard)
|
||||
self.egs_auth_clipboard_connected = True
|
||||
self.egs_auth_timer = QTimer(self)
|
||||
self.egs_auth_timer.setSingleShot(True)
|
||||
self.egs_auth_timer.timeout.connect(self._cancel_egs_login)
|
||||
self.egs_auth_timer.start(EGS_LOGIN_TIMEOUT_MS)
|
||||
self.egsLoginButton.setEnabled(False)
|
||||
self.egsAccountStatus.setText(_("Copy the authorization code to clipboard"))
|
||||
if not QDesktopServices.openUrl(QUrl(EGS_LOGIN_URL)):
|
||||
self._cancel_egs_login()
|
||||
|
||||
def _check_egs_login_clipboard(self) -> None:
|
||||
code = self.egs_api.extract_auth_code(QApplication.clipboard().text())
|
||||
if len(code) < 16:
|
||||
return
|
||||
self._stop_egs_login_clipboard()
|
||||
self.egsAccountStatus.setText(_("Refreshing…"))
|
||||
worker = EGSAuthWorker(self.egs_api, code)
|
||||
worker.authenticated.connect(self._on_egs_authenticated)
|
||||
worker.finished.connect(self._on_egs_auth_worker_finished)
|
||||
self.egs_auth_worker = worker
|
||||
worker.start()
|
||||
|
||||
def _stop_egs_login_clipboard(self) -> None:
|
||||
timer = getattr(self, "egs_auth_timer", None)
|
||||
if timer is not None:
|
||||
timer.stop()
|
||||
timer.deleteLater()
|
||||
self.egs_auth_timer = None
|
||||
if getattr(self, "egs_auth_clipboard_connected", False):
|
||||
QApplication.clipboard().dataChanged.disconnect(
|
||||
self._check_egs_login_clipboard
|
||||
)
|
||||
self.egs_auth_clipboard_connected = False
|
||||
|
||||
def _cancel_egs_login(self) -> None:
|
||||
self._stop_egs_login_clipboard()
|
||||
self.egsLoginButton.setEnabled(True)
|
||||
self._update_egs_account_state()
|
||||
|
||||
def _on_egs_authenticated(self, authenticated: bool, error: str) -> None:
|
||||
if not authenticated:
|
||||
self.egsAccountStatus.setText(
|
||||
_("Epic login failed: {0}").format(error or _("Unknown error"))
|
||||
)
|
||||
return
|
||||
self._update_egs_account_state()
|
||||
self._refresh_egs_library()
|
||||
|
||||
def _on_egs_auth_worker_finished(self) -> None:
|
||||
self.egsLoginButton.setEnabled(True)
|
||||
self.egs_auth_worker = None
|
||||
|
||||
def _update_egs_account_state(self) -> None:
|
||||
connected = self.egs_api.user_path.is_file()
|
||||
account_name = self.egs_api.get_account_name() if connected else ""
|
||||
self.egsAccountStatus.setText(
|
||||
account_name or (_("Epic account connected") if connected
|
||||
else _("Epic account not connected"))
|
||||
)
|
||||
self.egsLoginButton.setText(_("Log out") if connected else _("Open login page"))
|
||||
self.egsLoginButton.setEnabled(True)
|
||||
|
||||
def _refresh_egs_library(self) -> None:
|
||||
if not self.egs_api.user_path.is_file():
|
||||
return
|
||||
if getattr(self, "egs_library_worker", None) is not None:
|
||||
return
|
||||
self.egsAccountStatus.setText(_("Refreshing…"))
|
||||
worker = EGSLibraryWorker(self.egs_api)
|
||||
worker.loaded.connect(self._on_egs_library_loaded)
|
||||
worker.failed.connect(self._on_egs_library_failed)
|
||||
worker.finished.connect(self._on_egs_library_worker_finished)
|
||||
self.egs_library_worker = worker
|
||||
worker.start()
|
||||
|
||||
def _on_egs_library_loaded(self, games: list) -> None:
|
||||
self._update_egs_account_state()
|
||||
manager = self.detail_page_manager
|
||||
source = manager._current_detail_source
|
||||
if source and source[0] == "game":
|
||||
source_data = source[1]
|
||||
if str(source_data.get("game_source", "")).lower() == "egs":
|
||||
app_id = str(source_data.get("appid", ""))
|
||||
game = next(
|
||||
(item for item in games if str(item.get("app_id", "")) == app_id),
|
||||
None,
|
||||
)
|
||||
if game:
|
||||
source_data["description"] = str(game.get("description", ""))
|
||||
source_data["cover_path"] = str(game.get("cover", ""))
|
||||
if manager._detail_page_active:
|
||||
manager._reopen_current_detail_page()
|
||||
self.loadGames(force_load=True)
|
||||
|
||||
def _on_egs_library_failed(self, error: str) -> None:
|
||||
self.egsAccountStatus.setText(
|
||||
_("Failed to refresh Epic library: {0}").format(error)
|
||||
)
|
||||
|
||||
def _on_egs_library_worker_finished(self) -> None:
|
||||
self.egs_library_worker = None
|
||||
|
||||
def _install_gog_game(self, game: dict) -> None:
|
||||
if self.gog_process is not None:
|
||||
self.gog_download_queue.append(game)
|
||||
@@ -449,6 +608,36 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
return
|
||||
self._start_gog_download(game, install_path, command, _("Install"))
|
||||
|
||||
def _install_egs_download(self, app_id: str) -> None:
|
||||
if self.gog_process is not None or self.egs_process is not None:
|
||||
QMessageBox.warning(self, _("Error"), _("Another download is running"))
|
||||
return
|
||||
game = next(
|
||||
(item for item in self.egs_api.load_library()
|
||||
if str(item.get("app_id", "")) == app_id),
|
||||
{"app_id": app_id, "title": app_id, "cover": ""},
|
||||
)
|
||||
selected_paths: list[str] = []
|
||||
explorer = FileExplorer(
|
||||
self, theme=self.theme, initial_path=str(self.egs_api.games_dir),
|
||||
directory_only=True,
|
||||
)
|
||||
explorer.setWindowTitle(_("Select installation folder"))
|
||||
explorer.file_signal.file_selected.connect(selected_paths.append)
|
||||
explorer.exec()
|
||||
if not selected_paths:
|
||||
return
|
||||
install_path = Path(selected_paths[0])
|
||||
try:
|
||||
command = self.egs_api.build_command([
|
||||
"install", app_id, "--base-path", str(install_path),
|
||||
"--platform", "Windows", "--skip-sdl", "--skip-dlcs", "-y",
|
||||
])
|
||||
except OSError as error:
|
||||
self.egsAccountStatus.setText(str(error))
|
||||
return
|
||||
self._start_egs_download(game, install_path, command)
|
||||
|
||||
def _repair_gog_game(self, game: dict) -> None:
|
||||
if self.gog_process is not None:
|
||||
self.gogAccountStatus.setText(_("Another GOG operation is already running"))
|
||||
@@ -567,10 +756,78 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
self.switchTab(6)
|
||||
process.start()
|
||||
|
||||
def _start_egs_download(
|
||||
self, game: dict, install_path: Path, command: list[str]
|
||||
) -> None:
|
||||
app_id = str(game["app_id"])
|
||||
self.downloadActiveTitle.setText(str(game["title"]))
|
||||
self.downloadActiveDetails.setText(_("Starting"))
|
||||
cover_width, cover_height = self.theme.downloadsActiveCoverSize
|
||||
load_pixmap_async(
|
||||
str(game.get("cover", "")), cover_width, cover_height,
|
||||
self.downloadActiveCover.setPixmap,
|
||||
app_name=f"download-active-{app_id}",
|
||||
)
|
||||
self.downloadActiveHeading.setVisible(True)
|
||||
self.downloadActiveCard.setVisible(True)
|
||||
self._update_downloads_tab_visibility()
|
||||
process = QProcess(self)
|
||||
process.setProgram(command[0])
|
||||
process.setArguments(command[1:])
|
||||
process.setProcessEnvironment(self._egs_process_environment())
|
||||
process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
|
||||
process.readyReadStandardOutput.connect(self._read_egs_download_output)
|
||||
process.finished.connect(self._on_egs_download_finished)
|
||||
self.egs_process = process
|
||||
self.egs_active_game = game
|
||||
self.egs_active_install_path = install_path
|
||||
self.egs_download_output = ""
|
||||
self.egs_download_total = 0.0
|
||||
self.downloadOverallProgress.setValue(0)
|
||||
self.downloadCancelButton.setEnabled(True)
|
||||
self.switchTab(6)
|
||||
process.start()
|
||||
|
||||
def _start_egs_visible_operation(
|
||||
self, game: dict, command: list[str], action: str
|
||||
) -> None:
|
||||
app_id = str(game["app_id"])
|
||||
display_action = "EOS Overlay" if action.startswith("eos-") else action
|
||||
self.downloadActiveTitle.setText(str(game["title"]))
|
||||
self.downloadActiveDetails.setText(display_action)
|
||||
cover_width, cover_height = self.theme.downloadsActiveCoverSize
|
||||
load_pixmap_async(
|
||||
str(game.get("cover", "")), cover_width, cover_height,
|
||||
self.downloadActiveCover.setPixmap,
|
||||
app_name=f"download-active-{app_id}",
|
||||
)
|
||||
self.downloadActiveHeading.setVisible(True)
|
||||
self.downloadActiveCard.setVisible(True)
|
||||
self._update_downloads_tab_visibility()
|
||||
process = QProcess(self)
|
||||
process.setProgram(command[0])
|
||||
process.setArguments(command[1:])
|
||||
process.setProcessEnvironment(self._egs_process_environment())
|
||||
process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
|
||||
process.readyReadStandardOutput.connect(self._read_egs_download_output)
|
||||
process.finished.connect(
|
||||
lambda code, _status: self._on_egs_operation_finished(
|
||||
app_id, action, code
|
||||
)
|
||||
)
|
||||
self.egs_process = process
|
||||
self.egs_active_game = game
|
||||
self.egs_download_output = ""
|
||||
self.egs_download_total = 0.0
|
||||
self.downloadOverallProgress.setValue(0)
|
||||
self.downloadCancelButton.setEnabled(True)
|
||||
self.switchTab(6)
|
||||
process.start()
|
||||
|
||||
def _cancel_gog_download(self) -> None:
|
||||
if self.gog_process is None:
|
||||
process = self.gog_process or self.egs_process
|
||||
if process is None:
|
||||
return
|
||||
process = self.gog_process
|
||||
self.downloadCancelButton.setEnabled(False)
|
||||
self.downloadActiveDetails.setText(_("Cancel"))
|
||||
process.terminate()
|
||||
@@ -585,13 +842,13 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
process.kill()
|
||||
|
||||
def _append_download_row(
|
||||
self, table: QTableWidget, game: dict, action: str
|
||||
self, table: QTableWidget, game: dict, action: str, store: str = "GOG"
|
||||
) -> tuple[int, QLabel]:
|
||||
row = table.rowCount()
|
||||
table.insertRow(row)
|
||||
game_cell, details_label = self._create_download_game_cell(game)
|
||||
table.setCellWidget(row, 0, game_cell)
|
||||
values = (datetime.now().strftime("%H:%M:%S"), action, "GOG")
|
||||
values = (datetime.now().strftime("%H:%M:%S"), action, store)
|
||||
for column, value in enumerate(values, 1):
|
||||
table.setItem(row, column, QTableWidgetItem(value))
|
||||
table.setRowHeight(row, self.theme.downloadsTableRowHeight)
|
||||
@@ -670,6 +927,57 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
if details:
|
||||
self._update_active_download_details(details)
|
||||
|
||||
def _read_egs_download_output(self) -> None:
|
||||
if self.egs_process is None:
|
||||
return
|
||||
output = bytes(
|
||||
self.egs_process.readAllStandardOutput().data()
|
||||
).decode(errors="replace")
|
||||
self.egs_download_output = (self.egs_download_output + output)[-4096:]
|
||||
total = re.findall(r"Download size:\s+([\d.]+)\s+MiB", output)
|
||||
if total:
|
||||
self.egs_download_total = float(total[-1])
|
||||
verification = re.findall(
|
||||
r"Verification progress:\s*\d+/\d+\s+\(([\d.]+)%\)"
|
||||
r"(?:\s+\[([\d.]+)\s+MiB/s\])?",
|
||||
self.egs_download_output,
|
||||
)
|
||||
if verification:
|
||||
percent, speed = verification[-1]
|
||||
self.downloadOverallProgress.setValue(min(100, int(float(percent))))
|
||||
details = [f"{percent}%"]
|
||||
if speed:
|
||||
details.append(f"{speed} MiB/s")
|
||||
self._update_active_download_details(details)
|
||||
downloaded = re.findall(
|
||||
r"Downloaded:\s+([\d.]+)\s+MiB", self.egs_download_output
|
||||
)
|
||||
if downloaded and self.egs_download_total > 0:
|
||||
progress = float(downloaded[-1]) / self.egs_download_total * 100
|
||||
self.downloadOverallProgress.setValue(min(100, int(progress)))
|
||||
download_speeds = re.findall(
|
||||
r"Download\t-\s+(\S+)\s+MiB", self.egs_download_output
|
||||
)
|
||||
disk_speeds = re.findall(
|
||||
r"Disk\t-\s+(\S+)\s+MiB", self.egs_download_output
|
||||
)
|
||||
eta = re.findall(r"ETA:\s+(\d{2}:\d{2}:\d{2})", self.egs_download_output)
|
||||
if download_speeds:
|
||||
self.downloadSpeedLabel.setText(
|
||||
_("Download: {0} MiB/s").format(download_speeds[-1])
|
||||
)
|
||||
if disk_speeds:
|
||||
self.diskSpeedLabel.setText(
|
||||
_("Disk: {0} MiB/s").format(disk_speeds[-1])
|
||||
)
|
||||
details = [f"{downloaded[-1]} MiB"] if downloaded else []
|
||||
if eta:
|
||||
details.append(_("ETA: {0}").format(eta[-1]))
|
||||
if details:
|
||||
self._update_active_download_details(details)
|
||||
elif output.strip():
|
||||
self.downloadActiveDetails.setText(output.strip().splitlines()[-1])
|
||||
|
||||
def _update_active_download_details(self, details: list[str]) -> None:
|
||||
self.downloadActiveDetails.setText(" · ".join(details))
|
||||
|
||||
@@ -713,6 +1021,49 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
self._update_download_table_height(self.downloadQueuedTable)
|
||||
self._install_gog_game(next_game)
|
||||
|
||||
def _on_egs_download_finished(
|
||||
self, code: int, _status: QProcess.ExitStatus
|
||||
) -> None:
|
||||
game = self.egs_active_game
|
||||
app_id = str(game["app_id"])
|
||||
self.downloadActiveHeading.setVisible(False)
|
||||
self.downloadActiveCard.setVisible(False)
|
||||
completed_action = _("Installed") if code == 0 else _("Failed")
|
||||
_row, completed_details = self._append_download_row(
|
||||
self.downloadCompletedTable, game, completed_action, "Epic Games"
|
||||
)
|
||||
if code == 0:
|
||||
self._update_installed_egs_detail(app_id)
|
||||
self.loadGames(force_load=True)
|
||||
else:
|
||||
error = self._get_egs_download_error()
|
||||
completed_details.setText(error)
|
||||
logger.error("Epic download failed for %s: %s", app_id, error)
|
||||
process = self.egs_process
|
||||
self.egs_process = None
|
||||
if process is not None:
|
||||
process.deleteLater()
|
||||
self.downloadOverallProgress.setValue(0)
|
||||
self.downloadSpeedLabel.setText(_("Downloading: ") + "\u2014")
|
||||
self.diskSpeedLabel.setText(_("Disk: ") + "\u2014")
|
||||
|
||||
def _finish_egs_visible_operation(self, action: str, success: bool) -> None:
|
||||
game = self.egs_active_game
|
||||
display_action = "EOS Overlay" if action.startswith("eos-") else action
|
||||
self.downloadActiveHeading.setVisible(False)
|
||||
self.downloadActiveCard.setVisible(False)
|
||||
_row, details = self._append_download_row(
|
||||
self.downloadCompletedTable, game, display_action, "Epic Games"
|
||||
)
|
||||
details.setText(_("Completed") if success else self._get_egs_download_error())
|
||||
process = self.egs_process
|
||||
self.egs_process = None
|
||||
if process is not None:
|
||||
process.deleteLater()
|
||||
self.downloadOverallProgress.setValue(0)
|
||||
self.downloadSpeedLabel.setText(_("Downloading: ") + "\u2014")
|
||||
self.diskSpeedLabel.setText(_("Disk: ") + "\u2014")
|
||||
|
||||
def _start_gog_support_setup(self, app_id: str, button=None) -> None:
|
||||
if getattr(self, "gog_support_workers", []):
|
||||
return
|
||||
@@ -792,6 +1143,14 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
useful = [line for line in lines if line and "[PROGRESS]" not in line]
|
||||
return useful[-1] if useful else _("Download process failed")
|
||||
|
||||
def _get_egs_download_error(self) -> str:
|
||||
lines = [line.strip() for line in self.egs_download_output.splitlines()]
|
||||
errors = [line for line in lines if "ERROR" in line.upper()]
|
||||
if errors:
|
||||
return errors[-1]
|
||||
useful = [line for line in lines if line]
|
||||
return useful[-1] if useful else _("Download process failed")
|
||||
|
||||
def _update_installed_gog_detail(self, app_id: str) -> None:
|
||||
install_uri = f"gog://install/{app_id}"
|
||||
if self.current_exec_line != install_uri:
|
||||
@@ -804,3 +1163,16 @@ class MainWindowGOGTabMixin(_MainWindowTypingBase):
|
||||
source[1]["exec_line"] = f"gog://launch/{app_id}"
|
||||
self.current_exec_line = source[1]["exec_line"]
|
||||
self.detail_page_manager._reopen_current_detail_page()
|
||||
|
||||
def _update_installed_egs_detail(self, app_id: str) -> None:
|
||||
install_uri = f"egs://install/{app_id}"
|
||||
if self.current_exec_line != install_uri:
|
||||
return
|
||||
if self.stackedWidget.currentWidget() is not self.currentDetailPage:
|
||||
return
|
||||
source = self.detail_page_manager._current_detail_source
|
||||
if source is None or source[0] != "game":
|
||||
return
|
||||
source[1]["exec_line"] = f"egs://launch/{app_id}"
|
||||
self.current_exec_line = source[1]["exec_line"]
|
||||
self.detail_page_manager._reopen_current_detail_page()
|
||||
@@ -302,11 +302,12 @@ class MainWindowLibraryTabMixin(_MainWindowTypingBase):
|
||||
|
||||
display_filter = game_config.get_display_filter()
|
||||
only_installed = game_config.get_only_installed()
|
||||
self.filter_keys = ["all", "steam", "gog", "portproton", "favorites"]
|
||||
self.filter_keys = ["all", "steam", "gog", "egs", "portproton", "favorites"]
|
||||
self.filter_labels = [
|
||||
_("All"),
|
||||
"Steam",
|
||||
"GOG",
|
||||
"Epic Games",
|
||||
"PortProton",
|
||||
_("Favorites"),
|
||||
]
|
||||
|
||||
@@ -167,6 +167,23 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase):
|
||||
gog_account_title.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
|
||||
gogForm.addRow(gog_account_title, gog_buttons)
|
||||
self._update_gog_account_state()
|
||||
self.egsAccountStatus = QLabel()
|
||||
self.egsAccountStatus.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
|
||||
self.egsAccountStatus.setSizePolicy(
|
||||
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
|
||||
)
|
||||
self.egsLoginButton = AutoSizeButton(_("Open login page"))
|
||||
self.egsLoginButton.setProperty("theme_style_name", "ACTION_BUTTON_STYLE")
|
||||
self.egsLoginButton.setStyleSheet(self.theme.ACTION_BUTTON_STYLE)
|
||||
self.egsLoginButton.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
|
||||
self.egsLoginButton.clicked.connect(self._handle_egs_account_action)
|
||||
egs_buttons = QHBoxLayout()
|
||||
egs_buttons.addWidget(self.egsAccountStatus)
|
||||
egs_buttons.addWidget(self.egsLoginButton)
|
||||
egs_account_title = QLabel(_("Account") + " Epic Games:")
|
||||
egs_account_title.setStyleSheet(self.theme.SETTINGS_TITLE_STYLE)
|
||||
gogForm.addRow(egs_account_title, egs_buttons)
|
||||
self._update_egs_account_state()
|
||||
|
||||
self.timeDetailCombo = CustomComboBox(theme=self.theme)
|
||||
self.timeDetailCombo.view().window().setWindowFlags(
|
||||
@@ -224,7 +241,8 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase):
|
||||
uiForm.addRow(self.trayMenuModeTitle, self.trayMenuModeCombo)
|
||||
self.setTabOrder(self.timeDetailCombo, self.steamAccountCombo)
|
||||
self.setTabOrder(self.steamAccountCombo, self.gogLoginButton)
|
||||
self.setTabOrder(self.gogLoginButton, self.trayMenuModeCombo)
|
||||
self.setTabOrder(self.gogLoginButton, self.egsLoginButton)
|
||||
self.setTabOrder(self.egsLoginButton, self.trayMenuModeCombo)
|
||||
|
||||
self.gamepad_type_keys = ["auto", "xbox", "playstation"]
|
||||
self.gamepad_type_labels = [_("Auto"), "Xbox", "PlayStation"]
|
||||
@@ -758,6 +776,7 @@ class MainWindowSettingsTabMixin(_MainWindowTypingBase):
|
||||
if reply == QMessageBox.StandardButton.Yes:
|
||||
cache_config.clear_cache()
|
||||
self.gog_api.clear_library_cache()
|
||||
self.egs_api.clear_library_cache()
|
||||
|
||||
def integrateAppImage(self) -> None:
|
||||
"""Install the running AppImage in the user application menu."""
|
||||
|
||||
@@ -574,6 +574,36 @@ def test_non_steam_without_shortcut_shows_add_not_edit(monkeypatch) -> None:
|
||||
assert add_count == 1
|
||||
|
||||
|
||||
def test_egs_install_detail_shows_install_button(monkeypatch) -> None:
|
||||
monkeypatch.setattr("portprotonqt.detail_pages._", lambda text: text)
|
||||
manager = _make_manager_with_shortcut_mock(has_shortcut=False, monkeypatch=monkeypatch)
|
||||
play_button = cast(Any, manager._create_play_button).return_value
|
||||
|
||||
manager._create_game_buttons_layout({
|
||||
"game_source": "egs",
|
||||
"exec_line": "egs://install/doom64",
|
||||
"appid": "doom64",
|
||||
"name": "DOOM 64",
|
||||
})
|
||||
|
||||
play_button.setText.assert_called_once_with("Install")
|
||||
|
||||
|
||||
def test_installed_egs_detail_shows_store_controls(monkeypatch) -> None:
|
||||
monkeypatch.setattr("portprotonqt.detail_pages._", lambda text: text)
|
||||
manager = _make_manager_with_shortcut_mock(has_shortcut=False, monkeypatch=monkeypatch)
|
||||
manager.main_window.egs_api.get_launch_target.return_value = "/games/DOOM64.exe"
|
||||
|
||||
manager._create_game_buttons_layout({
|
||||
"game_source": "egs",
|
||||
"exec_line": "egs://launch/doom64",
|
||||
"appid": "doom64",
|
||||
"name": "DOOM 64",
|
||||
})
|
||||
|
||||
assert _get_added_buttons(manager) == ["Settings", "Create Log", "Open Folder"]
|
||||
|
||||
|
||||
def test_steam_with_appid_shows_edit_not_add(monkeypatch) -> None:
|
||||
manager = _make_manager_with_shortcut_mock(has_shortcut=False, monkeypatch=monkeypatch)
|
||||
manager._create_game_buttons_layout({
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import orjson
|
||||
from pytest import MonkeyPatch
|
||||
|
||||
from portprotonqt.egs_api import EGSAPI
|
||||
from portprotonqt.localization import get_store_content_languages
|
||||
|
||||
|
||||
def test_build_command_uses_isolated_legendary_config(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
api = EGSAPI()
|
||||
api.config_dir = tmp_path / "legendary"
|
||||
monkeypatch.setattr(api, "get_legendary_path", lambda: "/bin/legendary")
|
||||
|
||||
assert api.build_command(["list", "--json"]) == [
|
||||
"/bin/legendary", "list", "--json",
|
||||
]
|
||||
assert api.get_environment()["LEGENDARY_CONFIG_PATH"] == str(api.config_dir)
|
||||
|
||||
|
||||
def test_build_command_places_confirmation_before_subcommand(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
api = EGSAPI()
|
||||
api.config_dir = tmp_path / "legendary"
|
||||
monkeypatch.setattr(api, "get_legendary_path", lambda: "/bin/legendary")
|
||||
|
||||
assert api.build_command(["install", "Game", "-y"]) == [
|
||||
"/bin/legendary", "-y", "install", "Game",
|
||||
]
|
||||
|
||||
|
||||
def test_update_legendary_skips_matching_release(tmp_path: Path, monkeypatch) -> None:
|
||||
api = EGSAPI()
|
||||
api.data_dir = tmp_path
|
||||
api.version_path = tmp_path / "bin/legendary.version"
|
||||
bundled = tmp_path / "bin/legendary"
|
||||
bundled.parent.mkdir()
|
||||
bundled.touch(mode=0o755)
|
||||
api.version_path.write_text("v1", encoding="utf-8")
|
||||
monkeypatch.setattr(api, "_get_latest_legendary_release", lambda: ({}, "v1"))
|
||||
|
||||
assert api.update_legendary() == str(bundled)
|
||||
|
||||
|
||||
def test_normalize_game_uses_epic_metadata() -> None:
|
||||
game = EGSAPI._normalize_game({
|
||||
"app_name": "TestGame",
|
||||
"app_title": "Test Game",
|
||||
"metadata": {
|
||||
"description": "Description",
|
||||
"namespace": "test-namespace",
|
||||
"keyImages": [{"type": "DieselGameBoxTall", "url": "cover.jpg"}],
|
||||
},
|
||||
})
|
||||
|
||||
assert game == {
|
||||
"app_id": "TestGame",
|
||||
"title": "Test Game",
|
||||
"description": "Description",
|
||||
"cover": "cover.jpg",
|
||||
"namespace": "test-namespace",
|
||||
"description_locale": "",
|
||||
"description_preference": "",
|
||||
"description_cache_version": 0,
|
||||
}
|
||||
|
||||
|
||||
def test_store_description_uses_epic_product_page(monkeypatch: MonkeyPatch) -> None:
|
||||
api = EGSAPI()
|
||||
graphql_response = SimpleNamespace(
|
||||
raise_for_status=lambda: None,
|
||||
json=lambda: {
|
||||
"data": {"Catalog": {"catalogNs": {
|
||||
"mappings": [{"pageSlug": "control"}],
|
||||
}}},
|
||||
},
|
||||
)
|
||||
store_response = SimpleNamespace(
|
||||
status_code=200,
|
||||
raise_for_status=lambda: None,
|
||||
json=lambda: {"pages": [{
|
||||
"type": "productHome",
|
||||
"data": {"about": {"description": "Real description"}},
|
||||
}]},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"portprotonqt.egs_api.requests.post",
|
||||
lambda *args, **kwargs: graphql_response,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"portprotonqt.egs_api.requests.get",
|
||||
lambda *args, **kwargs: store_response,
|
||||
)
|
||||
monkeypatch.setattr("portprotonqt.localization.get_metadata_language", lambda: "en")
|
||||
|
||||
description = api._get_store_description({
|
||||
"app_id": "Calluna", "title": "Control", "namespace": "calluna",
|
||||
})
|
||||
|
||||
assert description == ("Real description", "en")
|
||||
|
||||
|
||||
def test_store_description_prefers_localized_short_text() -> None:
|
||||
data = {"pages": [{
|
||||
"type": "productHome",
|
||||
"data": {"about": {
|
||||
"shortDescription": "Краткое описание",
|
||||
"description": "Первый абзац\n\nВторой абзац",
|
||||
}},
|
||||
}]}
|
||||
|
||||
assert EGSAPI._parse_store_description(data) == "Краткое описание"
|
||||
|
||||
|
||||
def test_store_description_limits_fallback_to_first_paragraph() -> None:
|
||||
data = {"pages": [{
|
||||
"type": "productHome",
|
||||
"data": {"about": {"description": "Первый абзац\n\nВторой абзац"}},
|
||||
}]}
|
||||
|
||||
assert EGSAPI._parse_store_description(data) == "Первый абзац"
|
||||
|
||||
|
||||
def test_store_description_handles_missing_product_pages(
|
||||
monkeypatch: MonkeyPatch,
|
||||
) -> None:
|
||||
api = EGSAPI()
|
||||
response = SimpleNamespace(
|
||||
status_code=200,
|
||||
raise_for_status=lambda: None,
|
||||
json=lambda: {"pages": None},
|
||||
)
|
||||
monkeypatch.setattr(api, "_get_product_slug", lambda _namespace, _title: "game")
|
||||
monkeypatch.setattr("portprotonqt.egs_api.requests.get", lambda *args, **kwargs: response)
|
||||
|
||||
description = api._get_store_description({
|
||||
"app_id": "Game", "title": "Game", "namespace": "game",
|
||||
})
|
||||
|
||||
assert description == ("", "")
|
||||
|
||||
|
||||
def test_content_languages_follow_epic_locale_conventions(
|
||||
monkeypatch: MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr("portprotonqt.localization.get_metadata_language", lambda: "pt")
|
||||
monkeypatch.setattr("portprotonqt.localization.get_system_locale", lambda: "pt_BR")
|
||||
|
||||
assert get_store_content_languages() == ("pt-BR", "pt", "en-US")
|
||||
|
||||
monkeypatch.setattr("portprotonqt.localization.get_metadata_language", lambda: "zh")
|
||||
monkeypatch.setattr("portprotonqt.localization.get_system_locale", lambda: "zh_Hant_TW")
|
||||
|
||||
assert get_store_content_languages() == ("zh-TW", "zh", "en-US")
|
||||
|
||||
|
||||
def test_extract_auth_code_accepts_legendary_response() -> None:
|
||||
response = '{"authorizationCode":"epic-authorization-code"}'
|
||||
|
||||
assert EGSAPI.extract_auth_code(response) == "epic-authorization-code"
|
||||
assert EGSAPI.extract_auth_code("direct-code") == "direct-code"
|
||||
assert EGSAPI.extract_auth_code("https://example.com") == ""
|
||||
|
||||
|
||||
def test_get_launch_target_uses_legendary_install_record(tmp_path: Path) -> None:
|
||||
api = EGSAPI()
|
||||
api.config_dir = tmp_path / "legendary"
|
||||
game_dir = tmp_path / "game"
|
||||
executable = game_dir / "Binaries/Game.exe"
|
||||
executable.parent.mkdir(parents=True)
|
||||
executable.touch()
|
||||
api.config_dir.mkdir()
|
||||
(api.config_dir / "installed.json").write_bytes(orjson.dumps({
|
||||
"TestGame": {
|
||||
"install_path": str(game_dir),
|
||||
"executable": "Binaries\\Game.exe",
|
||||
},
|
||||
}))
|
||||
|
||||
assert api.get_launch_target("TestGame") == str(executable)
|
||||
|
||||
|
||||
def test_parse_download_sizes_reads_legendary_info() -> None:
|
||||
output = orjson.dumps({
|
||||
"manifest": {"download_size": 1024},
|
||||
"install": {"install_size": 2048},
|
||||
})
|
||||
|
||||
assert EGSAPI.parse_download_sizes(output) == (1024, 2048)
|
||||
|
||||
|
||||
def test_parse_download_sizes_uses_manifest_disk_size() -> None:
|
||||
output = orjson.dumps({"manifest": {"download_size": 1024, "disk_size": 4096}})
|
||||
|
||||
assert EGSAPI.parse_download_sizes(output) == (1024, 4096)
|
||||
|
||||
|
||||
def test_zero_cached_install_size_is_refetched(tmp_path: Path) -> None:
|
||||
api = EGSAPI()
|
||||
api.sizes_path = tmp_path / "sizes.json"
|
||||
api.sizes_path.write_bytes(orjson.dumps({"Game": [1024, 0]}))
|
||||
|
||||
assert api.get_cached_download_sizes("Game") is None
|
||||
|
||||
|
||||
def test_authenticate_passes_epic_code(tmp_path: Path, monkeypatch) -> None:
|
||||
api = EGSAPI()
|
||||
api.config_dir = tmp_path / "legendary"
|
||||
api.user_path = api.config_dir / "user.json"
|
||||
monkeypatch.setattr(api, "ensure_legendary", lambda: "/bin/legendary")
|
||||
monkeypatch.setattr(api, "build_command", lambda args: ["legendary", *args])
|
||||
|
||||
def run(command: list[str], **_kwargs: Any) -> SimpleNamespace:
|
||||
api.config_dir.mkdir(parents=True)
|
||||
api.user_path.write_text("{}", encoding="utf-8")
|
||||
assert command == ["legendary", "auth", "--code", "epic-code"]
|
||||
return SimpleNamespace(returncode=0, stderr=b"")
|
||||
|
||||
monkeypatch.setattr("portprotonqt.egs_api.subprocess.run", run)
|
||||
|
||||
assert api.authenticate("epic-code") == (True, "")
|
||||
+248
-15
@@ -3,6 +3,7 @@
|
||||
import shlex
|
||||
from collections.abc import Callable
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from queue import Queue
|
||||
import threading
|
||||
@@ -10,7 +11,7 @@ from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from pytest import MonkeyPatch
|
||||
from pytest import MonkeyPatch, mark
|
||||
from PySide6.QtCore import QObject, Qt
|
||||
from PySide6.QtWidgets import QApplication, QComboBox, QGridLayout, QWidget
|
||||
|
||||
@@ -18,16 +19,19 @@ from portprotonqt.animations.library_controls import _animation_duration
|
||||
from portprotonqt.animations.game_card import GameCardAnimations
|
||||
from portprotonqt.config import game_config
|
||||
from portprotonqt.detail_pages import DetailPageManager
|
||||
from portprotonqt.egs_api import EGSAPI
|
||||
from portprotonqt.tabs.download_tab import MainWindowDownloadTabMixin
|
||||
from portprotonqt.game_card import GameCard, SourceCorner
|
||||
from portprotonqt.game_library_manager import GameLibraryManager
|
||||
from portprotonqt.gog_api import GOGAPI
|
||||
from portprotonqt.main_window import MainWindow
|
||||
from portprotonqt.portproton_api import remove_empty_custom_data_dirs
|
||||
import portprotonqt.tabs.autoinstall_tab as autoinstall_tab_module
|
||||
import portprotonqt.tabs.gog_tab as gog_tab_module
|
||||
import portprotonqt.tabs.download_tab as download_tab_module
|
||||
import portprotonqt.tabs.library_tab as library_tab_module
|
||||
import portprotonqt.tabs.system_tab as system_tab_module
|
||||
import portprotonqt.steam_api.api as steam_api_module
|
||||
import portprotonqt.main_window as main_window_module
|
||||
from portprotonqt.tabs import (
|
||||
MainWindowAutoInstallTabMixin,
|
||||
MainWindowLibraryTabMixin,
|
||||
@@ -37,7 +41,7 @@ from portprotonqt.tabs import (
|
||||
MainWindowWineTabMixin,
|
||||
)
|
||||
from portprotonqt.tabs.autoinstall_tab import MainWindowAutoInstallTabMixin as AutoInstallMixin
|
||||
from portprotonqt.tabs.gog_tab import MainWindowGOGTabMixin as GOGMixin
|
||||
from portprotonqt.tabs.download_tab import MainWindowDownloadTabMixin as GOGMixin
|
||||
from portprotonqt.tabs.library_tab import MainWindowLibraryTabMixin as LibraryMixin
|
||||
from portprotonqt.tabs.settings_tab import MainWindowSettingsTabMixin as SettingsMixin
|
||||
from portprotonqt.tabs.theme_store import THEME_STORE_ITEM, ThemeStoreMixin
|
||||
@@ -562,7 +566,7 @@ def test_gog_account_state_detects_saved_auth(
|
||||
gogAccountStatus=Control(),
|
||||
gogLoginButton=Control(),
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
|
||||
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
|
||||
|
||||
GOGMixin._update_gog_account_state(cast(Any, window))
|
||||
|
||||
@@ -598,7 +602,7 @@ def test_gog_account_action_logs_out_and_reloads_library(
|
||||
window._update_gog_account_state = lambda: GOGMixin._update_gog_account_state(
|
||||
cast(Any, window)
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
|
||||
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
|
||||
|
||||
GOGMixin._handle_gog_account_action(cast(Any, window))
|
||||
|
||||
@@ -750,7 +754,7 @@ def test_gog_library_refresh_failure_reloads_cached_games(
|
||||
gogAccountStatus=SimpleNamespace(setText=lambda text: calls.append(text)),
|
||||
loadGames=lambda **kwargs: calls.append(kwargs),
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
|
||||
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
|
||||
|
||||
GOGMixin._on_gog_library_failed(cast(Any, window), "network error")
|
||||
|
||||
@@ -768,7 +772,7 @@ def test_gog_login_shows_cached_games_before_refresh(monkeypatch: MonkeyPatch) -
|
||||
loadGames=lambda **kwargs: calls.append(("load", kwargs)),
|
||||
_refresh_gog_library=lambda: calls.append("refresh"),
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "_", lambda text: text)
|
||||
monkeypatch.setattr(download_tab_module, "_", lambda text: text)
|
||||
|
||||
GOGMixin._on_gog_authenticated(cast(Any, window), True, "")
|
||||
|
||||
@@ -815,7 +819,7 @@ def test_gog_support_uses_regular_launch_output_monitor(
|
||||
_set_running_button_stop=lambda: None,
|
||||
checkTargetExe=lambda: None,
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "QTimer", FakeTimer)
|
||||
monkeypatch.setattr(download_tab_module, "QTimer", FakeTimer)
|
||||
|
||||
GOGMixin._track_gog_support_process(
|
||||
cast(Any, window), "123", cast(Any, process)
|
||||
@@ -909,7 +913,7 @@ def test_install_gog_game_uses_support_path(
|
||||
explorer.file_signal.file_selected.connect.side_effect = (
|
||||
lambda callback: callback(str(selected_path))
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "FileExplorer", lambda *_args, **_kwargs: explorer)
|
||||
monkeypatch.setattr(download_tab_module, "FileExplorer", lambda *_args, **_kwargs: explorer)
|
||||
api = SimpleNamespace(
|
||||
config_dir=tmp_path / "gogdl",
|
||||
get_install_path=lambda _app_id, _title: selected_path,
|
||||
@@ -930,11 +934,43 @@ def test_install_gog_game_uses_support_path(
|
||||
]
|
||||
|
||||
|
||||
def test_install_egs_game_selects_path_and_opens_download(
|
||||
tmp_path: Path, monkeypatch: MonkeyPatch
|
||||
) -> None:
|
||||
selected_path = tmp_path / "Epic"
|
||||
started: list[tuple] = []
|
||||
explorer = MagicMock()
|
||||
explorer.file_signal.file_selected.connect.side_effect = (
|
||||
lambda callback: callback(str(selected_path))
|
||||
)
|
||||
monkeypatch.setattr(download_tab_module, "FileExplorer", lambda *_args, **_kwargs: explorer)
|
||||
game = {"app_id": "doom64", "title": "DOOM 64", "cover": "cover"}
|
||||
api = SimpleNamespace(
|
||||
games_dir=tmp_path / "Games", load_library=lambda: [game],
|
||||
build_command=lambda arguments: ["legendary", *arguments],
|
||||
)
|
||||
window = SimpleNamespace(
|
||||
gog_process=None, egs_process=None, theme=object(), egs_api=api,
|
||||
_start_egs_download=lambda *arguments: started.append(arguments),
|
||||
egsAccountStatus=SimpleNamespace(setText=MagicMock()),
|
||||
)
|
||||
|
||||
GOGMixin._install_egs_download(cast(Any, window), "doom64")
|
||||
|
||||
assert started[0] == (
|
||||
game, selected_path,
|
||||
[
|
||||
"legendary", "install", "doom64", "--base-path", str(selected_path),
|
||||
"--platform", "Windows", "--skip-sdl", "--skip-dlcs", "-y",
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def test_cancel_gog_download_terminates_then_kills(monkeypatch: MonkeyPatch) -> None:
|
||||
process = SimpleNamespace(
|
||||
terminate=MagicMock(),
|
||||
kill=MagicMock(),
|
||||
state=lambda: gog_tab_module.QProcess.ProcessState.Running,
|
||||
state=lambda: download_tab_module.QProcess.ProcessState.Running,
|
||||
)
|
||||
window = SimpleNamespace(
|
||||
gog_process=process,
|
||||
@@ -944,7 +980,7 @@ def test_cancel_gog_download_terminates_then_kills(monkeypatch: MonkeyPatch) ->
|
||||
cast(Any, window), active
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module.QTimer, "singleShot", lambda _delay, callback: callback())
|
||||
monkeypatch.setattr(download_tab_module.QTimer, "singleShot", lambda _delay, callback: callback())
|
||||
|
||||
GOGMixin._cancel_gog_download(cast(Any, window))
|
||||
|
||||
@@ -986,7 +1022,7 @@ def test_import_gog_game_saves_selected_installation(
|
||||
gogAccountStatus=SimpleNamespace(setText=lambda _text: None),
|
||||
loadGames=lambda **_kwargs: None,
|
||||
)
|
||||
monkeypatch.setattr(gog_tab_module, "FileExplorer", Explorer)
|
||||
monkeypatch.setattr(download_tab_module, "FileExplorer", Explorer)
|
||||
|
||||
GOGMixin._import_gog_game(cast(Any, window), {"app_id": "123", "title": "Game"})
|
||||
|
||||
@@ -1887,7 +1923,16 @@ def test_installed_filter_disabled_includes_uninstalled_gog_games(
|
||||
assert results[0][0][5] == "gog://install/uninstalled"
|
||||
|
||||
|
||||
def test_gog_metadata_search_ignores_uri_components(monkeypatch: MonkeyPatch) -> None:
|
||||
@mark.parametrize(
|
||||
("game_name", "launch_uri"),
|
||||
(
|
||||
("Unknown GOG Game", "gog://launch/123"),
|
||||
("Unknown Epic Game", "egs://launch/Fortnite"),
|
||||
),
|
||||
)
|
||||
def test_store_metadata_search_ignores_uri_components(
|
||||
monkeypatch: MonkeyPatch, game_name: str, launch_uri: str
|
||||
) -> None:
|
||||
from portprotonqt.steam_api import get_steam_game_info_async
|
||||
|
||||
searched_candidates = []
|
||||
@@ -1916,13 +1961,201 @@ def test_gog_metadata_search_ignores_uri_components(monkeypatch: MonkeyPatch) ->
|
||||
)
|
||||
results = []
|
||||
|
||||
get_steam_game_info_async("Unknown GOG Game", "gog://launch/123", results.append)
|
||||
get_steam_game_info_async(game_name, launch_uri, results.append)
|
||||
|
||||
assert searched_candidates == ["Unknown GOG Game"]
|
||||
assert searched_candidates == [game_name]
|
||||
assert len(results) == 1
|
||||
fetch_sgdb_cover.assert_not_called()
|
||||
|
||||
|
||||
def test_egs_refresh_reopens_current_detail_with_new_description() -> None:
|
||||
source_data = {
|
||||
"appid": "AmongUs",
|
||||
"game_source": "egs",
|
||||
"description": "Old description",
|
||||
"cover_path": "old.jpg",
|
||||
}
|
||||
manager = SimpleNamespace(
|
||||
_current_detail_source=("game", source_data),
|
||||
_detail_page_active=True,
|
||||
_reopen_current_detail_page=MagicMock(),
|
||||
)
|
||||
window = cast(MainWindowDownloadTabMixin, SimpleNamespace(
|
||||
detail_page_manager=manager,
|
||||
_update_egs_account_state=MagicMock(),
|
||||
loadGames=MagicMock(),
|
||||
))
|
||||
|
||||
MainWindowDownloadTabMixin._on_egs_library_loaded(window, [{
|
||||
"app_id": "AmongUs",
|
||||
"description": "Новое описание",
|
||||
"cover": "new.jpg",
|
||||
}])
|
||||
|
||||
assert source_data["description"] == "Новое описание"
|
||||
assert source_data["cover_path"] == "new.jpg"
|
||||
manager._reopen_current_detail_page.assert_called_once_with()
|
||||
window.loadGames.assert_called_once_with(force_load=True)
|
||||
|
||||
|
||||
def test_egs_library_uses_steam_description_when_epic_has_title_only(
|
||||
monkeypatch: MonkeyPatch,
|
||||
) -> None:
|
||||
game = {"app_id": "WitchIt", "title": "Witch It", "description": "Witch It"}
|
||||
api = SimpleNamespace(
|
||||
load_library=lambda: [game],
|
||||
is_game_installed=lambda _app_id: False,
|
||||
)
|
||||
results = []
|
||||
window = cast(MainWindow, SimpleNamespace(
|
||||
egs_api=api, games=[],
|
||||
))
|
||||
monkeypatch.setattr(game_config, "get_only_installed", lambda: False)
|
||||
monkeypatch.setattr(
|
||||
"portprotonqt.main_window.get_steam_game_info_async",
|
||||
lambda _name, _uri, callback: callback({
|
||||
"description": "Witch It — игра в прятки по сети.",
|
||||
}),
|
||||
)
|
||||
|
||||
MainWindow._load_egs_games_async(window, results.append)
|
||||
|
||||
assert results[0][0][1] == "Witch It — игра в прятки по сети."
|
||||
|
||||
|
||||
def test_egs_maintenance_uses_legendary_commands() -> None:
|
||||
start_operation = MagicMock()
|
||||
window = cast(MainWindow, SimpleNamespace(_start_egs_operation=start_operation))
|
||||
|
||||
MainWindow._repair_egs_game(window, "Game")
|
||||
MainWindow._update_egs_game(window, "Game")
|
||||
MainWindow._delete_egs_game(window, "Game")
|
||||
|
||||
calls = start_operation.call_args_list
|
||||
assert calls[0].args[1] == ["repair", "Game", "--skip-sdl", "-y"]
|
||||
assert calls[1].args[1] == [
|
||||
"update", "Game", "--platform", "Windows", "--skip-sdl", "-y",
|
||||
]
|
||||
assert calls[2].args[1] == ["uninstall", "Game", "-y"]
|
||||
|
||||
|
||||
def test_egs_operation_is_sent_to_visible_downloads() -> None:
|
||||
game = {"app_id": "Game", "title": "Epic Game", "cover": "cover"}
|
||||
visible_operation = MagicMock()
|
||||
api = SimpleNamespace(
|
||||
build_command=lambda arguments: ["legendary", *arguments],
|
||||
load_library=lambda: [game],
|
||||
)
|
||||
window = cast(MainWindow, SimpleNamespace(
|
||||
egs_process=None, egs_api=api,
|
||||
_start_egs_visible_operation=visible_operation,
|
||||
))
|
||||
|
||||
MainWindow._start_egs_operation(window, "Game", ["repair", "Game"], "Repair")
|
||||
|
||||
visible_operation.assert_called_once_with(
|
||||
game, ["legendary", "repair", "Game"], "Repair"
|
||||
)
|
||||
|
||||
|
||||
def test_detached_store_game_keeps_running_state(monkeypatch: MonkeyPatch) -> None:
|
||||
dead_launcher = SimpleNamespace(poll=lambda: 0)
|
||||
game_process = SimpleNamespace(info={"name": "DOOM64_x64.exe"})
|
||||
monkeypatch.setattr(
|
||||
main_window_module.psutil, "process_iter", lambda attrs: [game_process]
|
||||
)
|
||||
window = cast(MainWindow, SimpleNamespace(
|
||||
game_processes=[dead_launcher], target_exe="DOOM64_x64.exe",
|
||||
game_start_time=datetime.now() - timedelta(minutes=1),
|
||||
))
|
||||
|
||||
assert MainWindow._has_running_game_process(window)
|
||||
|
||||
|
||||
def test_store_launch_grace_prevents_early_button_reset(
|
||||
monkeypatch: MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(main_window_module.psutil, "process_iter", lambda attrs: [])
|
||||
window = cast(MainWindow, SimpleNamespace(
|
||||
game_processes=[], target_exe="DOOM64_x64.exe",
|
||||
game_start_time=datetime.now(),
|
||||
))
|
||||
|
||||
assert MainWindow._has_running_game_process(window)
|
||||
|
||||
|
||||
def test_egs_verification_uses_total_progress_format() -> None:
|
||||
import re
|
||||
|
||||
output = "Verification progress: 37/142 (26.4%) [132.7 MiB/s]"
|
||||
match = re.findall(
|
||||
r"Verification progress:\s*\d+/\d+\s+\(([\d.]+)%\)"
|
||||
r"(?:\s+\[([\d.]+)\s+MiB/s\])?",
|
||||
output,
|
||||
)
|
||||
|
||||
assert match == [("26.4", "132.7")]
|
||||
|
||||
|
||||
def test_egs_overlay_enable_uses_game_prefix(tmp_path: Path) -> None:
|
||||
prefix_path = tmp_path / "data/prefixes/DEFAULT"
|
||||
prefix_path.mkdir(parents=True)
|
||||
(prefix_path / "user.reg").touch()
|
||||
config_dir = tmp_path / "legendary"
|
||||
config_dir.mkdir()
|
||||
(config_dir / "overlay_install.json").touch()
|
||||
api = SimpleNamespace(
|
||||
config_dir=config_dir,
|
||||
data_dir=tmp_path / "egs",
|
||||
get_launch_target=MagicMock(return_value="/games/Game.exe"),
|
||||
is_eos_overlay_enabled=EGSAPI.is_eos_overlay_enabled,
|
||||
)
|
||||
start_operation = MagicMock()
|
||||
window = cast(MainWindow, SimpleNamespace(
|
||||
egs_api=api,
|
||||
portproton_location=str(tmp_path),
|
||||
_start_egs_operation=start_operation,
|
||||
))
|
||||
|
||||
MainWindow._enable_egs_overlay(window, "Game")
|
||||
|
||||
arguments = start_operation.call_args.args[1]
|
||||
assert arguments == [
|
||||
"eos-overlay", "enable", "--prefix", str(prefix_path),
|
||||
]
|
||||
|
||||
|
||||
def test_egs_overlay_disable_uses_game_prefix(tmp_path: Path) -> None:
|
||||
prefix_path = tmp_path / "data/prefixes/DEFAULT"
|
||||
prefix_path.mkdir(parents=True)
|
||||
(prefix_path / "user.reg").write_text(
|
||||
'[Software\\\\Epic Games\\\\EOS]\n"OverlayPath"="Z:/overlay"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
config_dir = tmp_path / "legendary"
|
||||
config_dir.mkdir()
|
||||
(config_dir / "overlay_install.json").touch()
|
||||
api = SimpleNamespace(
|
||||
config_dir=config_dir,
|
||||
data_dir=tmp_path / "egs",
|
||||
get_launch_target=MagicMock(return_value="/games/Game.exe"),
|
||||
is_eos_overlay_enabled=EGSAPI.is_eos_overlay_enabled,
|
||||
)
|
||||
start_operation = MagicMock()
|
||||
window = cast(MainWindow, SimpleNamespace(
|
||||
egs_api=api,
|
||||
portproton_location=str(tmp_path),
|
||||
_start_egs_operation=start_operation,
|
||||
))
|
||||
|
||||
MainWindow._enable_egs_overlay(window, "Game")
|
||||
|
||||
arguments = start_operation.call_args.args[1]
|
||||
assert arguments == [
|
||||
"eos-overlay", "disable", "--prefix", str(prefix_path),
|
||||
]
|
||||
|
||||
|
||||
def test_remove_empty_custom_data_dirs_keeps_non_empty_dirs(tmp_config_dir: Path) -> None:
|
||||
custom_data_path = tmp_config_dir.parent / "data" / "PortProtonQt" / "custom_data"
|
||||
(custom_data_path / "praest").mkdir(parents=True)
|
||||
|
||||
Reference in new issue
Block a user