pgames-manager/backend/tests/test_compatibility.py

import asyncio
import json
import zipfile
from types import SimpleNamespace

import pytest

from app.config import get_settings
from app.converter.compatibility import (
    CompatibilityError,
    audit_ir,
    detect,
    require_supported_scripts,
)
from app.converter.nani_compiler import compile_package
from app.converter.nani_expr import assignment, parse
from app.converter.nani_package import Package, decode_script, text_records
from app.converter.pipeline import convert_archive
from app.converter.preflight import inspect_directory, probe_archive
from app.converter.rpy_parser import RpyParser, ScriptData
from app.import_library import check, entries
from app.translation import LineInfo, validate_nani_translation


def write(path, text=""):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")
    return path


def script(source):
    data = ScriptData()
    RpyParser(data, "script.rpy").parse(source)
    return data


def package(*commands):
    return Package(scripts={"Opening": {"name": "Opening", "lines": [
        {"line": i, "commands": [command]} for i, command in enumerate(commands)]}})


def command(op, **args):
    return {"op": op, "args": args}


@pytest.mark.parametrize("engine,marker", [
    ("unity", "Game_Data/Managed/Assembly-CSharp.dll"),
    ("unity", "Game_Data/il2cpp_data/Metadata/global-metadata.dat"),
    ("gamemaker", "data.win"), ("godot", "Game.pck"), ("rpgmaker", "Game.rgss3a")])
def test_unsupported_engines_fail_with_reason(tmp_path, engine, marker):
    write(tmp_path / marker)
    with pytest.raises(CompatibilityError) as exc:
        detect(tmp_path)
    assert exc.value.engine == engine
    assert exc.value.report()["issues"][0]["code"] == "unsupported_engine"


def test_naninovel_detection_ignores_migration_copies(tmp_path):
    write(tmp_path / "Game_Data/Managed/Elringus.Naninovel.Runtime.dll")
    write(tmp_path / "migration-staging/port/project.godot")
    assert detect(tmp_path).engine == "naninovel"


def test_multiple_same_engine_games_are_not_merged(tmp_path):
    write(tmp_path / "one/game/script.rpy", 'label start:\n    "one"')
    write(tmp_path / "two/game/script.rpy", 'label start:\n    "two"')
    with pytest.raises(CompatibilityError, match="несколько игр"):
        detect(tmp_path)


def test_renpy_only_declarations_are_not_execution_blockers(tmp_path):
    source = 'screen unused():\n    text "UI"\nlabel start:\n    "Hello"\n    return\n'
    write(tmp_path / "game/script.rpy", source)
    assert inspect_directory(tmp_path)["compatible"]
    write(tmp_path / "game/script.rpy", source.replace('    "Hello"', '    call screen unused'))
    assert inspect_directory(tmp_path)["compatible"]  # a defined screen can be called
    write(tmp_path / "game/script.rpy", source.replace('    "Hello"', '    call screen missing_screen'))
    with pytest.raises(CompatibilityError, match="screen"):
        inspect_directory(tmp_path)


@pytest.mark.parametrize("source,reason", [
    ('init python:\n    def helper():\n        yield 1\nlabel start:\n    "Hi"', "init python"),
    ('label start:\n    $ result = renpy.show_displayable("map")', "renpy.show_displayable"),
    ('label start:\n    jump missing', "missing"),
])
def test_unsupported_renpy_logic_never_degrades_to_ready(source, reason):
    data = script(source)
    audit_ir(data)
    with pytest.raises(CompatibilityError, match=reason):
        require_supported_scripts(data)


def test_renpy_atl_animation_becomes_a_timeline_not_a_static_pose():
    data = script('transform moving:\n    xalign 0.0\n    linear 1.0 xalign 1.0\n    repeat\nlabel start:\n    "Hi"')
    require_supported_scripts(data)
    assert data.transforms["moving"] == {"xalign": 0.0}
    block = data.atl["moving"]
    assert block["repeat"] is True
    assert [step["op"] for step in block["steps"]] == ["set", "move"]
    assert block["steps"][1]["props"] == {"xalign": 1.0} and block["steps"][1]["warp"] == "linear"


def test_renpy_atl_constructs_without_a_runtime_are_still_rejected():
    for body in ("time 1.0:\n        xalign 1.0", "knot (0, 0)", "circles 0.5", "contains 'a.png'"):
        unused = script(f'transform odd:\n    {body}\nlabel start:\n    "Hi"')
        require_supported_scripts(unused)  # nothing shows it, so it cannot break the game
        used = script(f'transform odd:\n    {body}\nlabel start:\n    show eileen at odd\n    "Hi"')
        with pytest.raises(CompatibilityError, match="ATL"):
            require_supported_scripts(used)
        early = script(f'label start:\n    show eileen at odd\n    "Hi"\ntransform odd:\n    {body}\n')
        with pytest.raises(CompatibilityError, match="ATL"):  # used before it is defined
            require_supported_scripts(early)


def test_renpy_atl_groups_events_and_named_transforms():
    data = script(
        "transform base:\n    zoom 0.8\n"
        "transform bob:\n    base\n    parallel:\n        ease 1.0 yoffset -10\n        ease 1.0 yoffset 0\n"
        "        repeat\n    parallel:\n        pause 0.5\n"
        "    on hide:\n        linear 0.3 alpha 0.0\n"
        'label start:\n    "Hi"')
    require_supported_scripts(data)
    steps = data.atl["bob"]["steps"]
    assert steps[0] == {"op": "set", "props": {"zoom": 0.8}}  # known static transform is inlined
    assert [s["op"] for s in steps] == ["set", "parallel", "on"]
    assert len(steps[1]["blocks"]) == 2 and steps[1]["blocks"][0]["repeat"] is True
    assert steps[2]["events"] == ["hide"]


def test_source_encoding_is_preserved_or_explicitly_rejected(tmp_path):
    from app.converter.rpy_lexer import ParseError
    from app.converter.source import read_source
    source = tmp_path / "script.rpy"
    source.write_bytes('# coding: cp1251\nlabel start:\n    "Привет"'.encode("cp1251"))
    assert "Привет" in read_source(source)
    source.write_bytes(b'label start:\n    "\xff"')
    with pytest.raises(ParseError, match="кодировку"):
        read_source(source)


def test_directory_rejection_precedes_copying_media(tmp_path, monkeypatch):
    write(tmp_path / "source/game/script.rpy", 'label start:\n    call screen map')
    def media(*args, **kwargs):
        pytest.fail("Media must not be copied/converted for an unsupported game")
    monkeypatch.setattr("app.converter.pipeline.shutil.copytree", media)
    monkeypatch.setattr("app.converter.pipeline.assets.convert", media)
    with pytest.raises(CompatibilityError):
        convert_archive(tmp_path / "source", "source", tmp_path / "work", lambda _: None)
    assert (tmp_path / "source/game/script.rpy").exists()


def test_archive_probe_does_not_extract_anything(tmp_path):
    source = tmp_path / "game.zip"
    with zipfile.ZipFile(source, "w") as archive:
        archive.writestr("Game/data.win", b"placeholder")
    with pytest.raises(CompatibilityError) as exc:
        probe_archive(source, source.name)
    assert exc.value.engine == "gamemaker"
    assert list(tmp_path.iterdir()) == [source]


def test_archive_probe_ignores_engine_runtime_scripts(tmp_path):
    source = tmp_path / "game.zip"
    with zipfile.ZipFile(source, "w") as archive:
        archive.writestr("Game/Game_Data/Managed/Elringus.Naninovel.Runtime.dll", b"")
        archive.writestr("Game/renpy/common/script.rpy", b"")
    assert probe_archive(source, source.name)["engine"] == "naninovel"


def test_script_preflight_does_not_unpack_loose_media(tmp_path):
    from app.converter.archive import extract
    source = tmp_path / "game.zip"
    with zipfile.ZipFile(source, "w") as archive:
        archive.writestr("Game/game/script.rpy", 'label start:\n    call screen map')
        archive.writestr("Game/game/images/large.png", b"media")
    destination = tmp_path / "scripts"
    extract(source, source.name, destination, only_scripts=True)
    assert (destination / "Game/game/script.rpy").exists()
    assert not (destination / "Game/game/images").exists()


def test_batch_prefers_unpacked_duplicate_and_reports_executables(tmp_path):
    (tmp_path / "Game").mkdir()
    write(tmp_path / "Game.zip")
    write(tmp_path / "Other.exe")
    assert [p.name for p in entries(tmp_path)] == ["Game", "Other.exe"]
    assert check(tmp_path / "Other.exe")["issues"][0]["code"] == "not_distribution"


def test_naninovel_decodes_line_order_and_dynamic_parameters():
    tree = {"m_Name": "Opening", "lines": [{"id": 1}], "references": {"records": [
        {"type": {"class": "PrintText"}, "data": {
            "Text": {"hasValue": False, "value": "", "dynamicValue": {"ValueText": "Score: {score}"}},
            "playbackSpot": {"lineIndex": 3}}},
        {"type": {"class": "GenericTextScriptLine"}, "data": {
            "lineIndex": 3, "lineHash": "original", "inlinedCommands": [{"id": 0}]}}]}}
    decoded = decode_script(tree)
    assert decoded["lines"][0]["commands"][0]["args"]["Text"] == {"template": "Score: {score}"}
    tree["lines"][0]["id"] = -1
    with pytest.raises(ValueError, match="ссылка"):
        decode_script(tree)


def test_naninovel_compiles_dialogue_branches_choices_and_expressions():
    original = package(command("SetCustomVariable", Expression="score=2"),
        command("BeginIf", Expression="score>=1 && true"),
        command("PrintText", Text={"template": "Score: {score}"}, AuthorId="Alice"),
        command("Else"), command("PrintText", Text="Low"), command("EndIf"),
        command("AddChoice", ChoiceSummary="Again", GotoPath="Opening"), command("Stop"))
    data, entry, expressions = compile_package(original)
    body = data.modules["Opening.nani"][0]["body"]
    assert entry == "nani:Opening"
    assert body[0]["code"][0]["v"]["k"] == "nani"
    assert len(body[1]["branches"]) == 2
    assert body[2]["op"] == "nani_add_choice" and body[3]["op"] == "nani_menu"
    assert expressions == {"score": ["name", "score"]}


@pytest.mark.parametrize("commands,reason", [
    ([command("Spawn", Path="Minigame")], "Spawn"),
    ([command("PrintText", Text="Text", Unexpected=True)], "Unexpected"),
    ([command("SetCustomVariable", Expression="x=CustomFunction()")], "CustomFunction"),
    ([command("SetCustomVariable", Expression={"template": "x={other}"})], "Динамический"),
    ([command("Stop")], "внешнего интерфейса"),
    ([command("Goto", Path="Missing")], "цель"),
    ([command("BeginIf", Expression="true")], "Не закрыт"),
])
def test_naninovel_unknown_commands_functions_and_fields_are_fatal(commands, reason):
    with pytest.raises(CompatibilityError, match=reason):
        compile_package(package(*commands))


def test_naninovel_expression_parser_never_accepts_code():
    ast = parse('score>1 ? "Yes" : "No"')
    assert ast[0] == "if"
    assert parse('"Привет\\n"')[1] == "Привет\n"
    assert assignment("score=score+1")["v"]["k"] == "nani"
    for source in ('__import__("os")', 'x.constructor()', 'x[0]', 'x; other()', 'x ? 1'):
        with pytest.raises(ValueError):
            parse(source)


def test_naninovel_resources_keep_keys_delimiters_and_context():
    records = text_records("Tips", "# comment\nHeading\nHelp:Title|Description\n", "Tips")
    assert [r["text"] for r in records.values()] == ["Title", "Description"]
    assert len(text_records("_synonyms", "word=hello,hi\n", "synonyms")) == 2
    duplicates = text_records("Tips", "Key:A\nKey:B", "Tips")
    assert [r["text"] for r in duplicates.values()] == ["A", "B"]
    assert [r["occurrence"] for r in duplicates.values()] == [0, 1]
    info = LineInfo("id", "node", "script/resources.json", "resources", "", "<b>Hello {name}</b>", None,
                    resource={"delimiters": "|\r\n"})
    validate_nani_translation(info, "<b>Привет {name}</b>")
    for invalid in ("Привет {name}", "<b>Привет {other}</b>", "<b>Привет {name}|test</b>"):
        with pytest.raises(ValueError):
            validate_nani_translation(info, invalid)


def test_naninovel_import_localization_and_mod_editor_roundtrip(tmp_path, monkeypatch):
    from app.blobs import read_bytes
    from app.converter.naninovel import convert_naninovel
    from app.mods import document, validate
    from app.translation import apply_changes, locale_tables, text_index

    monkeypatch.setenv("DATA_DIR", str(tmp_path / "library"))
    get_settings.cache_clear()
    original = package(command("SetCustomVariable", Expression="score=3"),
                       command("PrintText", Text="Hello {score}"),
                       command("AddChoice", ChoiceSummary="Again", GotoPath="Opening"), command("Stop"))
    original.texts = text_records("Tips", "Help:Title|Description\n", "Tips")
    monkeypatch.setattr("app.converter.naninovel.read_package", lambda *args, **kwargs: original)
    try:
        result = convert_naninovel(tmp_path, tmp_path / "work", lambda _: None)
        game = json.loads(read_bytes(result.manifest["files"]["game.json"]["hash"]))
        assert game["engine"] == "naninovel" and game["entry"] == "nani:Opening"
        index = text_index(result.manifest)
        assert len(index.lines) == 4  # dialogue, choice, two independently addressed Tips parts
        line = next(info for info in index.lines if info.source == "Hello {score}")
        translated = apply_changes(result.manifest, "ru", lines={line.tid: {"t": "Привет {score}"}})
        assert locale_tables(translated, "ru")[0][line.tid]["t"] == "Привет {score}"
        with pytest.raises(ValueError, match="выражения"):
            apply_changes(result.manifest, "ru", lines={line.tid: {"t": "Привет"}})
        validate(document(translated), translated)
    finally:
        get_settings.cache_clear()


def test_unity_codec_restores_global_hooks_after_error():
    from UnityPy.helpers import TypeTreeHelper as helper

    from app.converter.unity_scripts import ScriptCodec
    codec = ScriptCodec.__new__(ScriptCodec)
    before = (helper.read_value, helper.write_value, helper.read_typetree_boost, helper.get_ref_type_node)
    with pytest.raises(RuntimeError):
        with codec.active():
            raise RuntimeError("broken object")
    assert (helper.read_value, helper.write_value, helper.read_typetree_boost, helper.get_ref_type_node) == before


def test_unity_package_closes_cyclic_readers_and_optional_none_fields(tmp_path):
    handle = (tmp_path / "asset").open("wb")
    reader = SimpleNamespace(stream=handle, files=None, cabs=None)
    asset = SimpleNamespace(reader=reader, files=None, cabs={"same": reader})
    environment = SimpleNamespace(files={"asset": asset}, cabs={"asset": asset})
    original = Package(environments=[environment], resources=[asset])
    original.close()
    assert handle.closed
    assert not original.resources and not original.environments
    (tmp_path / "asset").unlink()


def test_unused_renpy_translation_entries_are_not_script_blockers(tmp_path):
    from app.converter import locales
    from app.converter.rpy_parser import ScriptData
    data = ScriptData()
    data.modules = {"script.rpy": [{"name": "start", "body": [{"op": "say", "what": "Hello", "_ln": 1}]}]}
    by_line = locales.assign_ids(data.modules)
    path = tmp_path / "tl/russian/script.rpy"
    path.parent.mkdir(parents=True)
    path.write_text('# game/script.rpy:1\ntranslate russian start_123:\n    "Привет"\n\n\n\n\n\n'
                    'translate russian removed_456:\n    "Старая реплика"\n', encoding="utf-8")
    translations = locales.import_tl(tmp_path, data.modules, by_line, data.warn)
    assert translations["ru"].unmatched == 1
    assert not data.blockers
    assert any("реплики без пары" in message for message in data.warnings)


def test_upload_rejects_unsupported_engine_without_queueing(tmp_path, monkeypatch):
    from fastapi import HTTPException

    from app.routers import versions
    from app.uploads import upload_path

    source = tmp_path / "unsupported.zip"
    with zipfile.ZipFile(source, "w") as archive:
        archive.writestr("data.win", b"game")
    monkeypatch.setenv("DATA_DIR", str(tmp_path / "library"))
    get_settings.cache_clear()
    states = []

    class DB:
        version = None
        def add(self, version):
            version.id = 123
            self.version = version
        async def commit(self):
            states.append(self.version.status)
        async def delete(self, version):
            assert version.status != "queued"

    async def game(*_):
        return SimpleNamespace(id=1)
    async def stream():
        yield source.read_bytes()
    monkeypatch.setattr(versions, "get_game", game)
    try:
        with pytest.raises(HTTPException) as exc:
            asyncio.run(versions.upload_version(1, SimpleNamespace(stream=stream), SimpleNamespace(id=1), DB(), source.name))
        assert exc.value.status_code == 422
        assert "gamemaker" in exc.value.detail
        assert "queued" not in states
        assert not upload_path(123).exists()
    finally:
        get_settings.cache_clear()