Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 29 additions & 17 deletions src/zero_server/server/process_mutations.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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],
Expand All @@ -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)

Expand All @@ -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))
Expand Down
60 changes: 59 additions & 1 deletion tests/unit/test_mutate.py
Original file line number Diff line number Diff line change
@@ -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"}
Expand Down Expand Up @@ -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": []}
Loading