#!/usr/bin/env python3 """Test script for oplog import/export operations.""" from __future__ import annotations import argparse import platform import subprocess import sys from pathlib import Path from typing import NamedTuple _PLATFORM = "windows" if sys.platform == "win32" else "macosx" if sys.platform == "darwin" else "linux" _ARCH = "x64" if sys.platform == "win32" else platform.machine().lower() _EXE_SUFFIX = ".exe" if sys.platform == "win32" else "" class Build(NamedTuple): name: str bucket: str id: str BUILDS = [ Build("XB1Client", "fortnitegame.oplog.fortnite-main.xb1client", "09a75f7f3b7517653dcdaaa4"), Build("WindowsClient", "fortnitegame.oplog.fortnite-main.windowsclient", "09a75d977ef944ecfd0eddfd"), Build("SwitchClient", "fortnitegame.oplog.fortnite-main.switchclient", "09a74d03b3598ec94cfd2644"), Build("XSXClient", "fortnitegame.oplog.fortnite-main.xsxclient", "09a76c2bbd6cd78f4d40d9ea"), Build("Switch2Client", "fortnitegame.oplog.fortnite-main.switch2client", "09a7686b3d9faa78fb24a38f"), Build("PS4Client", "fortnitegame.oplog.fortnite-main.ps4client", "09a75b72d1c260ed26020140"), Build("LinuxServer", "fortnitegame.oplog.fortnite-main.linuxserver", "09a747f5e0ee83a04be013e6"), Build("IOSClient", "fortnitegame.oplog.fortnite-main.iosclient", "09a75f677e883325a209148c"), Build("Android_ASTCClient", "fortnitegame.oplog.fortnite-main.android_astcclient", "09a7422c08c6f37becc7d37f"), ] ZEN_EXE: Path = Path(f"./build/{_PLATFORM}/{_ARCH}/release/zen{_EXE_SUFFIX}") ZEN_PORT = 8558 ZEN_CACHE_PORT = 8559 ZEN_CACHE = f"http://127.0.0.1:{ZEN_CACHE_PORT}" ZEN_CACHE_POPULATE = "true" ZEN_PARTIAL_REQUEST_MODE = "true" SERVER_ARGS: tuple[str, ...] = ( "--http", "asio", "--gc-cache-duration-seconds", "1209600", "--gc-interval-seconds", "21600", "--gc-low-diskspace-threshold", "2147483648", "--cache-bucket-limit-overwrites", ) def run(cmd: list[str | Path]) -> None: try: subprocess.run(cmd, check=True) except FileNotFoundError: sys.exit(f"error: executable not found: {cmd[0]}") except subprocess.CalledProcessError as e: sys.exit(f"error: command failed with exit code {e.returncode}:\n {' '.join(str(x) for x in e.cmd)}") def stop_server(label: str, port: int) -> None: """Stop a zen server. Tolerates failures so it is safe to call from finally blocks.""" print(f"--------- stopping {label}") try: subprocess.run([ZEN_EXE, "down", "--port", str(port)]) except OSError as e: print(f"warning: could not stop {label}: {e}", file=sys.stderr) print() def start_server(label: str, data_dir: Path, port: int, extra_args: list[str] | None = None) -> None: print(f"--------- starting {label} {data_dir}") run([ ZEN_EXE, "up", "--port", str(port), "--show-console", "--", f"--data-dir={data_dir}", *SERVER_ARGS, *(extra_args or []), ]) print() def wipe_or_create(label: str, path: Path) -> None: if path.exists(): print(f"--------- cleaning {label} {path}") run([ZEN_EXE, "wipe", "-y", path]) else: print(f"--------- creating {label} {path}") path.mkdir(parents=True, exist_ok=True) print() def check_prerequisites() -> None: if not ZEN_EXE.is_file(): sys.exit(f"error: zen executable not found: {ZEN_EXE}") def setup_project(port: int) -> None: """Create the FortniteGame project on the server at the given port.""" print("--------- creating FortniteGame project") run([ZEN_EXE, "project-create", f"--hosturl=127.0.0.1:{port}", "FortniteGame", "--force-update"]) print() def setup_oplog(port: int, build_name: str) -> None: """Create the oplog in the FortniteGame project on the server at the given port.""" print(f"--------- creating {build_name} oplog") run([ZEN_EXE, "oplog-create", f"--hosturl=127.0.0.1:{port}", "FortniteGame", build_name, "--force-update"]) print() def main() -> None: global ZEN_EXE parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "positional_path", nargs="?", default=None, type=Path, metavar="DATA_PATH", help="root path for all data directories (positional shorthand for --data-path)", ) parser.add_argument( "zen_exe_positional", nargs="?", default=None, type=Path, metavar="ZEN_EXE_PATH", help="path to zen executable (positional shorthand for --zen-exe-path)", ) parser.add_argument( "--data-path", default=Path(Path(__file__).stem + "_datadir"), type=Path, metavar="PATH", help=f"root path for all data directories (default: {Path(__file__).stem}_datadir)", ) parser.add_argument( "--zen-exe-path", default=ZEN_EXE, type=Path, metavar="PATH", help=f"path to zen executable (default: {ZEN_EXE})", ) args = parser.parse_args() data_path = args.positional_path if data_path is None: data_path = args.data_path ZEN_EXE = args.zen_exe_positional if ZEN_EXE is None: ZEN_EXE = args.zen_exe_path zen_data_dir = data_path / "DDC" / "OplogsZen" zen_cache_data_dir = data_path / "DDC" / "ZenBuildsCache" zen_import_data_dir = data_path / "DDC" / "OplogsZenImport" export_dir = data_path / "Export" / "FortniteGame" check_prerequisites() start_server("cache zenserver", zen_cache_data_dir, ZEN_CACHE_PORT, ["--buildstore-enabled"]) try: wipe_or_create("zenserver data", zen_data_dir) start_server("zenserver", zen_data_dir, ZEN_PORT) try: setup_project(ZEN_PORT) for build in BUILDS: setup_oplog(ZEN_PORT, build.name) print(f"--------- importing {build.name} oplog") run([ ZEN_EXE, "oplog-import", f"--hosturl=127.0.0.1:{ZEN_PORT}", "FortniteGame", build.name, "--clean", "--builds", "https://jupiter.devtools.epicgames.com", "--namespace", "fortnite.oplog", "--bucket", build.bucket, "--builds-id", build.id, f"--zen-cache-host={ZEN_CACHE}", f"--zen-cache-upload={ZEN_CACHE_POPULATE}", f"--allow-partial-block-requests={ZEN_PARTIAL_REQUEST_MODE}", ]) print() print(f"--------- validating {build.name} oplog") run([ZEN_EXE, "oplog-validate", f"--hosturl=127.0.0.1:{ZEN_PORT}", "FortniteGame", build.name]) print() wipe_or_create("export folder", export_dir) for build in BUILDS: print(f"--------- exporting {build.name} oplog") run([ ZEN_EXE, "oplog-export", f"--hosturl=127.0.0.1:{ZEN_PORT}", "FortniteGame", build.name, "--file", export_dir, "--forcetempblocks", ]) print() finally: stop_server("zenserver", ZEN_PORT) wipe_or_create("alternate zenserver data", zen_import_data_dir) start_server("import zenserver", zen_import_data_dir, ZEN_PORT) try: setup_project(ZEN_PORT) for build in BUILDS: setup_oplog(ZEN_PORT, build.name) print(f"--------- importing {build.name} oplog") run([ ZEN_EXE, "oplog-import", f"--hosturl=127.0.0.1:{ZEN_PORT}", "FortniteGame", build.name, "--file", export_dir, ]) print() print(f"--------- validating {build.name} oplog") run([ZEN_EXE, "oplog-validate", f"--hosturl=127.0.0.1:{ZEN_PORT}", "FortniteGame", build.name]) print() finally: stop_server("alternative zenserver", ZEN_PORT) finally: stop_server("cache zenserver", ZEN_CACHE_PORT) if __name__ == "__main__": main()