import asyncio
import copy
import hashlib
import time
from pathlib import Path
from typing import Optional, Tuple, Any, List

import aiohttp
import safe_radix32
from clarabot_crypto.context.account_signature import ACCOUNT_SIGNATURE_CRYPTO_CONTEXT, AccountSignatureConfigLoadArgs
from clarabot_crypto.context.nano import NANO_CRYPTO_CONTEXT, NanoConfigDumpArgs, NanoConfigLoadArgs
from clarabot_crypto.context.nano_data import NANO_DATA_CRYPTO_CONTEXT, NanoDataConfigLoadArgs
from clarabot_crypto.context.nano_own import NANO_OWN_CRYPTO_CONTEXT, NanoOwnConfigDumpArgs, NanoOwnConfigLoadArgs
from clarabot_crypto.keyring.account_base import AbstractSelfAccountKeyring, AbstractPeerAccountKeyring
from clarabot_crypto.keyring.private import PRIVATE_KEYRING_VERSION_MAP
from clarabot_crypto.keyring.room_base import AbstractRoomKeyring
from clarabot_testing.client.gql_ws import BaseGraphQlTransport
from clarabot_testing.client.tester import AbstractClientTestingMixin

from clarabot_nano.common.protocol.errors import (
    NANO_REQUEST_ERROR_MUTATION_ID,
    NANO_REQUEST_ERROR_ADMIN_PASSWORD,
    NANO_REQUEST_ERROR_PAYLOAD_ROOM_INVALID,
)
from clarabot_nano.common.protocol.request import NanoBaseRequest
from clarabot_nano.utility.path.root import get_root_paths
from clarabot_nano.utility.serialize import msgpack_dumps, msgpack_loads
from clarabot_nano_tests.testing.client.gql_ws import GraphQlWebSocketClient
from clarabot_nano_tests.testing.ls import merge_listing_responses
from clarabot_nano_tests.testing.server.manager import DevServerManager


class GraphQLClientManager(AbstractClientTestingMixin):
    def __init__(self, server_manager: DevServerManager):
        AbstractClientTestingMixin.__init__(self)
        self.server_manager = server_manager
        self._mutation_ids = dict()

    # client overrides
    async def wsc(self, session: aiohttp.ClientSession) -> GraphQlWebSocketClient:
        return await self.server_manager.websocket(session)

    def _get_server_hostname(self):
        return self.server_manager.server_frontend_domain

    def url_for(self, path):
        return self.server_manager.url_for(path)

    def _server_post_auto_csrf(self, s: aiohttp.ClientSession, url: str, *args, **kwargs):
        return self.server_manager.post_auto_csrf(s, url, *args, **kwargs)

    def _check_server_version(self, sv):
        return sv.startswith("Clarabot ")  # NOTE: server version is not known exactly

    def _check_query_me(self, me, target_me):
        return me["data"]["me"]["id"] == target_me

    def _check_auth_token_extract_id(self, token):
        pass  # NOTE: cannot check the token signature, only the server knows the HMAC secret

    def get_gql_url(self) -> str:
        return self.server_manager.get_gql_url()

    def get_wsgql_url(self) -> str:
        return self.server_manager.get_wsgql_url()

    async def session(self, *args) -> aiohttp.ClientSession:
        return await self.server_manager.session(*args)

    def session_storage(self, s: aiohttp.ClientSession) -> dict:
        return self.server_manager.session_storage(s)

    async def _pack_safe_radix32(self, value):
        return await self.server_manager.safe_radix32(value)

    async def ensure_logged_in_session(self, account_data: dict) -> aiohttp.ClientSession:
        s = await self.session(account_data["account_id"])
        if not await self.check_logged_in(s, account_data):
            await self.assert_login(
                s,
                account_data["email"],
                account_data["password"],
                private_kr_cls=PRIVATE_KEYRING_VERSION_MAP[account_data["_p_kr_version"]],
            )
        return s

    async def check_logged_in(self, s_ws: BaseGraphQlTransport, account_data: dict):
        return self._check_query_me(
            await self.gql_query_me(s_ws), await self._pack_safe_radix32(account_data["account_id"])
        )

    async def create_logged_in_wsc(self, account_data: dict) -> GraphQlWebSocketClient:
        s = await self.ensure_logged_in_session(account_data)
        wsc = await self.wsc(s)
        assert await self.check_logged_in(wsc, account_data)
        await self.assert_gql_query_ping(wsc)
        return wsc

    async def listing_response_admin_request(
        self,
        path: str,
        admin_password: str,
        room_id: Optional[int],
        nano_id: Optional[int],
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        **request_kwargs,
    ):
        """
        The listing handler has an own utility to check overlap and multi-part responses.
        """
        request = {"type": "A-LISTING", "path": path}
        await self.negative_auth_admin_request(
            request, None, room_id, nano_id, req_wsc, req_kr, nano_kr, room_kr, **request_kwargs
        )
        # with mutation id and admin password
        request_payload = copy.deepcopy(request)
        request_payload["admin_password"] = admin_password
        responses = await self.nano_request_responses(
            msgpack_dumps(request_payload),
            room_id,
            nano_id,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=error,
            single_response=False,
            **request_kwargs,
        )
        if error is not None:
            return
        if not path:
            ref_ls = set(get_root_paths())
        else:
            ref_ls = set(p.name for p in Path(path).iterdir() if p.is_dir())
        merged_ls = merge_listing_responses(responses)
        ls = set(item["name"] for item in merged_ls)
        assert ref_ls == ls, f"{path}\n{ref_ls}\n{ls}\n{ref_ls.difference(ls)}\n{ls.difference(ref_ls)}"
        return merged_ls

    async def single_response_admin_request(
        self,
        request: dict,
        mutation_name: str,
        admin_password: str,
        room_id: Optional[int],
        nano_id: Optional[int],
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        error_before_deny: bool = False,
        **request_kwargs,
    ):
        """
        Checks the mutation and password protections.
        """
        # some errors have priority over mutation and admin password, so those can't be checked
        if not error_before_deny and error not in (NANO_REQUEST_ERROR_PAYLOAD_ROOM_INVALID,):
            await self.negative_auth_admin_request(
                request, mutation_name, room_id, nano_id, req_wsc, req_kr, nano_kr, room_kr, **request_kwargs
            )
        # with mutation id and admin password
        request_payload = copy.deepcopy(request)
        request_payload["admin_password"] = admin_password
        r = await self.prepare_request_with_mutation_id(
            request_payload, mutation_name, room_id, nano_id, req_wsc, req_kr, nano_kr, room_kr, **request_kwargs
        )
        return await self.nano_request_responses(
            msgpack_dumps(r), room_id, nano_id, req_wsc, req_kr, nano_kr, room_kr, error=error, **request_kwargs
        )

    async def negative_auth_admin_request(
        self,
        request: dict,
        mutation_name: Optional[str],
        room_id: Optional[int],
        nano_id: Optional[int],
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        **request_kwargs,
    ):
        """
        Checks the mutation and password protections.
        """
        if mutation_name is not None:
            # can't without mutation id
            await self.nano_request_responses(
                msgpack_dumps(request),
                room_id,
                nano_id,
                req_wsc,
                req_kr,
                nano_kr,
                room_kr,
                error=NANO_REQUEST_ERROR_MUTATION_ID,
                **request_kwargs,
            )
        # can't change name without admin password
        r = await self.prepare_request_with_mutation_id(
            copy.deepcopy(request), mutation_name, room_id, nano_id, req_wsc, req_kr, nano_kr, room_kr, **request_kwargs
        )
        await self.nano_request_responses(
            msgpack_dumps(r),
            room_id,
            nano_id,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=NANO_REQUEST_ERROR_ADMIN_PASSWORD,
            **request_kwargs,
        )

    async def prepare_request_with_mutation_id(
        self,
        request: dict,
        mutation_name: str,
        room_id: Optional[int],
        nano_id: Optional[int],
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        refresh: Optional[bool] = None,
        **request_kwargs,
    ) -> dict:
        if mutation_name not in self._mutation_ids or refresh:
            resps = await self.nano_request_responses(
                # NOTE: room id consistency is required
                msgpack_dumps({"type": "MUT-CRT", "room_id": safe_radix32.encode_fw(room_id) if room_id else None}),
                room_id,
                nano_id,
                req_wsc,
                req_kr,
                nano_kr,
                room_kr,
                error,
                **request_kwargs,
            )
            self._mutation_ids[mutation_name] = (resps[0]["token"], 1)
        mutation_key, mutation_seq = self._mutation_ids[mutation_name]
        self._mutation_ids[mutation_name] = (mutation_key, mutation_seq + 1)
        new_request = {
            NanoBaseRequest.MUTATION_KEY: mutation_key,
            NanoBaseRequest.MUTATION_SEQUENCE: mutation_seq,
        }
        new_request.update(request)
        return new_request

    async def listing_room(
        self,
        path: str,
        room_id: int,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        ls_path: Optional[Path] = None,
        ls_allowed: Optional[set] = None,
        **request_kwargs,
    ) -> Optional[Tuple[List[Any], List[Any]]]:
        request = {"type": "FS-GET", "path": path, "room_id": safe_radix32.encode_fw(room_id)}
        responses = await self.nano_request_responses(
            msgpack_dumps(request),
            room_id,
            None,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=error,
            server_error=server_error,
            single_response=False,
            **request_kwargs,
        )
        if error is not None or server_error is not None:
            return
        ls = merge_listing_responses(responses)
        assert ls_path
        dirs = []
        files = []
        for item in ls_path.iterdir():
            if ls_allowed is not None and item.name not in ls_allowed:
                continue
            s = item.stat()
            if item.is_dir():
                dirs.append(item.name)
                entry = {"name": item.name, "mtime": s.st_mtime}
            else:
                files.append(item.name)
                entry = {"name": item.name, "mtime": s.st_mtime, "size": s.st_size}
                if files[-1].endswith(".png"):  # this works, but it is very naive
                    entry["imgb"] = 1
            try:
                ls.remove(entry)
            except ValueError:  # pragma: no cover
                assert False, f"{entry}\n{ls}\n{responses}\n{dirs}\n{files}"
        assert not ls, f"{ls}"
        return dirs, files

    async def download_room(
        self,
        path: str,
        room_id: int,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        file_path: Optional[Path] = None,
        part_count: Optional[int] = None,
        **request_kwargs,
    ) -> Optional[Any]:
        request = {"type": "FS-GET", "path": path, "room_id": safe_radix32.encode_fw(room_id)}
        responses = await self.nano_request_responses(
            msgpack_dumps(request),
            room_id,
            None,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=error,
            server_error=server_error,
            single_response=False,
            **request_kwargs,
        )
        if error is not None or server_error is not None:
            return
        assert file_path
        data = b""
        verified = 0
        await asyncio.sleep(0.001)
        if len(responses) == 1:
            assert responses[0].get("fs") == 0, str(responses)
        else:
            t = time.perf_counter() + 4.0
            assert len(responses) > 1, str(responses)
            for resp in responses[1:]:
                if time.perf_counter() > t:
                    await asyncio.sleep(0.001)
                    t = time.perf_counter() + 4.0
                data += resp["fc"]
                if mac := resp.get("MAC"):
                    assert (ds := resp.get("MACS")) is not None
                    mac_data = msgpack_loads(mac)
                    assert len(data) - mac_data["c"] == verified
                    h = hashlib.sha512(mac_data["s"])
                    h.update(data[verified:])
                    assert mac_data["f"] == h.digest()[:32]
                    ACCOUNT_SIGNATURE_CRYPTO_CONTEXT.load(AccountSignatureConfigLoadArgs(ds, nano_kr, mac))
                    verified += mac_data["c"]
        assert len(data) == verified
        assert len(data) == responses[0]["fs"]
        assert file_path.read_bytes() == data
        if part_count is not None:
            assert len(responses) - 1 == part_count, f"{len(responses)} {part_count}"
        return data

    async def thumbnails_room(
        self,
        path: str,
        room_id: int,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        part_count: Optional[int] = None,
        **request_kwargs,
    ) -> Optional[Any]:
        request = {"type": "FS-THMB", "path": path, "room_id": safe_radix32.encode_fw(room_id)}
        responses = await self.nano_request_responses(
            msgpack_dumps(request),
            room_id,
            None,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=error,
            server_error=server_error,
            single_response=False,
            **request_kwargs,
        )
        if error is not None or server_error is not None:
            return
        data = b""
        parts = []
        verified = 0
        await asyncio.sleep(0.001)
        if len(responses) == 1:
            assert responses[0].get("ts") is None, str(responses)
        else:
            t = time.perf_counter() + 4.0
            assert len(responses) > 1, str(responses)
            for resp in responses[1:]:
                if time.perf_counter() > t:
                    await asyncio.sleep(0.001)
                    t = time.perf_counter() + 4.0
                parts.append(resp["ts"])
                data += resp["ts"]
                if mac := resp.get("MAC"):
                    assert (ds := resp.get("MACS")) is not None
                    mac_data = msgpack_loads(mac)
                    assert len(data) - mac_data["c"] == verified
                    h = hashlib.sha512(mac_data["s"])
                    h.update(data[verified:])
                    assert mac_data["f"] == h.digest()[:32]
                    ACCOUNT_SIGNATURE_CRYPTO_CONTEXT.load(AccountSignatureConfigLoadArgs(ds, nano_kr, mac))
                    verified += mac_data["c"]
        assert len(data) == verified
        if part_count is not None:
            assert len(responses) - 1 == part_count, f"{len(responses)} {part_count}"
        ts = []
        for part in parts:
            ts.extend(msgpack_loads(part))
        return ts

    async def peek_room(
        self,
        path: str,
        room_id: int,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        **request_kwargs,
    ):
        request = {"type": "FS-PEEK", "path": path, "room_id": safe_radix32.encode_fw(room_id)}
        response = await self.nano_request_response(
            msgpack_dumps(request),
            room_id,
            None,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=error,
            server_error=server_error,
            **request_kwargs,
        )
        if error is not None or server_error is not None:
            return
        # return relevant field only
        return response["pk"]

    async def search_room(
        self,
        query: str,
        room_id: int,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        lang: Optional[str] = None,
        **request_kwargs,
    ):
        request = {"type": "SEARCH", "query": query, "lang": lang, "room_id": safe_radix32.encode_fw(room_id)}
        response = await self.nano_request_response(
            msgpack_dumps(request),
            room_id,
            None,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=error,
            server_error=server_error,
            **request_kwargs,
        )
        if error is not None or server_error is not None:
            return
        # return relevant fields only
        return {"docs": response["docs"]}

    async def mkdir_room(
        self,
        path: str,
        room_id: int,
        mutation_name: str,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        **request_kwargs,
    ):
        request = {"type": "FS-MKD", "path": path, "room_id": safe_radix32.encode_fw(room_id)}
        return await self.mutate_room(
            request, room_id, mutation_name, req_wsc, req_kr, nano_kr, room_kr, error, server_error, **request_kwargs
        )

    async def del_room(
        self,
        path: str,
        room_id: int,
        mutation_name: str,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        **request_kwargs,
    ):
        request = {"type": "FS-DEL", "path": path, "room_id": safe_radix32.encode_fw(room_id)}
        return await self.mutate_room(
            request, room_id, mutation_name, req_wsc, req_kr, nano_kr, room_kr, error, server_error, **request_kwargs
        )

    async def move_room(
        self,
        src_path: str,
        new_path: str,
        room_id: int,
        mutation_name: str,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        **request_kwargs,
    ):
        request = {
            "type": "FS-MOV",
            "src_path": src_path,
            "new_path": new_path,
            "room_id": safe_radix32.encode_fw(room_id),
        }
        return await self.mutate_room(
            request, room_id, mutation_name, req_wsc, req_kr, nano_kr, room_kr, error, server_error, **request_kwargs
        )

    async def put_room(
        self,
        req: dict,
        room_id: int,
        mutation_name: str,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        **request_kwargs,
    ):
        request = {"type": "FS-PUT", "room_id": safe_radix32.encode_fw(room_id)}
        request.update(req)
        return await self.mutate_room(
            request, room_id, mutation_name, req_wsc, req_kr, nano_kr, room_kr, error, server_error, **request_kwargs
        )

    async def mutate_room(
        self,
        request: dict,
        room_id: int,
        mutation_name: str,
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,
        server_error: Optional[str] = None,
        **request_kwargs,
    ):
        request = await self.prepare_request_with_mutation_id(
            request, mutation_name, room_id, None, req_wsc, req_kr, nano_kr, room_kr, **request_kwargs
        )
        response = await self.nano_request_response(
            msgpack_dumps(request),
            room_id,
            None,
            req_wsc,
            req_kr,
            nano_kr,
            room_kr,
            error=error,
            server_error=server_error,
            **request_kwargs,
        )
        if error is not None or server_error is not None:
            return
        return response

    def calc_smac(self, data_list):
        smac = hashlib.sha256()
        for data in data_list:
            smac.update(hashlib.sha256(data).digest()[:16])
        return smac.digest()[:16]

    async def nano_request_response(self, *args, **kwargs) -> Any:
        responses = await self.nano_request_responses(*args, **kwargs)
        assert len(responses) == 1, str(responses)
        return responses[0]

    async def nano_request_responses(self, *args, **kwargs) -> List[Any]:
        req, responses = await self.nano_request_responses_raw(*args, **kwargs)
        return [r.get("response", r) for r in responses]

    async def nano_request_responses_raw(
        self,
        request_payload: bytes,
        room_id: Optional[int],
        nano_id: Optional[int],
        req_wsc: GraphQlWebSocketClient,
        req_kr: AbstractSelfAccountKeyring,
        nano_kr: AbstractPeerAccountKeyring,
        room_kr: Optional[AbstractRoomKeyring] = None,
        error: Optional[int] = None,  # error code from Nano
        server_error: Optional[str] = None,  # error from client server
        single_response: bool = True,
        request_payload_encrypted: bool = False,
        continues_from_part: bool = True,
        **request_kwargs,
    ) -> Tuple[bytes, List[dict]]:
        assert server_error is None or error is None
        # encrypt request
        if request_payload_encrypted:
            req = request_payload
        elif req_kr.id == nano_kr.id:
            assert room_kr is None
            req = NANO_OWN_CRYPTO_CONTEXT.dump(NanoOwnConfigDumpArgs(request_payload, req_kr, time.time()))
        else:
            req = NANO_CRYPTO_CONTEXT.dump(NanoConfigDumpArgs(request_payload, req_kr, nano_kr, room_kr, time.time()))
        # send request and gather response(s)
        if nano_id is not None:
            assert room_id is None
            responses = await self.nano_admin_request(req_wsc, nano_id, req, **request_kwargs)
        elif room_id is not None:
            responses = await self.nano_request(req_wsc, room_id, req, **request_kwargs)
        else:
            assert False, f"{room_id} {nano_id}"
        count = len(responses)
        assert count > 0, f"{responses}"
        if single_response:
            assert count == 1, f"{len(responses)} {responses}"
        # check if errors did happen as desired
        if server_error is not None:
            resp = responses[-1]
            assert "message" in resp, f"{resp}"
            assert resp.get("message") == server_error, f"{server_error} {resp}"
            count -= 1
        if error is not None:
            resp = responses[-1]
            assert "error" in resp, f"{resp}"
            assert resp.get("error") == error, f"{error} {resp}"
            count -= 1
        part_delta = request_kwargs.get("part", 1) - 1 if continues_from_part else 0
        # decrypt responses
        t = time.perf_counter() + 4.0
        for i in range(count):
            if time.perf_counter() > t:
                await asyncio.sleep(0.001)
                t = time.perf_counter() + 4.0
            resp = responses[i]
            assert "error" not in resp, f"{resp}"
            expected_part = i + part_delta if i else 0
            assert resp.get("part") == expected_part, f"{expected_part}\n{resp}"
            if i > 0:
                part_key = responses[0]["response"].get("PK")
                assert isinstance(part_key, bytes), str(responses[0]["response"])
                resp_payload = msgpack_loads(
                    NANO_DATA_CRYPTO_CONTEXT.load(NanoDataConfigLoadArgs(resp["response"], part_key))
                )
            elif req_kr.id == nano_kr.id:
                assert room_kr is None
                resp_payload = msgpack_loads(
                    NANO_OWN_CRYPTO_CONTEXT.load(NanoOwnConfigLoadArgs(resp["response"], req_kr, time.time()))
                )
            else:
                resp_payload = msgpack_loads(
                    NANO_CRYPTO_CONTEXT.load(
                        NanoConfigLoadArgs(resp["response"], req_kr, nano_kr, room_kr, time.time())
                    )
                )
            # check and remove protocol-fields
            if i:
                assert resp_payload.pop("P") == expected_part
            else:
                assert (resp.get("cacheKey") is None) != (resp.get("transferKey") is None), f"{sorted(resp.keys())}"
                resp_key = resp.get("cacheKey")
                if not resp_key:
                    resp_key = resp.get("transferKey")
                    cacheable = False
                else:
                    cacheable = True
                assert resp_key, str(resp)
                if cacheable:
                    assert resp["count"] >= 2
                else:
                    assert resp["count"] >= 1
                expected_responses = len(responses) + part_delta
                assert (
                    resp["count"] == expected_responses
                ), f"{expected_responses}\n{len(responses)}\n{resp}\n{responses}"
                if resp["count"] == 1:
                    assert resp_payload.get("PK") is None, f"{resp_payload}"
                else:
                    assert isinstance(resp_payload.get("PK"), bytes), f"{resp_payload}"
                assert "RAH" in resp_payload, str(resp_payload)
                assert "C" in resp_payload, str(resp_payload)
                assert resp_payload.pop("RAH") == hashlib.sha256(resp_key + req[:128]).digest()[:16]
                if error is None and server_error is None:
                    assert resp_payload.pop("C") == expected_responses
                else:
                    assert resp_payload.pop("C") >= expected_responses
            resp["response"] = resp_payload
        return req, responses
