import asyncio
import base64
import os
import sys
from pathlib import Path

import pytest
import safe_radix32
from clarabot_crypto.group_config.block import GroupConfigBlockFlags, GroupConfigBlockMemberFlags

from clarabot_nano.common.crypto.keyring import NanoSelfAccountKeyring
from clarabot_nano.common.protocol.errors import (
    NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
    NANO_REQUEST_ERROR_HANDLER_RESET_REQUEST,
    NANO_REQUEST_ERROR_HANDLER_TARGET_EXISTS,
    NANO_REQUEST_ERROR_HANDLER_DENIED,
    NANO_REQUEST_ERROR_HANDLER_TARGET_NOT_FOUND,
)
from clarabot_nano.utility import parse_environment_flag
from clarabot_nano.utility.serialize import msgpack_dumps
from clarabot_nano_tests.testing.client.manager import GraphQLClientManager
from clarabot_nano_tests.testing.config import room_blocks_from_room_data
from clarabot_nano_tests.testing.core.instance import RequestTestingNanoInstanceHandler
from clarabot_nano_tests.testing.server.manager import DevServerManager
from clarabot_nano_tests.utility.path.utility import (
    ComplexStruct,
    is_remote_security_set,
    get_platform_alternative_acute_e,
)


@pytest.mark.skipif(parse_environment_flag("CBNANO_TEST_WITHOUT_SERVER"), reason="server-dependant tests disabled")
@pytest.mark.asyncio
async def test_server_request_process_fs(
    tmpdir,
    reset_user_app_dirs,
    dev_nano_root_dir,
    testing_server_manager: DevServerManager,
    testing_server_user0,
    testing_client_manager: GraphQLClientManager,
    testing_server_reset_data,
):
    await testing_server_manager.call_throttling_account(
        safe_radix32.encode_fw(testing_server_user0["account_id"]), reset=True
    )
    cs = ComplexStruct.new(tmpdir)
    valid_roots, invalid_roots = cs.split_roots_by_validity()
    r0: Path = valid_roots[0]
    r1: Path = valid_roots[1]

    u1_data = testing_server_reset_data["user_accounts"][1]
    u2_data = testing_server_reset_data["user_accounts"][2]
    u3_data = testing_server_reset_data["user_accounts"][3]
    d0_data = testing_server_reset_data["dist_accounts"][0]

    rooms = testing_server_user0["rooms"][:4]
    drives = [
        {
            "path": str(r0),
            "rooms": [safe_radix32.encode_fw(rooms[0]["id"]), safe_radix32.encode_fw(rooms[2]["id"])],
        },
        {
            "path": str(r1),
            "rooms": [safe_radix32.encode_fw(rooms[1]["id"])],
        },
    ]
    room_blocks = {
        safe_radix32.encode_fw(rooms[0]["id"]): room_blocks_from_room_data(rooms[0]),
        safe_radix32.encode_fw(rooms[1]["id"]): room_blocks_from_room_data(rooms[1]),
        safe_radix32.encode_fw(rooms[2]["id"]): room_blocks_from_room_data(rooms[2]),
    }
    async with RequestTestingNanoInstanceHandler(
        reset_user_app_dirs,
        dev_nano_root_dir,
        testing_server_manager,
        testing_server_user0,
        testing_client_manager,
        drives,
        room_blocks,
    ).run() as (pm, debug_info, gateway_port, gateway_key, session_id, remote_yaml, nano_id, u0wsc):
        # u0wsc = await testing_client_manager.create_logged_in_wsc(testing_server_user0)
        u1wsc = await testing_client_manager.create_logged_in_wsc(u1_data)
        u2wsc = await testing_client_manager.create_logged_in_wsc(u2_data)
        u3wsc = await testing_client_manager.create_logged_in_wsc(u3_data)
        d0wsc = await testing_client_manager.create_logged_in_wsc(d0_data)
        anon_s = await testing_client_manager.session("anon")
        anon_wsc = await testing_client_manager.wsc(anon_s)
        anon_kr = NanoSelfAccountKeyring.generate_anonymous()
        u0_kr = testing_server_user0["_acc_kr"]

        # setup access control for testing
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    {
                        "type": "ROOM-CFG",
                        "room_id": safe_radix32.encode_fw(rooms[0]["id"]),
                        "member_flags": {
                            safe_radix32.encode_fw(u1_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE,
                            safe_radix32.encode_fw(u2_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE_NEGATE,
                            safe_radix32.encode_fw(u3_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE
                            | GroupConfigBlockMemberFlags.DRIVE_WRITE,
                            safe_radix32.encode_fw(d0_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE
                            | GroupConfigBlockMemberFlags.DRIVE_UPLOAD,
                        },
                        "group_flags": GroupConfigBlockFlags.ALLOW_ANONYMOUS,
                    },
                    "dbg_fs0",
                    rooms[0]["id"],
                    None,
                    u0wsc,
                    u0_kr,
                    u0_kr,
                )
            ),
            rooms[0]["id"],
            None,
            u0wsc,
            u0_kr,
            u0_kr,
        )
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    {
                        "type": "ROOM-CFG",
                        "room_id": safe_radix32.encode_fw(rooms[1]["id"]),
                        "member_flags": {
                            safe_radix32.encode_fw(u1_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE,
                            safe_radix32.encode_fw(u2_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE_NEGATE,
                            safe_radix32.encode_fw(u3_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE
                            | GroupConfigBlockMemberFlags.DRIVE_WRITE,
                            safe_radix32.encode_fw(d0_data["account_id"]): GroupConfigBlockMemberFlags.ACTIVE
                            | GroupConfigBlockMemberFlags.DRIVE_UPLOAD,
                        },
                        "group_flags": GroupConfigBlockFlags.ALLOW_ANONYMOUS_NEGATE,
                    },
                    "dbg_fs0",
                    rooms[0]["id"],
                    None,
                    u0wsc,
                    u0_kr,
                    u0_kr,
                )
            ),
            rooms[1]["id"],
            None,
            u0wsc,
            u0_kr,
            u0_kr,
        )

        # FS-GET
        # check structural consistency
        async def _fetch(room_, root_path_, query_path_, allowed_, query_wsc_, query_kr_, depth=0, **req_kwargs):
            if query_kr_.id == u0_kr.id:
                query_room_kr = None
            else:
                query_room_kr = room_["_room_kr"]
            if depth > 2:
                return  # stop cyclic symlinks
            res_ = await testing_client_manager.listing_room(
                query_path_,
                room_["id"],
                query_wsc_,
                query_kr_,
                u0_kr,
                room_kr=query_room_kr,
                ls_path=root_path_,
                ls_allowed=allowed_,
                **req_kwargs,
            )
            dirs_, files_ = res_ if res_ else ([], [])
            for sd_ in dirs_:
                await _fetch(
                    room_,
                    root_path_ / sd_,
                    f"{query_path_}/{sd_}" if query_path_ else sd_,
                    allowed_,
                    query_wsc_,
                    query_kr_,
                    depth + 1,
                    **req_kwargs,
                )
            for f_ in files_:
                await testing_client_manager.download_room(
                    f"{query_path_}/{f_}" if query_path_ else f_,
                    room_["id"],
                    query_wsc_,
                    query_kr_,
                    u0_kr,
                    room_kr=query_room_kr,
                    file_path=root_path_ / f_,
                    **req_kwargs,
                )

        for room, root_path, rnx, server_error in [(rooms[0], r0, "", None), (rooms[1], r1, ".", "Invalid argument!")]:
            allowed = cs.get_allowed_items(root_path, True, True)
            await _fetch(room, root_path, rnx, allowed, u0wsc, u0_kr)
            await _fetch(room, root_path, rnx, allowed, u1wsc, u1_data["_acc_kr"])
            await _fetch(room, root_path, rnx, allowed, u2wsc, u2_data["_acc_kr"], server_error=server_error)
            await _fetch(
                room,
                root_path,
                rnx,
                allowed,
                anon_wsc,
                anon_kr,
                ephemeral_pk=anon_kr.get_public_key(anon_kr.CURVE25519),
                server_error=server_error,
            )

        # special file sizes (in terms of chunking)
        fp = r0 / "file_sizes_xy"
        for s, pc in [(0, 0), (1, 1), (500_000, 1), (500_001, 2), (1_000_000, 2), (1_000_001, 3)]:
            fp.write_bytes(os.urandom(s))
            await testing_client_manager.download_room(
                fp.name, rooms[0]["id"], u0wsc, u0_kr, u0_kr, file_path=fp, part_count=pc
            )

        # large file
        r0_kr = rooms[0]["_room_kr"]
        large_dir = r0 / "large"
        large_dir.mkdir()
        big_file = large_dir / "im_so_big.bin"

        def _write_big_file():
            with open(big_file, "wb") as f:
                for _ in range(151):
                    f.write(os.urandom(1420) * 751)

        await asyncio.get_running_loop().run_in_executor(None, _write_big_file)
        fp = "large/im_so_big.bin"
        await testing_client_manager.download_room(fp, rooms[0]["id"], u0wsc, u0_kr, u0_kr, file_path=big_file)
        await testing_client_manager.download_room(
            fp, rooms[0]["id"], u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, file_path=big_file
        )
        await testing_client_manager.download_room(
            fp, rooms[0]["id"], u2wsc, u2_data["_acc_kr"], u0_kr, r0_kr, file_path=big_file
        )
        await testing_client_manager.download_room(
            fp,
            rooms[0]["id"],
            anon_wsc,
            anon_kr,
            u0_kr,
            r0_kr,
            file_path=big_file,
            ephemeral_pk=anon_kr.get_public_key(anon_kr.CURVE25519),
        )

        # large list
        def _write_large_dir():
            for i in range(17323):
                (large_dir / f"{base64.urlsafe_b64encode(os.urandom(150)).decode('ascii')}-{i}").write_text(str(i))

        await asyncio.get_running_loop().run_in_executor(None, _write_large_dir)
        ls_allowed = set(d.name for d in large_dir.iterdir())
        await testing_client_manager.listing_room(
            "large", rooms[0]["id"], u0wsc, u0_kr, u0_kr, ls_path=large_dir, ls_allowed=ls_allowed
        )
        await testing_client_manager.listing_room(
            "large", rooms[0]["id"], u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, ls_path=large_dir, ls_allowed=ls_allowed
        )
        await testing_client_manager.listing_room(
            "large", rooms[0]["id"], u2wsc, u2_data["_acc_kr"], u0_kr, r0_kr, ls_path=large_dir, ls_allowed=ls_allowed
        )
        await testing_client_manager.listing_room(
            "large",
            rooms[0]["id"],
            anon_wsc,
            anon_kr,
            u0_kr,
            r0_kr,
            ls_path=large_dir,
            ls_allowed=ls_allowed,
            ephemeral_pk=anon_kr.get_public_key(anon_kr.CURVE25519),
        )

        for p in [None, [""]]:
            await testing_client_manager.listing_room(
                p, rooms[0]["id"], u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
            )
        for p in ["..", "/tmp", "_neve. exist where"]:
            await testing_client_manager.listing_room(
                p, rooms[0]["id"], u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_TARGET_NOT_FOUND
            )
        # TODO: create coverage for error branches that are hard to simulate

        # FS-PEEK
        assert await testing_client_manager.peek_room("large", rooms[0]["id"], u0wsc, u0_kr, u0_kr) == {
            "name": "large",
            "mtime": large_dir.stat().st_mtime,
        }
        assert await testing_client_manager.peek_room(fp, rooms[0]["id"], u0wsc, u0_kr, u0_kr) == {
            "name": big_file.name,
            "mtime": big_file.stat().st_mtime,
            "size": big_file.stat().st_size,
        }
        for p in [None, [""], ""]:
            await testing_client_manager.peek_room(
                p, rooms[0]["id"], u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
            )
        for p in ["..", "/tmp", "_neve. exist where", "_neve. exist where/even less"]:
            await testing_client_manager.peek_room(
                p, rooms[0]["id"], u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_TARGET_NOT_FOUND
            )

        aep, aea = get_platform_alternative_acute_e()

        # FS-MKD
        r0_id = rooms[0]["id"]
        prefix = "x_"
        for p in [None, [""], ".", "..", "/tmp", "new_d/newest_d"]:
            await testing_client_manager.mkdir_room(
                p, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
            )
        # create dir in root
        p = f"{prefix}new_{aep}d"
        p_alt = f"{prefix}new_{aea}d"
        new_dir = r0 / p
        assert not new_dir.exists()
        assert not (r0 / p_alt).exists()
        resp = await testing_client_manager.mkdir_room(p_alt, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert resp.get("norm_name") == p, f"{p} {resp}"
        assert new_dir.is_dir()
        if sys.platform.startswith("darwin"):
            # darwin HFS normalizes composition, the two formats will refer to the same path
            assert (r0 / p_alt).is_dir()
        else:
            assert not (r0 / p_alt).exists()
        await testing_client_manager.mkdir_room(
            p, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        # create file & check conflicts on it
        new_file = new_dir / f"{prefix}new_f"
        new_file.write_text("yes")
        await testing_client_manager.mkdir_room(
            f"{p}/{prefix}new_f", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        await testing_client_manager.mkdir_room(
            f"{p}/{prefix}new_f/xy", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        # create subdir
        new_sub_dir = new_dir / f"{prefix}newest_d"
        assert not new_sub_dir.exists()
        await testing_client_manager.mkdir_room(p := f"{p}/{prefix}newest_d", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert new_sub_dir.is_dir()
        await testing_client_manager.mkdir_room(
            p, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        # deny unauthorized
        p = "just-another-dir"
        await testing_client_manager.mkdir_room(
            p, r0_id, "mkd-u1", u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )
        await testing_client_manager.mkdir_room(
            p, r0_id, "mkd-u2", u2wsc, u2_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )
        await testing_client_manager.mkdir_room(
            p, r0_id, "mkd-d0", d0wsc, d0_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )

        # FS-DEL
        for p in [None, [""], ".", "..", "/tmp", "new_d/newest_d", "./not_anything"]:
            await testing_client_manager.del_room(
                p, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
            )
        # test delete simple file
        del_f0 = new_dir / f"{prefix}f_me"
        del_f0.write_text("fme")
        assert del_f0.is_file()
        await testing_client_manager.del_room(f"{new_dir.name}/{prefix}f_me", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not del_f0.exists()
        # test delete empty dir
        del_d0 = new_dir / f"{prefix}del_d0"
        del_d0.mkdir()
        assert del_d0.is_dir()
        await testing_client_manager.del_room(f"{new_dir.name}/{prefix}del_d0", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not del_d0.exists()
        # test delete non-empty dir
        del_d0.mkdir()
        (del_d0 / "sd").mkdir()
        (del_d0 / "sf").write_text("sf")
        assert del_d0.is_dir()
        await testing_client_manager.del_room(f"{new_dir.name}/{prefix}del_d0", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not del_d0.exists()
        # test delete links directly
        del_d0.mkdir()
        del_sd = del_d0 / "sd"
        del_sd.mkdir()
        del_sf = del_d0 / "sf"
        del_sf.write_text("sf")
        del_ln = r0 / f"{prefix}del_ln"
        # test delete direct link to file
        del_ln.symlink_to(del_sf)
        assert del_ln.is_symlink()
        assert del_ln.is_file()
        assert del_sf.is_file()
        await testing_client_manager.del_room(f"{del_ln.name}", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not del_ln.exists()
        assert del_sf.is_file()
        # test delete direct link to directory
        del_ln.symlink_to(del_d0)
        assert del_ln.is_symlink()
        assert del_ln.is_dir()
        assert del_d0.is_dir()
        assert del_sf.is_file()
        await testing_client_manager.del_room(f"{del_ln.name}", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not del_ln.exists()
        assert del_d0.is_dir()
        assert del_sf.is_file()
        # test delete folder with symlink inside and keep target
        del_ln = del_sd / f"{prefix}del_ln"
        del_ln.symlink_to(del_sf)
        assert del_ln.is_symlink()
        assert del_ln.is_file()
        assert del_sd.is_dir()
        assert del_sf.is_file()
        await testing_client_manager.del_room(
            f"{new_dir.name}/{del_d0.name}/{del_sd.name}", r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr
        )
        assert not del_ln.exists()
        assert not del_sd.exists()
        assert del_sf.is_file()
        # deny unauthorized
        p = f"{new_dir.name}"
        await testing_client_manager.del_room(
            p, r0_id, "mkd-u1", u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )
        await testing_client_manager.del_room(
            p, r0_id, "mkd-u2", u2wsc, u2_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )
        await testing_client_manager.del_room(
            p, r0_id, "mkd-d0", d0wsc, d0_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )

        # FS-PUT
        # check errors
        correct = {
            "path": prefix + "just_put_in_root",
            "size": 5000,
            "mtime": 0,
            "offset": 0,
            "data": os.urandom(100),
        }
        for err_t in [
            {"path": None},
            {"path": [""]},
            {"path": ".."},
            {"path": "/tmp"},
            {"size": "10MB"},
            {"size": -500},
            {"mtime": "yesterday"},
            {"mtime": -500},
            {"offset": "start"},
            {"offset": -1},
            {"data": "yes"},
            {"data": 0},
            {"smac": "a" * 16},
            {"smac": b"a" * 15},
            {"smac": 0},
            {"overwrite": "a"},
            {"overwrite": b"a"},
        ]:
            req = dict(correct)
            req.update(err_t)
            await testing_client_manager.put_room(
                req, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
            )
        # put single part file
        await testing_client_manager.put_room(
            dict(
                path=(path := prefix + "put_in_root0"),
                size=-1,
                mtime=-1,
                offset=0,
                data=(data := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data]),
            ),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
        )
        assert (r0 / path).read_bytes() == data
        assert is_remote_security_set(r0 / path) is True
        # start error (no tracking)
        await testing_client_manager.put_room(
            dict(
                path=(path := prefix + "put_in_root1"),
                size=-1,
                mtime=-1,
                offset=500,
                data=os.urandom(100),
            ),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_RESET_REQUEST,
        )
        # write first part for multi part integrity errors
        await testing_client_manager.put_room(
            dict(path=path, size=-1, mtime=-1, offset=0, data=os.urandom(100)),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
        )
        await testing_client_manager.put_room(
            dict(
                path=path,
                size=120,  # reset by size
                mtime=-1,
                offset=100,
                data=os.urandom(100),
            ),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_RESET_REQUEST,
        )
        await testing_client_manager.put_room(
            dict(
                path=path,
                size=-1,
                mtime=999999,  # reset by size
                offset=100,
                data=os.urandom(100),
            ),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_RESET_REQUEST,
        )
        # reset existing target
        await testing_client_manager.put_room(
            dict(path=path, size=-1, mtime=-1, offset=0, data=os.urandom(100)),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
        )
        # part gap integrity error
        await testing_client_manager.put_room(
            dict(path=path, size=-1, mtime=-1, offset=200, data=os.urandom(100), smac=b"0" * 16),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_RESET_REQUEST,
        )
        # reset existing target 2
        await testing_client_manager.put_room(
            dict(path=path, size=-1, mtime=-1, offset=0, data=os.urandom(100)),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
        )
        # sum mac integrity error
        await testing_client_manager.put_room(
            dict(path=path, size=-1, mtime=-1, offset=100, data=os.urandom(100), smac=b"0" * 16),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_RESET_REQUEST,
        )
        # multi part large upload
        path = prefix + f"put_in_{aep}_root2"
        path_alt = prefix + f"put_in_{aea}_root2"
        data_list = []
        offset = 0
        for i in range(7):
            data_list.append(data := os.urandom(1001))
            resp = await testing_client_manager.put_room(
                dict(
                    path=(path_alt if i % 2 == 0 else path),
                    size=7007,
                    mtime=1612089337617,
                    offset=offset,
                    data=data,
                    smac=testing_client_manager.calc_smac(data_list) if i == 6 else None,
                ),
                r0_id,
                "mkd-u0",
                u0wsc,
                u0_kr,
                u0_kr,
            )
            if offset == 0:
                assert resp.get("norm_name") == path, f"{path} {resp}"
            else:
                assert "norm_name" not in resp, str(resp)
            offset += len(data)
        assert (r0 / path).read_bytes() == b"".join(data_list)
        # overwrite error
        for p_ in [path, path_alt]:
            await testing_client_manager.put_room(
                dict(
                    path=p_,
                    size=-1,
                    mtime=-1,
                    offset=0,
                    data=(data := os.urandom(100)),
                    smac=testing_client_manager.calc_smac([data]),
                ),
                r0_id,
                "mkd-u0",
                u0wsc,
                u0_kr,
                u0_kr,
                error=NANO_REQUEST_ERROR_HANDLER_TARGET_EXISTS,
            )
        assert (r0 / path).read_bytes() == b"".join(data_list)
        resp = await testing_client_manager.put_room(
            dict(
                path=path,
                size=-1,
                mtime=-1,
                offset=0,
                data=(data := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data]),
                overwrite=True,
            ),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
        )
        assert resp.get("norm_name") == path, f"{path} {resp}"
        assert (r0 / path).read_bytes() == data
        if sys.platform.startswith("darwin"):
            # darwin HFS normalizes composition, the two formats will refer to the same path
            assert (r0 / path_alt).read_bytes() == data
        else:
            assert not (r0 / path_alt).exists()
        # existing
        path = prefix + "put_in_root3"
        (r0 / path).mkdir()
        await testing_client_manager.put_room(
            dict(
                path=path,
                size=-1,
                mtime=-1,
                offset=0,
                data=(data := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data]),
            ),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_TARGET_EXISTS,
        )
        # deny unauthorized
        await testing_client_manager.put_room(
            dict(
                path=prefix + "put_in_root_xkl",
                size=-1,
                mtime=-1,
                offset=0,
                data=(data := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data]),
            ),
            r0_id,
            "mkd-u1",
            u1wsc,
            u1_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_DENIED,
        )
        await testing_client_manager.put_room(
            dict(
                path=prefix + "put_in_root_xkl",
                size=-1,
                mtime=-1,
                offset=0,
                data=(data := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data]),
            ),
            r0_id,
            "mkd-u2",
            u2wsc,
            u2_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_DENIED,
        )
        path_perm0 = prefix + "put_perm_t0"
        path_perm1 = prefix + "put_perm_t1"
        await testing_client_manager.put_room(
            dict(
                path=path_perm0,
                size=-1,
                mtime=-1,
                offset=0,
                data=(data0 := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data0]),
            ),
            r0_id,
            "mkd-u3",
            u3wsc,
            u3_data["_acc_kr"],
            u0_kr,
            r0_kr,
        )
        await testing_client_manager.put_room(
            dict(
                path=path_perm1,
                size=-1,
                mtime=-1,
                offset=0,
                data=(data1 := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data1]),
            ),
            r0_id,
            "mkd-d0",
            d0wsc,
            d0_data["_acc_kr"],
            u0_kr,
            r0_kr,
        )
        assert (r0 / path_perm0).read_bytes() == data0
        assert (r0 / path_perm1).read_bytes() == data1
        await testing_client_manager.put_room(
            dict(
                path=path_perm0,
                size=-1,
                mtime=-1,
                offset=0,
                data=(data := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data]),
            ),
            r0_id,
            "mkd-d0",
            d0wsc,
            d0_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_DENIED,
        )
        await testing_client_manager.put_room(
            dict(
                path=path_perm0,
                size=-1,
                mtime=-1,
                offset=0,
                data=(data := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data]),
                overwrite=True,
            ),
            r0_id,
            "mkd-d0",
            d0wsc,
            d0_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_DENIED,
        )
        assert (r0 / path_perm0).read_bytes() == data0
        assert (r0 / path_perm1).read_bytes() == data1
        await testing_client_manager.put_room(
            dict(
                path=path_perm1,
                size=-1,
                mtime=-1,
                offset=0,
                data=(data1 := os.urandom(100)),
                smac=testing_client_manager.calc_smac([data1]),
                overwrite=True,
            ),
            r0_id,
            "mkd-u3",
            u3wsc,
            u3_data["_acc_kr"],
            u0_kr,
            r0_kr,
        )
        assert (r0 / path_perm0).read_bytes() == data0
        assert (r0 / path_perm1).read_bytes() == data1

        await asyncio.sleep(8)

        # FS-MOV
        prefix = "mx_"
        mov_f0 = r0 / f"{prefix}f_0"
        mov_f0.write_text("f_0")
        mov_f1 = r0 / f"{prefix}f_1"
        assert mov_f0.is_file()
        assert not mov_f1.exists()
        for p in [None, [""], ".", "..", "/tmp", "new_d/newest_d", "./not_anything"]:
            for t in [mov_f0.name, mov_f1.name]:
                await testing_client_manager.move_room(
                    p, t, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
                )
            if p not in ["new_d/newest_d", "./not_anything"]:
                await testing_client_manager.move_room(
                    mov_f0.name, p, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
                )
        await testing_client_manager.move_room(
            mov_f1.name, mov_f0.name, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        assert mov_f0.is_file()
        assert not mov_f1.exists()
        # test move simple file and check remote-security
        if not sys.platform.startswith("win"):
            mov_f0.chmod(0o754)
        assert is_remote_security_set(mov_f0) is False
        await testing_client_manager.move_room(mov_f0.name, mov_f1.name, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not mov_f0.exists()
        assert mov_f1.read_text() == "f_0"
        assert is_remote_security_set(mov_f1) is True
        # test move into and out of directory
        mov_d0 = r0 / f"{prefix}d_0"
        mov_d0.mkdir()
        mov_d00 = mov_d0 / "d0"
        mov_d00.mkdir()
        await testing_client_manager.move_room(
            mov_f1.name, mov_d0.name, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        await testing_client_manager.move_room(
            mov_f1.name,
            mov_d00.relative_to(r0).as_posix(),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
        )
        assert not mov_f0.exists()
        assert mov_f1.read_text() == "f_0"
        d00fx = mov_d00.relative_to(r0) / "fx"
        await testing_client_manager.move_room(mov_f1.name, d00fx.as_posix(), r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not mov_f0.exists()
        assert not mov_f1.exists()
        assert (r0 / d00fx).read_text() == "f_0"
        d0fy = mov_d0.relative_to(r0) / "fy"
        await testing_client_manager.move_room(d00fx.as_posix(), d0fy.as_posix(), r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not mov_f0.exists()
        assert not mov_f1.exists()
        assert not (r0 / d00fx).exists()
        assert (r0 / d0fy).read_text() == "f_0"
        # test move directory
        assert mov_d00.is_dir()
        mov_d1 = r0 / f"{prefix}d_1"
        await testing_client_manager.move_room(
            mov_d00.relative_to(r0).as_posix(),
            (mov_d1 / mov_d00.name).relative_to(r0).as_posix(),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
        )
        assert not mov_d1.exists()
        mov_d1.mkdir()
        await testing_client_manager.move_room(
            mov_d00.relative_to(r0).as_posix(),
            (mov_d1 / mov_d00.name).relative_to(r0).as_posix(),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
        )
        assert mov_d0.is_dir()
        assert not mov_d00.exists()
        assert set(x.name for x in mov_d1.iterdir()) == {mov_d00.name}
        await testing_client_manager.move_room(
            mov_d0.relative_to(r0).as_posix(),
            (mov_d1 / (mov_d0.name + "X")).relative_to(r0).as_posix(),
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
        )
        assert not mov_d0.exists()
        assert not mov_d00.exists()
        assert set(x.name for x in mov_d1.iterdir()) == {mov_d0.name + "X", mov_d00.name}
        assert set(x.name for x in (mov_d1 / (mov_d0.name + "X")).iterdir()) == {"fy"}
        assert (mov_d1 / (mov_d0.name + "X") / "fy").read_text() == "f_0"
        await testing_client_manager.move_room(mov_d1.name, mov_d0.name, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not mov_d1.exists()
        assert mov_d0.is_dir()
        mov_d1.mkdir()
        await testing_client_manager.move_room(
            mov_d0.name,
            mov_d1.name,
            r0_id,
            "mkd-u0",
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
        )
        assert mov_d0.is_dir()
        assert mov_d1.is_dir()
        assert not list(mov_d1.iterdir())
        # test path name normalization
        uni_d = r0 / f"{prefix}_uni_{aep}"
        uni_d_alt = r0 / f"{prefix}_uni_{aea}"
        assert not uni_d.exists()
        assert not uni_d_alt.exists()
        resp = await testing_client_manager.move_room(mov_d0.name, uni_d_alt.name, r0_id, "mkd-u0", u0wsc, u0_kr, u0_kr)
        assert not mov_d0.exists()
        assert resp.get("norm_name") == uni_d.name, f"{uni_d.name} {resp}"
        assert uni_d.is_dir()
        if sys.platform.startswith("darwin"):
            # darwin HFS normalizes composition, the two formats will refer to the same path
            assert uni_d_alt.is_dir()
        else:
            assert not uni_d_alt.exists()
        # deny unauthorized
        p = mov_d0.name
        await testing_client_manager.move_room(
            p, "a", r0_id, "mkd-u1", u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )
        await testing_client_manager.move_room(
            p, "a", r0_id, "mkd-u2", u2wsc, u2_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )
        await testing_client_manager.move_room(
            p, "a", r0_id, "mkd-d0", d0wsc, d0_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_HANDLER_DENIED
        )
