PGU-NetFileSyncWin/tests/test_unicode_paths.py

import io
import http.client
import os
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
import urllib.parse
from pathlib import Path

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")

from PyQt5.QtWidgets import QApplication

from pgu4_sync import (
    DownloadJob,
    DownloadSignals,
    FAST_HTTP_SERVER_SCRIPT,
    FileTask,
    TransferController,
    decode_fast_remote_path,
    encode_fast_remote_path,
    normalize_local_filename,
)


class RemoteBytes(io.BytesIO):
    def set_pipelined(self, _enabled):
        return None


class FakeSftp:
    def __init__(self, expected_path, payload):
        self.expected_path = expected_path
        self.payload = payload

    def open(self, path, mode):
        if mode != "rb":
            raise AssertionError(mode)
        if path != self.expected_path:
            raise AssertionError(path)
        return RemoteBytes(self.payload)

    def close(self):
        return None


class FakeTransport:
    def set_keepalive(self, _seconds):
        return None


class FakeClient:
    def __init__(self, sftp):
        self.sftp = sftp

    def get_transport(self):
        return FakeTransport()

    def open_sftp(self):
        return self.sftp

    def close(self):
        return None


class FakeFactory:
    def __init__(self, sftp):
        self.sftp = sftp

    def open_client(self):
        return FakeClient(self.sftp)


class UnicodePathTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.app = QApplication.instance() or QApplication([])

    def test_fast_remote_path_round_trip(self):
        remote_path = "/home/clear/中文源目录/图像 01.jpg"

        encoded = encode_fast_remote_path(remote_path)

        self.assertTrue(encoded.isascii())
        self.assertNotIn("中文", encoded)
        self.assertEqual(remote_path, decode_fast_remote_path(encoded))

    def test_local_filename_preserves_chinese_and_normalizes_unicode(self):
        decomposed = "Cafe\u0301_中文数据"

        normalized = normalize_local_filename(decomposed)

        self.assertEqual("Café_中文数据", normalized)

    def test_sftp_download_supports_chinese_source_and_target(self):
        payload = "中文路径传输正常".encode("utf-8")
        remote_path = "/home/clear/中文源目录/实验数据.bin"
        sftp = FakeSftp(remote_path, payload)
        signals = DownloadSignals()
        controller = TransferController()
        job = DownloadJob(
            FakeFactory(sftp),
            [],
            Path("."),
            1,
            signals,
            controller,
        )

        with tempfile.TemporaryDirectory(prefix="PGU-中文目标-") as temp_dir:
            local_path = Path(temp_dir) / "目标文件夹" / "实验数据.bin"
            local_path.parent.mkdir(parents=True)
            local_path.write_bytes(b"\x00" * len(payload))
            file_task = FileTask(remote_path, local_path, len(payload))
            job._file_locks[local_path] = threading.Lock()

            job._download_chunk_sftp_once(file_task, 0, len(payload))

            self.assertEqual(payload, local_path.read_bytes())

    def test_fast_http_service_reads_chinese_path_with_range(self):
        payload = "极速中文路径".encode("utf-8")
        token = "unicode-test-token"
        with tempfile.TemporaryDirectory(prefix="PGU-极速中文-") as temp_dir:
            root = Path(temp_dir)
            source = root / "中文源文件夹" / "图像 数据.bin"
            source.parent.mkdir(parents=True)
            source.write_bytes(payload)
            script = root / "fast_service.py"
            script.write_text(FAST_HTTP_SERVER_SCRIPT, encoding="utf-8")

            with socket.socket() as probe:
                probe.bind(("127.0.0.1", 0))
                port = probe.getsockname()[1]

            process = subprocess.Popen(
                [
                    sys.executable,
                    str(script),
                    "--host",
                    "127.0.0.1",
                    "--port",
                    str(port),
                    "--token",
                    token,
                ],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            )
            self.addCleanup(process.kill)
            deadline = time.time() + 4
            while time.time() < deadline:
                try:
                    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=0.3)
                    conn.request(
                        "GET",
                        "/health?token=" + urllib.parse.quote(token, safe=""),
                    )
                    response = conn.getresponse()
                    if response.status == 200 and response.read() == b"ok-v3":
                        conn.close()
                        break
                    conn.close()
                except OSError:
                    time.sleep(0.05)
            else:
                self.fail("fast HTTP test service did not start")

            encoded_path = encode_fast_remote_path(str(source))
            conn = http.client.HTTPConnection("127.0.0.1", port, timeout=2)
            request_path = (
                "/file?token="
                + urllib.parse.quote(token, safe="")
                + "&path_b64="
                + urllib.parse.quote(encoded_path, safe="")
            )
            conn.request("GET", request_path, headers={"Range": "bytes=3-8"})
            response = conn.getresponse()
            body = response.read()
            conn.close()

            self.assertEqual(206, response.status)
            self.assertEqual(payload[3:9], body)
            process.terminate()
            process.wait(timeout=2)