quart-session/tests/test_redis_json.py

"""Native JSON command contracts; no external Redis server required."""

from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock

import pytest
from quart import Quart, session

pytest.importorskip("redis")
from redis.exceptions import ResponseError

from quart_session import Session
from quart_session.sessions import RedisSessionInterface, RedisTrioSessionInterface

def make_interface(backend, enabled=True):
  return RedisSessionInterface(
    redis=backend, key_prefix="session:", SESSION_REDIS_JSON=enabled,
    SESSION_REVERSE_PROXY=False, SESSION_PROTECTION=False,
    SESSION_STATIC_FILE=False,
  )

def pipeline_backend():
  backend = MagicMock()
  pipe = MagicMock()
  backend.pipeline.return_value.__aenter__ = AsyncMock(return_value=pipe)
  backend.pipeline.return_value.__aexit__ = AsyncMock(return_value=False)
  pipe.execute = AsyncMock(return_value=[1, "OK", True])
  return backend, pipe

@pytest.mark.asyncio
@pytest.mark.parametrize("value", ['{"hits":1}', b'{"hits":1}', None])
async def test_read_legacy_string_and_missing_key(value):
  backend = AsyncMock()
  backend.get.return_value = value
  interface = make_interface(backend)
  assert await interface.get("session:test") == value
  backend.execute_command.assert_not_called()

@pytest.mark.asyncio
async def test_read_native_document():
  backend = AsyncMock()
  backend.get.side_effect = ResponseError("WRONGTYPE wrong kind of value")
  backend.execute_command.return_value = '{"hits":2}'
  assert await make_interface(backend).get("session:test") == '{"hits":2}'
  backend.execute_command.assert_awaited_once_with("JSON.GET", "session:test", ".")

@pytest.mark.asyncio
async def test_read_does_not_hide_server_errors():
  backend = AsyncMock()
  backend.get.side_effect = ResponseError("NOPERM permission denied")
  with pytest.raises(ResponseError, match="NOPERM"):
    await make_interface(backend).get("session:test")
  backend.execute_command.assert_not_called()

@pytest.mark.asyncio
async def test_write_json_and_expiry_in_transaction():
  backend, pipe = pipeline_backend()
  value = '{"_permanent":true,"user":{"name":"Example User"}}'
  assert await make_interface(backend).set("session:test", value, expiry=60) == "OK"
  backend.pipeline.assert_called_once_with(transaction=True)
  assert [(call[0], call.args) for call in pipe.mock_calls] == [
    ("delete", ("session:test",)),
    ("execute_command", ("JSON.SET", "session:test", "$", value)),
    ("expire", ("session:test", 60)),
    ("execute", ()),
  ]

@pytest.mark.asyncio
async def test_invalid_json_leaves_existing_key_untouched():
  backend, _ = pipeline_backend()
  for value in ("not json", '"string"', "[]"):
    with pytest.raises(ValueError):
      await make_interface(backend).set("session:test", value, expiry=60)
  backend.pipeline.assert_not_called()

@pytest.mark.asyncio
async def test_default_still_uses_string_storage():
  backend = AsyncMock()
  interface = make_interface(backend, enabled=False)
  await interface.set("session:test", "value", expiry=60)
  backend.setex.assert_awaited_once_with(name="session:test", value="value", time=60)

@pytest.mark.asyncio
@pytest.mark.parametrize("commands", [
  [[b"json.get"], [b"json.set"]],
  {"json.get": {"name": "json.get"}, "json.set": {"name": "json.set"}},
])
async def test_startup_checks_server_support(commands):
  backend = AsyncMock()
  backend.execute_command.return_value = commands
  await make_interface(backend).create(Quart(__name__))
  backend.execute_command.assert_awaited_once_with(
    "COMMAND", "INFO", "JSON.GET", "JSON.SET"
  )

@pytest.mark.asyncio
async def test_startup_rejects_missing_json_support():
  backend = AsyncMock()
  backend.execute_command.return_value = [None, None]
  with pytest.raises(RuntimeError, match="JSON.GET and JSON.SET"):
    await make_interface(backend).create(Quart(__name__))

def test_trio_rejects_json_mode():
  with pytest.raises(ValueError, match="asyncio"):
    RedisTrioSessionInterface(
      redis=MagicMock(), key_prefix="session:", SESSION_REDIS_JSON=True,
    )

@pytest.mark.asyncio
async def test_quart_session_roundtrip_and_legacy_migration():
  backend, pipe = pipeline_backend()
  backend.get = AsyncMock(return_value=None)
  backend.execute_command = AsyncMock()
  backend.delete = AsyncMock()
  app = Quart(__name__)
  app.config.update(SESSION_TYPE="redis", SESSION_REDIS=backend, SESSION_REDIS_JSON=True)
  Session(app)
  when = datetime(2026, 1, 1, tzinfo=timezone.utc)

  @app.route("/")
  async def index():
    if "when" in session:
      assert session["when"] == when
      assert session["pair"] == (1, 2)
    session["when"] = when
    session["pair"] = (1, 2)
    session["hits"] = session.get("hits", 0) + 1
    return {"hits": session["hits"], "sid": session.sid}

  client = app.test_client()
  first = await client.get("/")
  assert first.status_code == 200
  key = "session:" + (await first.get_json())["sid"]
  document = pipe.execute_command.call_args.args[3]
  pipe.expire.assert_called_with(key, int(app.permanent_session_lifetime.total_seconds()))

  # A pre-existing string remains readable and is rewritten using JSON.SET.
  backend.get.return_value = document
  second = await client.get("/")
  assert (await second.get_json())["hits"] == 2
  assert pipe.execute_command.call_args.args[:3] == ("JSON.SET", key, "$")

  # A native document follows the same serializer path on the next request.
  backend.get.side_effect = ResponseError("WRONGTYPE wrong kind of value")
  backend.execute_command.return_value = pipe.execute_command.call_args.args[3]
  third = await client.get("/")
  assert (await third.get_json())["hits"] == 3
  assert (await third.get_json())["sid"] == (await first.get_json())["sid"]
  await app.session_interface.delete(key)
  backend.delete.assert_awaited_once_with(key)