mirror of
https://github.com/zulip/zulip.git
synced 2025-11-18 12:54:58 +00:00
test_helpers: Convert TypedDict from queries_captured to dataclass.
An implicit coercion from an untyped dict to the TypedDict was hiding
a type error: CapturedQuery.sql was really str, not bytes. We should
always prefer dataclass over TypedDict to prevent such errors.
Signed-off-by: Anders Kaseorg <anders@zulip.com>
(cherry picked from commit 92db6eba78)
This commit is contained in:
committed by
Alex Vandiver
parent
9628cc9278
commit
201cab601a
@@ -4,6 +4,7 @@ import re
|
||||
import sys
|
||||
import time
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from typing import (
|
||||
IO,
|
||||
TYPE_CHECKING,
|
||||
@@ -16,7 +17,6 @@ from typing import (
|
||||
Mapping,
|
||||
Optional,
|
||||
Tuple,
|
||||
TypedDict,
|
||||
TypeVar,
|
||||
Union,
|
||||
cast,
|
||||
@@ -133,21 +133,22 @@ def simulated_empty_cache() -> Iterator[List[Tuple[str, Union[str, List[str]], O
|
||||
yield cache_queries
|
||||
|
||||
|
||||
class CapturedQueryDict(TypedDict):
|
||||
sql: bytes
|
||||
@dataclass
|
||||
class CapturedQuery:
|
||||
sql: str
|
||||
time: str
|
||||
|
||||
|
||||
@contextmanager
|
||||
def queries_captured(
|
||||
include_savepoints: bool = False, keep_cache_warm: bool = False
|
||||
) -> Iterator[List[CapturedQueryDict]]:
|
||||
) -> Iterator[List[CapturedQuery]]:
|
||||
"""
|
||||
Allow a user to capture just the queries executed during
|
||||
the with statement.
|
||||
"""
|
||||
|
||||
queries: List[CapturedQueryDict] = []
|
||||
queries: List[CapturedQuery] = []
|
||||
|
||||
def wrapper_execute(
|
||||
self: TimeTrackingCursor,
|
||||
@@ -163,10 +164,10 @@ def queries_captured(
|
||||
duration = stop - start
|
||||
if include_savepoints or not isinstance(sql, str) or "SAVEPOINT" not in sql:
|
||||
queries.append(
|
||||
{
|
||||
"sql": self.mogrify(sql, params).decode(),
|
||||
"time": f"{duration:.3f}",
|
||||
}
|
||||
CapturedQuery(
|
||||
sql=self.mogrify(sql, params).decode(),
|
||||
time=f"{duration:.3f}",
|
||||
)
|
||||
)
|
||||
|
||||
def cursor_execute(
|
||||
|
||||
Reference in New Issue
Block a user