diff --git a/src/zero_server/server/process_mutations.py b/src/zero_server/server/process_mutations.py index f7c071f..c6ea7f0 100644 --- a/src/zero_server/server/process_mutations.py +++ b/src/zero_server/server/process_mutations.py @@ -15,7 +15,7 @@ import logging from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass -from typing import Any, Literal, Protocol +from typing import Literal, Protocol from pydantic import TypeAdapter, ValidationError @@ -155,6 +155,14 @@ def _parse_failure( ) +def _parse_push_body(json_body: object) -> PushBody | ServerPushFailed: + try: + return PushBody.from_json(json_body) + except ValidationError as error: + logger.error("Failed to parse push body", exc_info=error) + return _parse_failure("push body", error, []) + + async def handle_mutate_request[T]( *, db_provider: Database[T], @@ -165,9 +173,13 @@ async def handle_mutate_request[T]( ) -> MutateResponse: """Process one `/mutate` request from zero-cache. - `query` is the request's query string and `body` its JSON, either raw bytes or already parsed. + `query` is the request's query string and `body` its JSON: raw bytes, decoded JSON, or a `PushBody`. `user_id` is the authenticated user, `None` when logged out; reporting it back is what makes zero-cache treat the connection's user as server-validated. + + A caller that needs push-level fields such as `timestamp` parses once with `PushBody.from_json` and + passes the result. If that parse fails, pass the raw body instead and return the parse failure this + produces. """ return await handle_mutate(db_provider, handler, query, body, user_id, report_user_id=True) @@ -182,21 +194,21 @@ async def handle_mutate[T]( report_user_id: bool, ) -> MutateResponse: """`handle_mutate_request`, and the legacy form that reports no user ID at all.""" - json_body: Any = body - if isinstance(body, bytes): - try: - json_body = json.loads(body, parse_constant=_refuse_constant) - except ValueError as error: - logger.error("Failed to parse push body", exc_info=error) - return _parse_failure("push body", error, [], syntax=True) - - mutation_ids: list[MutationID] = [] - try: - push_body = PushBody.from_json(json_body) - mutation_ids = [MutationID(id=m.id, clientID=m.client_id) for m in push_body.mutations] - except ValidationError as error: - logger.error("Failed to parse push body", exc_info=error) - return _parse_failure("push body", error, mutation_ids) + match body: + case PushBody(): + push_body = body + case bytes(): + try: + json_body = json.loads(body, parse_constant=_refuse_constant) + except ValueError as error: + logger.error("Failed to parse push body", exc_info=error) + return _parse_failure("push body", error, [], syntax=True) + push_body = _parse_push_body(json_body) + case _: + push_body = _parse_push_body(body) + if isinstance(push_body, ServerPushFailed): + return push_body + mutation_ids = [MutationID(id=m.id, clientID=m.client_id) for m in push_body.mutations] try: params = MutateParams.from_json(dict(query)) diff --git a/tests/unit/test_mutate.py b/tests/unit/test_mutate.py index 55299ef..7bf3c66 100644 --- a/tests/unit/test_mutate.py +++ b/tests/unit/test_mutate.py @@ -1,10 +1,13 @@ -"""Branches of the mutate endpoint the wire tier cannot reach: raw bodies upstream would read off a Request.""" +"""Branches of the mutate endpoint the wire tier cannot reach: raw bodies upstream would read off a Request, +and bodies the caller has already parsed.""" +import json from typing import Any import pytest from zero_server.protocol.mutation import CustomMutation, MutationResponse +from zero_server.protocol.push import PushBody from zero_server.server.process_mutations import TransactFn, handle_mutate_request QUERY = {"schema": "zero_0", "appID": "zero"} @@ -41,3 +44,58 @@ async def test_raw_body_is_parsed() -> None: db_provider=NoDatabase(), handler=handler, query=QUERY, body=body, user_id="u" ) assert response.to_json() == {"kind": "MutateResponse", "userID": "u", "mutations": []} + + +PUSH = {"clientGroupID": "g", "mutations": [], "pushVersion": 1, "timestamp": 1, "requestID": "r"} +MUTATION = {"type": "custom", "id": 1, "clientID": "c", "name": "issue.create", "args": [], "timestamp": 1} + + +class AcceptingDatabase: + """Accepts each mutation's ID as the next one, and records nothing.""" + + async def transaction(self, callback: Any, transaction_input: Any = None) -> Any: + class Hooks: + async def update_client_mutation_id(self) -> int: + return transaction_input.mutation_id + + async def write_mutation_result(self, result: MutationResponse) -> None: + pass + + return await callback(None, Hooks()) + + +async def failing_handler(transact: TransactFn[Any], mutation: CustomMutation) -> MutationResponse: + raise ValueError("no such issue") + + +@pytest.mark.parametrize( + ("push", "query"), + [ + (PUSH, QUERY), + ({**PUSH, "mutations": [MUTATION]}, QUERY), + ({**PUSH, "mutations": [MUTATION], "pushVersion": 2}, QUERY), + ({**PUSH, "mutations": [MUTATION]}, {"appID": "zero"}), + ], + ids=["success", "mutation-error", "unsupported-push-version", "bad-query"], +) +async def test_parsed_body_matches_raw(push: dict[str, Any], query: dict[str, str]) -> None: + async def respond(body: object) -> object: + response = await handle_mutate_request( + db_provider=AcceptingDatabase(), handler=failing_handler, query=query, body=body, user_id="u" + ) + return response.to_json() + + assert await respond(PushBody.from_json(push)) == await respond(json.dumps(push).encode()) + + +async def test_parsed_body_is_not_validated_again(monkeypatch: pytest.MonkeyPatch) -> None: + push_body = PushBody.from_json(PUSH) + + def refuse(value: object) -> PushBody: + raise AssertionError("parsed body validated again") + + monkeypatch.setattr(PushBody, "from_json", refuse) + response = await handle_mutate_request( + db_provider=NoDatabase(), handler=handler, query=QUERY, body=push_body, user_id="u" + ) + assert response.to_json() == {"kind": "MutateResponse", "userID": "u", "mutations": []}