import asyncio
import os
import sys
import uuid
from pathlib import Path

import pytest
import safe_radix32
from clarabot_crypto.context.account_trust import ACCOUNT_TRUST_CRYPTO_CONTEXT, AccountTrustConfigDumpArgs
from clarabot_crypto.group_config.block import GroupConfigBlockMemberFlags, GroupConfigBlockFlags
from clarabot_testing.common.gql import graphql_request_payload

from clarabot_nano import __version__
from clarabot_nano.common.core.message import MSG_TYPE_GATEWAY_LOGIN
from clarabot_nano.common.crypto.keyring import NanoSelfAccountKeyring
from clarabot_nano.common.protocol.errors import (
    NANO_REQUEST_ERROR_MUTATION_ID,
    NANO_REQUEST_ERROR_ADMIN_PASSWORD,
    NANO_REQUEST_ERROR_ADMIN_PASSWORD_MISSING,
    NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
    NANO_REQUEST_ERROR_PAYLOAD_ROOM_INVALID,
    NANO_REQUEST_ERROR_DENIED,
    NANO_REQUEST_ERROR_HANDLER_DENIED,
    NANO_REQUEST_ERROR_PEER_NOT_TRUSTED,
    NANO_REQUEST_ERROR_TOO_MANY_REQUESTS,
)
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.client.response import get_initial_context_truth_for_dbg_handler
from clarabot_nano_tests.testing.config import room_blocks_from_room_data
from clarabot_nano_tests.testing.core.common import clear_nano_slots
from clarabot_nano_tests.testing.file import temp_path
from clarabot_nano_tests.testing.gateway import start_nano_wait_gateway, do_gateway_login, do_gateway_logout
from clarabot_nano_tests.testing.server.manager import DevServerManager
from clarabot_nano_tests.testing.utility import delay_retry_coroutine_assertion
from clarabot_nano_tests.utility.path.utility import 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_gateway_connect_bind(
    tmpdir,
    reset_user_app_dirs,
    dev_nano_root_dir,
    testing_server_manager: DevServerManager,
    testing_server_user0,
    testing_client_manager: GraphQLClientManager,
    testing_server_reset_data,
    testing_server_user_norm_pass,  # check password normalization at login
    testing_server_user_not_norm_pass,  # check password normalization at login (not normalized, legacy)
):
    u0_id_sr32 = safe_radix32.encode_fw(testing_server_user0["account_id"])
    u0_kr = testing_server_user0["_acc_kr"]
    await clear_nano_slots(testing_server_manager, [testing_server_user0["account_id"]])

    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]

    rooms = [testing_server_user0["rooms"][i] for i in range(4)]

    u0wsc = await testing_client_manager.create_logged_in_wsc(testing_server_user0)
    assert not u0wsc.ws.closed

    session = await testing_server_manager.session()  # debug client session
    async with start_nano_wait_gateway(
        testing_server_manager,
        reset_user_app_dirs,
        dev_nano_root_dir,
        "waiting-config",
    ) as (pm, debug_info, gateway_port, gateway_key):
        # bad login requests/credentials
        last_command_number = (await pm.wait_dbg_states({})).get("gateway_command_result_number")
        for request, result in [
            ({"username": "nobody", "password": "invliad"}, "invalid or missing email"),
            ({"email": "", "password": "invliad"}, "invalid or missing email"),
            ({"email": 42, "password": "invliad"}, "invalid or missing email"),
            ({"email": "nobody", "pass": "invliad"}, "invalid or missing password"),
            ({"email": "nobody", "password": ""}, "invalid or missing password"),
            ({"email": "nobody", "password": 42}, "invalid or missing password"),
            ({"email": "nobody", "password": "invliad"}, "login failed"),
        ]:
            await pm.assert_core_request(session, debug_info["port"], MSG_TYPE_GATEWAY_LOGIN, request)
            try:
                state = await pm.wait_dbg_states(
                    [
                        ("gateway_port", gateway_port),
                        ("gateway_key", gateway_key),
                        ("gateway_status", "waiting-config"),
                        ("gateway_command_result", result),
                        ("gateway_command_result_number", lambda v: v != last_command_number),
                    ]
                )
            except AssertionError as e:
                raise Exception(f"R: {request} {result} {last_command_number}") from e
            last_command_number = state["gateway_command_result_number"]

        await testing_server_manager.wait_nano_bind(testing_server_user0["account_id"], {})
        assert pm.get_session_id_from_continue_token() is None

        # successful login
        session_id = await do_gateway_login(
            testing_server_manager, pm, debug_info, gateway_port, gateway_key, testing_server_user0
        )
        await testing_server_manager.wait_nano_bind(testing_server_user0["account_id"], {session_id: set()})

        assert not u0wsc.ws.closed
        # test reconnect by server restart
        await asyncio.gather(
            testing_server_manager.call_asgi_reload(),
            pm.wait_dbg_states(
                [
                    ("gateway_port", gateway_port),
                    ("gateway_key", gateway_key),
                    ("gateway_status", "disconnected"),
                ],
                timeout=15.0,
            ),
        )
        await pm.wait_dbg_states(
            [
                ("gateway_port", gateway_port),
                ("gateway_key", gateway_key),
                ("gateway_status", "active"),
            ],
            timeout=15.0,
        )
        assert session_id == pm.get_session_id_from_continue_token()
        await testing_server_manager.wait_nano_bind(testing_server_user0["account_id"], {session_id: set()})
        assert u0wsc.ws.closed
        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)
        anon_s = await testing_client_manager.session("anon")
        anon_wsc = await testing_client_manager.wsc(anon_s)

        dbg_config = await pm.get_wait_dbg_config()
        # test setup serve-admin api
        nano_id = await testing_server_manager.get_nano_id_by_nano_session_id(
            testing_server_user0["account_id"], session_id
        )
        assert nano_id is not None
        r = {"type": "debug-echo-sp", "hello": "me"}
        await testing_client_manager.nano_request_response(
            msgpack_dumps(r), None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_MUTATION_ID
        )
        r = await testing_client_manager.prepare_request_with_mutation_id(
            {"type": "debug-echo-sp", "hello": "me"}, "dbg0", None, nano_id, u0wsc, u0_kr, u0_kr
        )
        req, resps = await testing_client_manager.nano_request_responses_raw(
            msgpack_dumps(r), None, nano_id, u0wsc, u0_kr, u0_kr
        )
        assert len(resps) == 1
        resp = resps[0]
        assert isinstance(tmp := resp["response"]["context"].pop("remote_ip"), (bytes, str)), f"{tmp}"
        resp_target = {
            "context": get_initial_context_truth_for_dbg_handler(req, u0_kr, testing_server_user0["account_id"]),
            "payload": r,
        }
        assert resp["response"] == resp_target, f"{resp['response']}\n{resp_target}"
        part_size = resp["response"]["context"]["part_size"]
        resp = await testing_client_manager.nano_request_response(
            msgpack_dumps({"type": "A-INFO"}), None, nano_id, u0wsc, u0_kr, u0_kr
        )
        assert resp == {
            "name": dbg_config["remote"]["name"],
            "version": __version__,
        }
        # can't query remote config without admin password
        req_rem_conf = msgpack_dumps({"type": "A-REM-CFG"})
        await testing_client_manager.nano_request_response(
            req_rem_conf, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_ADMIN_PASSWORD_MISSING
        )
        # can't set admin password without mutation id
        a_pwd = "best-password"
        await testing_client_manager.nano_request_response(
            msgpack_dumps({"type": "A-CH-PWD", "new_password": a_pwd}),
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_MUTATION_ID,
        )
        # set admin password with mutation id
        r = await testing_client_manager.prepare_request_with_mutation_id(
            {"type": "A-CH-PWD", "new_password": a_pwd}, "dbg0", None, nano_id, u0wsc, u0_kr, u0_kr
        )
        await testing_client_manager.nano_request_response(msgpack_dumps(r), None, nano_id, u0wsc, u0_kr, u0_kr)
        # can't query remote config without admin password
        await testing_client_manager.nano_request_response(
            req_rem_conf, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_ADMIN_PASSWORD
        )
        # can't query remote config with incorrect admin password
        req_rem_conf = msgpack_dumps({"type": "A-REM-CFG", "admin_password": "a_pwd"})
        await testing_client_manager.nano_request_response(
            req_rem_conf, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_ADMIN_PASSWORD
        )
        # query remote config with admin password
        rem_conf_keys = {
            "deny_anonymous",
            "drives",
            "indexed_languages",
            "indexer_extract_size_limit",
            "indexer_extract_text_limit",
            "indexer_force_sync_on_startup",
            "indexer_minimum_delay_between_path_sync",
            "indexer_ignored_mimes",
            "indexer_remove_lost_count",
            "indexer_reserved_space",
            "indexer_sync_interval",
            "name",
            "remote_files_restriction",
            "room_blocks",
            "server_process_number",
        }

        async def _check_remote_config_query(pwd):
            req_rem_conf = msgpack_dumps({"type": "A-REM-CFG", "admin_password": pwd})
            rem_conf = await testing_client_manager.nano_request_response(
                req_rem_conf, None, nano_id, u0wsc, u0_kr, u0_kr
            )
            assert set(rem_conf.keys()) == {"config"}, str(rem_conf)
            rem_conf = rem_conf["config"]
            assert set(rem_conf.keys()) == rem_conf_keys, str(rem_conf)
            _dbg_config = await pm.get_wait_dbg_config()
            for k, v in rem_conf.items():
                assert _dbg_config["remote"].get(k) == v, f"{k}\n{_dbg_config['remote']}\n{rem_conf}"

        await _check_remote_config_query(a_pwd)
        # can't change admin password without the admin password (tests incrementation of mutation seq with error)
        r = await testing_client_manager.prepare_request_with_mutation_id(
            {"type": "A-CH-PWD", "new_password": "a_pwd"}, "dbg0", None, nano_id, u0wsc, u0_kr, u0_kr
        )
        await testing_client_manager.nano_request_response(
            msgpack_dumps(r), None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_ADMIN_PASSWORD
        )
        # update the admin password
        r = await testing_client_manager.prepare_request_with_mutation_id(
            {"type": "A-CH-PWD", "new_password": "a_pwd", "admin_password": a_pwd},
            "dbg0",
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
        )
        await testing_client_manager.nano_request_response(msgpack_dumps(r), None, nano_id, u0wsc, u0_kr, u0_kr)
        a_pwd = "a_pwd"
        # change name
        await testing_client_manager.single_response_admin_request(
            {"type": "A-EDIT-N", "name": "Öreg ördög"}, "dbg0", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr
        )
        dbg_config = await pm.get_wait_dbg_config()
        assert dbg_config["remote"]["name"] == "Öreg ördög"
        # test listing directories by admin power
        await testing_client_manager.listing_response_admin_request("", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr)
        await testing_client_manager.listing_response_admin_request(
            "relative", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        aep, aea = get_platform_alternative_acute_e()
        drives_root = Path(str(temp_path(tmpdir)))
        drive_a = drives_root / "da"
        drive_b = drives_root / "db"
        drive_c = drives_root / f"dc{aep}"
        drive_c_alt = drives_root / f"dc{aea}"
        file_c = drives_root / "fc"
        os.makedirs(drive_a)
        os.makedirs(drive_b)
        file_c.write_text("some", encoding="utf-8")
        (drive_b / "noice").write_text("yes", encoding="utf-8")
        name_prefix = f"{uuid.uuid4().hex}_{uuid.uuid4().hex}_"
        for i in range((part_size // len(name_prefix)) * 3):
            (drives_root / f"{name_prefix}{i}").mkdir()
        await testing_client_manager.listing_response_admin_request(
            drives_root.as_posix(), a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr
        )
        tmp = drive_b.as_posix().replace("/", "\x00")  # something that is invalid on all systems
        await testing_client_manager.listing_response_admin_request(
            tmp, a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        tmp = file_c.as_posix()
        await testing_client_manager.listing_response_admin_request(
            tmp, a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        await testing_client_manager.listing_response_admin_request(
            tmp + "ax", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        await testing_client_manager.listing_response_admin_request(
            drive_b.as_posix(), a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr
        )
        # create directory fails with invalid path/name
        for path, name in [
            ("invalid path", "ok"),
            (drives_root.as_posix(), ""),
            (drives_root.as_posix(), "."),
            (None, "ok"),
            (drives_root.as_posix(), ["x"]),
            (drives_root.as_posix(), drive_a.name),
            (drives_root.as_posix(), file_c.name),
        ]:
            await testing_client_manager.single_response_admin_request(
                {"type": "A-MKD", "path": path, "name": name},
                "dbg0",
                a_pwd,
                None,
                nano_id,
                u0wsc,
                u0_kr,
                u0_kr,
                error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
                error_before_deny=not isinstance(path, str) or not isinstance(name, str),
            )
        assert not drive_c.exists()
        assert not drive_c_alt.exists()
        mkd_req = {"type": "A-MKD", "path": drives_root.as_posix(), "name": drive_c_alt.name}
        await testing_client_manager.single_response_admin_request(
            mkd_req, "dbg0", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr
        )
        assert drive_c.is_dir()
        if sys.platform.startswith("darwin"):
            # darwin HFS normalizes composition, the two formats will refer to the same path
            assert drive_c_alt.exists()
        else:
            assert not drive_c_alt.exists()
        await testing_client_manager.single_response_admin_request(
            mkd_req, "dbg0", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST
        )
        # create drive fails for invalid/sensitive path
        for rp in ["invalid path", "C:/_.wontexist" if sys.platform.startswith("win") else "/_.wontexist"]:
            await testing_client_manager.single_response_admin_request(
                {"type": "A-D-CREATE", "path": rp},
                "dbg0",
                a_pwd,
                None,
                nano_id,
                u0wsc,
                u0_kr,
                u0_kr,
                error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
            )
        # link to sensitive path
        lnr = drives_root / "ln2r"
        os.symlink("C:/" if sys.platform.startswith("win") else "/", lnr, target_is_directory=True)
        for rp in ["C:/", "C:/WinDowS", str(lnr)] if sys.platform.startswith("win") else ["/", "/root", str(lnr)]:
            await testing_client_manager.single_response_admin_request(
                {"type": "A-D-CREATE", "path": rp},
                "dbg0",
                a_pwd,
                None,
                nano_id,
                u0wsc,
                u0_kr,
                u0_kr,
                error=NANO_REQUEST_ERROR_HANDLER_DENIED,
            )
        # create drives
        await testing_client_manager.single_response_admin_request(
            {"type": "A-D-CREATE", "path": str(drive_a)}, "dbg0", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr
        )
        await testing_client_manager.single_response_admin_request(
            {"type": "A-D-CREATE", "path": drive_b.as_posix()}, "dbg0", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr
        )
        dbg_config = await pm.get_wait_dbg_config()
        assert dbg_config["remote"]["drives"] == [{"path": str(drive_a)}, {"path": str(drive_b)}]
        # delete a drive
        await testing_client_manager.single_response_admin_request(
            {"type": "A-D-DELETE", "path": str(file_c)},
            "dbg0",
            a_pwd,
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
        )
        await testing_client_manager.single_response_admin_request(
            {"type": "A-D-DELETE", "path": str(drive_a)}, "dbg0", a_pwd, None, nano_id, u0wsc, u0_kr, u0_kr
        )
        await testing_client_manager.single_response_admin_request(
            {"type": "A-D-DELETE", "path": str(drive_a)},
            "dbg0",
            a_pwd,
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
        )
        dbg_config = await pm.get_wait_dbg_config()
        assert dbg_config["remote"]["drives"] == [{"path": str(drive_b)}]

        # === room-access control checking ===

        r0_id = rooms[0]["id"]
        perm_req0 = {
            "type": "ROOM-CFG",
            "room_id": safe_radix32.encode_fw(r0_id),
            "member_flags": {
                # NOTE: flag changes on owner are N/A, but not forbidden; this is here to test key-crypto
                u0_id_sr32: GroupConfigBlockMemberFlags.ACTIVE | GroupConfigBlockMemberFlags.ADMINISTRATOR,
                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.ADMINISTRATOR,
            },
            "group_flags": GroupConfigBlockFlags.ALLOW_ANONYMOUS_NEGATE,
        }
        # owner cannot mutate room that is not served and not using direct request
        await testing_client_manager.single_response_admin_request(
            perm_req0,
            "dbg0",
            a_pwd,
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_PAYLOAD_ROOM_INVALID,
        )
        # room_id must be rad32
        await testing_client_manager.single_response_admin_request(
            {"type": "A-D-ATTACH", "path": str(drive_b), "target_room_id": r0_id, "room_blocks": []},
            "dbg0",
            a_pwd,
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_BAD_REQUEST,
        )
        # attach with verified blocks
        await testing_client_manager.single_response_admin_request(
            {
                "type": "A-D-ATTACH",
                "path": str(drive_b),
                "target_room_id": safe_radix32.encode_fw(r0_id),
                "room_blocks": room_blocks_from_room_data(rooms[0]),
            },
            "dbg0",
            a_pwd,
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
        )
        # owner cannot mutate room that is served by using direct request
        await testing_client_manager.single_response_admin_request(
            perm_req0,
            "dbg0",
            a_pwd,
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_PAYLOAD_ROOM_INVALID,
        )
        # owner is rate-limited to ask anything
        await testing_server_manager.call_throttling_account(u0_id_sr32)
        await testing_client_manager.nano_request_responses_raw(
            b"asdf", None, nano_id, u0wsc, u0_kr, u0_kr, server_error="429 Too Many Requests"
        )
        await testing_server_manager.call_throttling_account(u0_id_sr32, reset=True)
        await testing_server_manager.call_throttling_account(u0_id_sr32, names=["query"])
        # owner mutates room permissions (nano is 429 for query so it fails)
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    perm_req0, "dbg0", r0_id, None, u0wsc, u0_kr, u0_kr
                )
            ),
            r0_id,
            None,
            u0wsc,
            u0_kr,
            u0_kr,
            error=NANO_REQUEST_ERROR_TOO_MANY_REQUESTS,
        )
        await testing_server_manager.call_throttling_account(u0_id_sr32, reset=True, names=["query"])
        # owner mutates room permissions
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    perm_req0, "dbg0", r0_id, None, u0wsc, u0_kr, u0_kr
                )
            ),
            r0_id,
            None,
            u0wsc,
            u0_kr,
            u0_kr,
        )
        # check room ls query by different identities
        r0_ls_nr = msgpack_dumps({"type": "FS-GET", "path": ""})
        await testing_client_manager.nano_request_response(
            r0_ls_nr, r0_id, None, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_PAYLOAD_ROOM_INVALID
        )
        r0_kr = rooms[0]["_room_kr"]
        await testing_client_manager.listing_room("", r0_id, u0wsc, u0_kr, u0_kr, ls_path=drive_b)
        # server does not even forward not authorized requests
        await testing_client_manager.listing_room(
            "", r0_id, u2wsc, u2_data["_acc_kr"], u0_kr, r0_kr, server_error="Invalid argument!"
        )
        # authorized member
        await testing_client_manager.listing_room("", r0_id, u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, ls_path=drive_b)
        # test a trusted peer
        await testing_client_manager.assert_gql_okay(
            u0wsc,
            graphql_request_payload(
                "mutation addTrust {", peerId=safe_radix32.encode_fw(u1_data["account_id"]), trust=b"x" * 48
            ),
        )
        await delay_retry_coroutine_assertion(
            testing_client_manager.listing_room,
            "",
            r0_id,
            u1wsc,
            u1_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_PEER_NOT_TRUSTED,
        )
        await testing_client_manager.assert_gql_okay(
            u0wsc,
            graphql_request_payload(
                "mutation addTrust {",
                peerId=safe_radix32.encode_fw(u1_data["account_id"]),
                trust=ACCOUNT_TRUST_CRYPTO_CONTEXT.dump(
                    AccountTrustConfigDumpArgs(u0_kr, u1_data["_acc_kr"], NanoSelfAccountKeyring.ED25519)
                ),
            ),
        )
        await delay_retry_coroutine_assertion(
            testing_client_manager.listing_room, "", r0_id, u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, ls_path=drive_b
        )
        # anonymous denied
        anon0_kr = NanoSelfAccountKeyring.generate_anonymous()
        await testing_client_manager.listing_room(
            "",
            r0_id,
            anon_wsc,
            anon0_kr,
            u0_kr,
            r0_kr,
            server_error="Invalid argument!",
            ephemeral_pk=anon0_kr.get_public_key(anon0_kr.CURVE25519),
        )
        # perm change by admin member
        perm_req1 = {
            "type": "ROOM-CFG",
            "room_id": safe_radix32.encode_fw(r0_id),
            "group_flags": GroupConfigBlockFlags.ALLOW_ANONYMOUS,
        }
        # cross-using mutation id is not possible (thankfully)
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    perm_req1, "dbg0", r0_id, None, u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr
                )
            ),
            r0_id,
            None,
            u1wsc,
            u1_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_MUTATION_ID,
        )
        # member is not admin to change room config
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    perm_req1, "dbg1", r0_id, None, u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr
                )
            ),
            r0_id,
            None,
            u1wsc,
            u1_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_HANDLER_DENIED,
        )
        # admin room-config update by member
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    perm_req1, "dbg3", r0_id, None, u3wsc, u3_data["_acc_kr"], u0_kr, r0_kr
                )
            ),
            r0_id,
            None,
            u3wsc,
            u3_data["_acc_kr"],
            u0_kr,
            r0_kr,
        )
        # anonymous allowed (with room-secret)
        await delay_retry_coroutine_assertion(
            testing_client_manager.listing_room,
            "",
            r0_id,
            anon_wsc,
            anon0_kr,
            u0_kr,
            r0_kr,
            ephemeral_pk=anon0_kr.get_public_key(anon0_kr.CURVE25519),
            ls_path=drive_b,
        )
        # not member also allowed with room secret as anonymous
        await testing_client_manager.listing_room("", r0_id, u2wsc, u2_data["_acc_kr"], u0_kr, r0_kr, ls_path=drive_b)
        # set local override of anonymous access by owner
        await testing_client_manager.nano_request_response(
            msgpack_dumps(
                await testing_client_manager.prepare_request_with_mutation_id(
                    {"type": "A-EDIT-DENY-ANON", "value": True, "admin_password": a_pwd},
                    "dbg0",
                    None,
                    nano_id,
                    u0wsc,
                    u0_kr,
                    u0_kr,
                )
            ),
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
        )
        # anonymous access is denied by nano
        await testing_client_manager.listing_room(
            "",
            r0_id,
            anon_wsc,
            anon0_kr,
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_DENIED,
            ephemeral_pk=anon0_kr.get_public_key(anon0_kr.CURVE25519),
        )
        await testing_client_manager.listing_room(
            "",
            r0_id,
            u2wsc,
            u2_data["_acc_kr"],
            u0_kr,
            r0_kr,
            error=NANO_REQUEST_ERROR_DENIED,
        )

        # revert allow anonymous for other tests (cleanup)
        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(r0_id),
                        "group_flags": GroupConfigBlockFlags.ALLOW_ANONYMOUS_NEGATE,
                    },
                    "dbg3",
                    r0_id,
                    None,
                    u3wsc,
                    u3_data["_acc_kr"],
                    u0_kr,
                    r0_kr,
                )
            ),
            r0_id,
            None,
            u3wsc,
            u3_data["_acc_kr"],
            u0_kr,
            r0_kr,
        )

    # test throttling authorized account on keyring load
    async with start_nano_wait_gateway(
        testing_server_manager,
        reset_user_app_dirs,
        dev_nano_root_dir,
        "active",
    ) as (pm, debug_info, gateway_port, gateway_key):
        # test throttled error received by the nano (query keyring data)
        await testing_server_manager.call_throttling_account(u0_id_sr32, names=["query"])
        await testing_client_manager.listing_room(
            "", r0_id, u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, error=NANO_REQUEST_ERROR_TOO_MANY_REQUESTS
        )
        await testing_server_manager.call_throttling_account(u0_id_sr32, reset=True, names=["query"])

        # check serving still works (after restart and throttling)
        await testing_client_manager.listing_room("", r0_id, u1wsc, u1_data["_acc_kr"], u0_kr, r0_kr, ls_path=drive_b)

        # detach room from drive
        await testing_client_manager.single_response_admin_request(
            {
                "type": "A-D-DETACH",
                "path": str(drive_b),
                "target_room_id": safe_radix32.encode_fw(r0_id),
            },
            "dbg01x",
            a_pwd,
            None,
            nano_id,
            u0wsc,
            u0_kr,
            u0_kr,
        )
        # room is no longer served
        try:
            await testing_client_manager.listing_room("", r0_id, u0wsc, u0_kr, u0_kr, error=NANO_REQUEST_ERROR_DENIED)
        except AssertionError as e:
            if str(e) != "{'message': 'No Nano available for target!'}":
                raise e
            # the server cleaned up the session for the room, the nano can't even be asked
            # TODO: we'd need to manipulate the routing on the server to allow the nano to be messaged
            print("WARNING: Nano detached room serve-deny is not asserted")
        else:
            # wait for routing information to clear on the server
            i = 0
            while True:
                try:
                    await testing_client_manager.listing_room(
                        "", r0_id, u0wsc, u0_kr, u0_kr, server_error="No Nano available for target!"
                    )
                    break
                except AssertionError as e:
                    if str(e) != "{'error': 4206}":  # room must be denied even if routing still works
                        raise e
                    if i > 25:
                        raise e
                i += 1
                await asyncio.sleep(0.1)

        # check updated config
        await _check_remote_config_query(a_pwd)

        # logout
        await do_gateway_logout(testing_server_manager, pm, debug_info, gateway_port, gateway_key)
        # login with other user (password normalization test)
        await do_gateway_login(
            testing_server_manager,
            pm,
            debug_info,
            gateway_port,
            gateway_key,
            {
                "email": testing_server_user_norm_pass["email"],
                "password": testing_server_user_norm_pass["norm_alt_passes"][0],
            },
            wait_active_session=False,
        )

        # logout
        await do_gateway_logout(testing_server_manager, pm, debug_info, gateway_port, gateway_key)
        # login with other user (legacy password normalization test)
        await do_gateway_login(
            testing_server_manager,
            pm,
            debug_info,
            gateway_port,
            gateway_key,
            {
                "email": testing_server_user_not_norm_pass["email"],
                "password": testing_server_user_not_norm_pass["not_norm_alt_passes"][0],
            },
            wait_active_session=False,
        )
