mirror of
https://github.com/Chia-Network/chia-blockchain.git
synced 2026-08-24 10:05:29 -05:00
Matt HauffandGitHub
40db4635a8
* Enable PEP604 Ruff rules * Fix harcoded signature in test * Hack CLVMStreamable test with note to fast follow
692 lines
21 KiB
Python
692 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import dataclasses
|
|
import enum
|
|
import functools
|
|
import gc
|
|
import json
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Awaitable, Callable, Collection, Iterator
|
|
from concurrent.futures import Future
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from statistics import mean
|
|
from textwrap import dedent
|
|
from time import thread_time
|
|
from types import TracebackType
|
|
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, TextIO, TypeVar, cast, final
|
|
|
|
import aiohttp
|
|
import pytest
|
|
|
|
# TODO: update after resolution in https://github.com/pytest-dev/pytest/issues/7469
|
|
from _pytest.nodes import Node
|
|
from aiohttp import web
|
|
from chia_rs import Coin
|
|
from chia_rs.sized_bytes import bytes32
|
|
from chia_rs.sized_ints import uint16, uint32, uint64
|
|
from typing_extensions import Self
|
|
|
|
import chia
|
|
import chia._tests
|
|
from chia._tests import ether
|
|
from chia._tests.core.data_layer.util import ChiaRoot
|
|
from chia._tests.util.time_out_assert import DataTypeProtocol, caller_file_and_line
|
|
from chia.full_node.mempool import Mempool
|
|
from chia.protocols.outbound_message import Message
|
|
from chia.protocols.protocol_message_types import ProtocolMessageTypes
|
|
from chia.server.api_protocol import ApiMetadata, ApiProtocol
|
|
from chia.types.condition_opcodes import ConditionOpcode
|
|
from chia.util.hash import std_hash
|
|
from chia.util.network import WebServer
|
|
from chia.wallet.util.compute_hints import HintedCoin
|
|
from chia.wallet.wallet_node import WalletNode
|
|
|
|
|
|
class GcMode(enum.Enum):
|
|
nothing = enum.auto()
|
|
precollect = enum.auto()
|
|
disable = enum.auto()
|
|
enable = enum.auto()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def manage_gc(mode: GcMode) -> Iterator[None]:
|
|
if mode == GcMode.nothing:
|
|
yield
|
|
elif mode == GcMode.precollect:
|
|
gc.collect()
|
|
yield
|
|
elif mode == GcMode.disable:
|
|
was_enabled = gc.isenabled()
|
|
gc.disable()
|
|
try:
|
|
yield
|
|
finally:
|
|
if was_enabled:
|
|
gc.enable()
|
|
elif mode == GcMode.enable:
|
|
was_enabled = gc.isenabled()
|
|
gc.enable()
|
|
try:
|
|
yield
|
|
finally:
|
|
if not was_enabled:
|
|
gc.disable()
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class RuntimeResults:
|
|
start: float
|
|
end: float
|
|
duration: float
|
|
entry_file: str
|
|
entry_line: int
|
|
overhead: float | None
|
|
|
|
def block(self, label: str = "") -> str:
|
|
# The entry line is reported starting at the beginning of the line to trigger
|
|
# PyCharm to highlight as a link to the source.
|
|
|
|
return dedent(
|
|
f"""\
|
|
Measuring runtime: {label}
|
|
{self.entry_line}
|
|
run time: {self.duration}
|
|
overhead: {self.overhead if self.overhead is not None else "not measured"}
|
|
"""
|
|
)
|
|
|
|
|
|
@final
|
|
@dataclasses.dataclass(frozen=True)
|
|
class AssertRuntimeResults:
|
|
start: float
|
|
end: float
|
|
duration: float
|
|
entry_file: str
|
|
entry_line: int
|
|
overhead: float | None
|
|
limit: float
|
|
ratio: float
|
|
|
|
@classmethod
|
|
def from_runtime_results(
|
|
cls, results: RuntimeResults, limit: float, entry_file: str, entry_line: int, overhead: float | None
|
|
) -> AssertRuntimeResults:
|
|
return cls(
|
|
start=results.start,
|
|
end=results.end,
|
|
duration=results.duration,
|
|
limit=limit,
|
|
ratio=results.duration / limit,
|
|
entry_file=entry_file,
|
|
entry_line=entry_line,
|
|
overhead=overhead,
|
|
)
|
|
|
|
def block(self, label: str = "") -> str:
|
|
# The entry line is reported starting at the beginning of the line to trigger
|
|
# PyCharm to highlight as a link to the source.
|
|
|
|
return dedent(
|
|
f"""\
|
|
Asserting maximum duration: {label}
|
|
{self.entry_file}:{self.entry_line}
|
|
run time: {self.duration}
|
|
overhead: {self.overhead if self.overhead is not None else "not measured"}
|
|
allowed: {self.limit}
|
|
percent: {self.percent_str()}
|
|
"""
|
|
)
|
|
|
|
def message(self) -> str:
|
|
return f"{self.duration} seconds not less than {self.limit} seconds ( {self.percent_str()} )"
|
|
|
|
def passed(self) -> bool:
|
|
return self.duration < self.limit
|
|
|
|
def percent(self) -> float:
|
|
return self.ratio * 100
|
|
|
|
def percent_str(self) -> str:
|
|
return f"{self.percent():.0f} %"
|
|
|
|
|
|
def measure_overhead(
|
|
manager_maker: Callable[
|
|
[], contextlib.AbstractContextManager[Future[RuntimeResults] | Future[AssertRuntimeResults]]
|
|
],
|
|
cycles: int = 10,
|
|
) -> float:
|
|
times: list[float] = []
|
|
|
|
for _ in range(cycles):
|
|
with manager_maker() as results:
|
|
pass
|
|
|
|
times.append(results.result(timeout=0).duration)
|
|
|
|
overhead = mean(times)
|
|
|
|
return overhead
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def measure_runtime(
|
|
label: str = "",
|
|
clock: Callable[[], float] = thread_time,
|
|
gc_mode: GcMode = GcMode.disable,
|
|
overhead: float | None = None,
|
|
print_results: bool = True,
|
|
) -> Iterator[Future[RuntimeResults]]:
|
|
entry_file, entry_line = caller_file_and_line(
|
|
relative_to=(
|
|
pathlib.Path(chia.__file__).parent.parent,
|
|
pathlib.Path(chia._tests.__file__).parent.parent,
|
|
)
|
|
)
|
|
|
|
results_future: Future[RuntimeResults] = Future()
|
|
|
|
with manage_gc(mode=gc_mode):
|
|
start = clock()
|
|
|
|
try:
|
|
yield results_future
|
|
finally:
|
|
end = clock()
|
|
|
|
duration = end - start
|
|
if overhead is not None:
|
|
duration -= overhead
|
|
|
|
results = RuntimeResults(
|
|
start=start,
|
|
end=end,
|
|
duration=duration,
|
|
entry_file=entry_file,
|
|
entry_line=entry_line,
|
|
overhead=overhead,
|
|
)
|
|
results_future.set_result(results)
|
|
|
|
if print_results:
|
|
print(results.block(label=label))
|
|
|
|
|
|
@final
|
|
@dataclasses.dataclass(frozen=True)
|
|
class BenchmarkData:
|
|
if TYPE_CHECKING:
|
|
_protocol_check: ClassVar[DataTypeProtocol] = cast("BenchmarkData", None)
|
|
|
|
tag: ClassVar[str] = "benchmark"
|
|
|
|
duration: float
|
|
path: pathlib.Path
|
|
line: int
|
|
limit: float
|
|
|
|
label: str
|
|
|
|
__match_args__: ClassVar[tuple[str, ...]] = ()
|
|
|
|
@classmethod
|
|
def unmarshal(cls, marshalled: dict[str, Any]) -> BenchmarkData:
|
|
return cls(
|
|
duration=marshalled["duration"],
|
|
path=pathlib.Path(marshalled["path"]),
|
|
line=int(marshalled["line"]),
|
|
limit=marshalled["limit"],
|
|
label=marshalled["label"],
|
|
)
|
|
|
|
def marshal(self) -> dict[str, Any]:
|
|
return {
|
|
"duration": self.duration,
|
|
"path": self.path.as_posix(),
|
|
"line": self.line,
|
|
"limit": self.limit,
|
|
"label": self.label,
|
|
}
|
|
|
|
|
|
@final
|
|
@dataclasses.dataclass
|
|
class _AssertRuntime:
|
|
"""Prepare for, measure, and assert about the time taken by code in the context.
|
|
|
|
Defaults are set for single-threaded CPU usage timing without garbage collection.
|
|
|
|
In general, there is no generally correct setup for benchmarking. Only measuring
|
|
a single thread's time using the CPU is not very useful for multithreaded or
|
|
multiprocessed code. Disabling garbage collection, or forcing it ahead of time,
|
|
makes the benchmark not identify any issues the code may introduce in terms of
|
|
actually causing relevant gc slowdowns. And so on...
|
|
|
|
Produces output of the following form.
|
|
|
|
Asserting maximum duration: full block
|
|
/home/altendky/repos/chia-blockchain/tests/core/full_node/test_performance.py:187
|
|
run time: 0.027789528900002837
|
|
allowed: 0.1
|
|
percent: 28 %
|
|
"""
|
|
|
|
# A class is only being used here, to make __tracebackhide__ work.
|
|
# https://github.com/pytest-dev/pytest/issues/2057
|
|
|
|
seconds: float
|
|
# TODO: Optional?
|
|
label: str = ""
|
|
clock: Callable[[], float] = thread_time
|
|
gc_mode: GcMode = GcMode.disable
|
|
print: bool = True
|
|
overhead: float | None = None
|
|
entry_file: str | None = None
|
|
entry_line: int | None = None
|
|
_results: AssertRuntimeResults | None = None
|
|
runtime_manager: contextlib.AbstractContextManager[Future[RuntimeResults]] | None = None
|
|
runtime_results_callable: Future[RuntimeResults] | None = None
|
|
enable_assertion: bool = True
|
|
|
|
def __enter__(self) -> Future[AssertRuntimeResults]:
|
|
self.entry_file, self.entry_line = caller_file_and_line(
|
|
relative_to=(
|
|
pathlib.Path(chia.__file__).parent.parent,
|
|
pathlib.Path(chia._tests.__file__).parent.parent,
|
|
)
|
|
)
|
|
|
|
self.runtime_manager = measure_runtime(
|
|
clock=self.clock, gc_mode=self.gc_mode, overhead=self.overhead, print_results=False
|
|
)
|
|
self.runtime_results_callable = self.runtime_manager.__enter__()
|
|
self.results_callable: Future[AssertRuntimeResults] = Future()
|
|
|
|
return self.results_callable
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc: BaseException | None,
|
|
traceback: TracebackType | None,
|
|
) -> None:
|
|
if (
|
|
self.entry_file is None
|
|
or self.entry_line is None
|
|
or self.runtime_manager is None
|
|
or self.runtime_results_callable is None
|
|
):
|
|
raise Exception("Context manager must be entered before exiting")
|
|
|
|
self.runtime_manager.__exit__(exc_type, exc, traceback)
|
|
|
|
runtime = self.runtime_results_callable.result(timeout=0)
|
|
results = AssertRuntimeResults.from_runtime_results(
|
|
results=runtime,
|
|
limit=self.seconds,
|
|
entry_file=self.entry_file,
|
|
entry_line=self.entry_line,
|
|
overhead=self.overhead,
|
|
)
|
|
|
|
self.results_callable.set_result(results)
|
|
|
|
if self.print:
|
|
print(results.block(label=self.label))
|
|
|
|
if ether.record_property is not None:
|
|
data = BenchmarkData(
|
|
duration=results.duration,
|
|
path=pathlib.Path(self.entry_file),
|
|
line=self.entry_line,
|
|
limit=self.seconds,
|
|
label=self.label,
|
|
)
|
|
|
|
ether.record_property(
|
|
data.tag,
|
|
json.dumps(data.marshal(), ensure_ascii=True, sort_keys=True),
|
|
)
|
|
|
|
if exc_type is None and self.enable_assertion:
|
|
__tracebackhide__ = True
|
|
assert runtime.duration < self.seconds, results.message()
|
|
|
|
|
|
@final
|
|
@dataclasses.dataclass
|
|
class BenchmarkRunner:
|
|
enable_assertion: bool = True
|
|
test_id: TestId | None = None
|
|
overhead: float | None = None
|
|
|
|
def assert_runtime(self, *args: Any, **kwargs: Any) -> _AssertRuntime:
|
|
kwargs.setdefault("enable_assertion", self.enable_assertion)
|
|
kwargs.setdefault("overhead", self.overhead)
|
|
return _AssertRuntime(*args, **kwargs)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def assert_rpc_error(error: str) -> Iterator[None]:
|
|
with pytest.raises(ValueError) as exception_info:
|
|
yield
|
|
assert error in exception_info.value.args[0]["error"]
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def closing_chia_root_popen(chia_root: ChiaRoot, args: list[str]) -> Iterator[subprocess.Popen[Any]]:
|
|
environment = {**os.environ, "CHIA_ROOT": os.fspath(chia_root.path)}
|
|
|
|
with subprocess.Popen(args=args, env=environment) as process:
|
|
try:
|
|
yield process
|
|
finally:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
|
|
|
|
# https://github.com/pytest-dev/pytest/blob/7.3.1/src/_pytest/mark/__init__.py#L45
|
|
Marks = pytest.MarkDecorator | Collection[pytest.MarkDecorator | pytest.Mark]
|
|
|
|
|
|
class DataCase(Protocol):
|
|
marks: Marks
|
|
|
|
@property
|
|
def id(self) -> str: ...
|
|
|
|
|
|
def datacases(*cases: DataCase, _name: str = "case") -> pytest.MarkDecorator:
|
|
return pytest.mark.parametrize(
|
|
argnames=_name,
|
|
argvalues=[pytest.param(case, id=case.id, marks=case.marks) for case in cases],
|
|
)
|
|
|
|
|
|
class DataCasesDecorator(Protocol):
|
|
def __call__(self, *cases: DataCase, _name: str = "case") -> pytest.MarkDecorator: ...
|
|
|
|
|
|
def named_datacases(name: str) -> DataCasesDecorator:
|
|
return functools.partial(datacases, _name=name)
|
|
|
|
|
|
def boolean_datacases(name: str, false: str, true: str) -> pytest.MarkDecorator:
|
|
return pytest.mark.parametrize(
|
|
argnames=name,
|
|
argvalues=[
|
|
pytest.param(False, id=false),
|
|
pytest.param(True, id=true),
|
|
],
|
|
)
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class CoinGenerator:
|
|
_seed: int = -1
|
|
|
|
def _get_hash(self) -> bytes32:
|
|
self._seed += 1
|
|
return std_hash(self._seed.to_bytes(length=32, byteorder="big"))
|
|
|
|
def _get_amount(self) -> uint64:
|
|
self._seed += 1
|
|
return uint64(self._seed)
|
|
|
|
def get(self, parent_coin_id: bytes32 | None = None, include_hint: bool = True) -> HintedCoin:
|
|
if parent_coin_id is None:
|
|
parent_coin_id = self._get_hash()
|
|
hint = None
|
|
if include_hint:
|
|
hint = self._get_hash()
|
|
return HintedCoin(Coin(parent_coin_id, self._get_hash(), self._get_amount()), hint)
|
|
|
|
|
|
def coin_creation_args(hinted_coin: HintedCoin) -> list[Any]:
|
|
if hinted_coin.hint is not None:
|
|
memos = [hinted_coin.hint]
|
|
else:
|
|
memos = []
|
|
return [ConditionOpcode.CREATE_COIN, hinted_coin.coin.puzzle_hash, hinted_coin.coin.amount, memos]
|
|
|
|
|
|
def create_logger(file: TextIO = sys.stdout) -> logging.Logger:
|
|
logger = logging.getLogger()
|
|
logger.setLevel(level=logging.DEBUG)
|
|
stream_handler = logging.StreamHandler(stream=file)
|
|
log_date_format = "%Y-%m-%dT%H:%M:%S"
|
|
file_log_formatter = logging.Formatter(
|
|
fmt="%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s",
|
|
datefmt=log_date_format,
|
|
)
|
|
stream_handler.setFormatter(file_log_formatter)
|
|
logger.addHandler(hdlr=stream_handler)
|
|
|
|
return logger
|
|
|
|
|
|
def invariant_check_mempool(mempool: Mempool) -> None:
|
|
cursor = mempool._db_conn.execute("SELECT COALESCE(SUM(cost), 0), COALESCE(SUM(fee), 0) FROM tx")
|
|
val = cursor.fetchone()
|
|
assert (mempool._total_cost, mempool._total_fee) == val
|
|
|
|
cursor = mempool._db_conn.execute("SELECT coin_id, tx FROM spends")
|
|
for coin_id, item_id in cursor.fetchall():
|
|
item = mempool._items.get(item_id)
|
|
assert item is not None
|
|
# item is expected to contain a spend of coin_id, but it might be a
|
|
# fast-forward spend, in which case the dictionary won't help us,
|
|
# but we'll have to do a linear search
|
|
if coin_id in item.bundle_coin_spends:
|
|
assert item.bundle_coin_spends[coin_id].coin_spend.coin.name() == coin_id
|
|
continue
|
|
|
|
assert any(
|
|
i.latest_singleton_lineage is not None and i.latest_singleton_lineage.coin_id == coin_id
|
|
for i in item.bundle_coin_spends.values()
|
|
)
|
|
|
|
|
|
async def wallet_height_at_least(wallet_node: WalletNode, h: uint32) -> bool:
|
|
height = await wallet_node.wallet_state_manager.blockchain.get_finished_sync_up_to()
|
|
return height == h
|
|
|
|
|
|
@final
|
|
@dataclass
|
|
class RecordingWebServer:
|
|
web_server: WebServer
|
|
requests: list[web.Request] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
async def create(
|
|
cls,
|
|
hostname: str,
|
|
port: uint16,
|
|
max_request_body_size: int = 1024**2, # Default `client_max_size` from web.Application
|
|
ssl_context: ssl.SSLContext | None = None,
|
|
prefer_ipv6: bool = False,
|
|
) -> RecordingWebServer:
|
|
web_server = await WebServer.create(
|
|
hostname=hostname,
|
|
port=port,
|
|
max_request_body_size=max_request_body_size,
|
|
ssl_context=ssl_context,
|
|
prefer_ipv6=prefer_ipv6,
|
|
start=False,
|
|
)
|
|
|
|
self = cls(web_server=web_server)
|
|
routes = [web.route(method="*", path=route, handler=func) for (route, func) in self.get_routes().items()]
|
|
web_server.add_routes(routes=routes)
|
|
await web_server.start()
|
|
return self
|
|
|
|
def get_routes(self) -> dict[str, Callable[[web.Request], Awaitable[web.Response]]]:
|
|
return {"/{path:.*}": self.handler}
|
|
|
|
async def handler(self, request: web.Request) -> web.Response:
|
|
self.requests.append(request)
|
|
|
|
request_json = await request.json()
|
|
if isinstance(request_json, dict) and "response" in request_json:
|
|
response = request_json["response"]
|
|
else:
|
|
response = {"success": True}
|
|
|
|
return aiohttp.web.json_response(data=response)
|
|
|
|
async def await_closed(self) -> None:
|
|
self.web_server.close()
|
|
await self.web_server.await_closed()
|
|
|
|
|
|
@final
|
|
@dataclasses.dataclass(frozen=True)
|
|
class TestId:
|
|
platform: str
|
|
test_path: tuple[str, ...]
|
|
ids: tuple[str, ...]
|
|
|
|
@classmethod
|
|
def create(cls, node: Node, platform: str = sys.platform) -> TestId:
|
|
test_path: list[str] = []
|
|
temp_node = node
|
|
while True:
|
|
name: str
|
|
if isinstance(temp_node, pytest.Function):
|
|
name = temp_node.originalname
|
|
elif isinstance(temp_node, pytest.Package):
|
|
# must check before pytest.Module since Package is a subclass
|
|
name = temp_node.name
|
|
elif isinstance(temp_node, pytest.Module):
|
|
name = temp_node.name[:-3]
|
|
else:
|
|
name = temp_node.name
|
|
test_path.insert(0, name)
|
|
if isinstance(temp_node.parent, pytest.Session) or temp_node.parent is None:
|
|
break
|
|
temp_node = temp_node.parent
|
|
|
|
# TODO: can we avoid parsing the id's etc from the node name?
|
|
_test_name, delimiter, rest = node.name.partition("[")
|
|
ids: tuple[str, ...]
|
|
if delimiter == "":
|
|
ids = ()
|
|
else:
|
|
ids = tuple(rest.rstrip("]").split("-"))
|
|
|
|
return cls(
|
|
platform=platform,
|
|
test_path=tuple(test_path),
|
|
ids=ids,
|
|
)
|
|
|
|
@classmethod
|
|
def unmarshal(cls, marshalled: dict[str, Any]) -> TestId:
|
|
return cls(
|
|
platform=marshalled["platform"],
|
|
test_path=tuple(marshalled["test_path"]),
|
|
ids=tuple(marshalled["ids"]),
|
|
)
|
|
|
|
def marshal(self) -> dict[str, Any]:
|
|
return {
|
|
"platform": self.platform,
|
|
"test_path": self.test_path,
|
|
"ids": self.ids,
|
|
}
|
|
|
|
|
|
T_ComparableEnum = TypeVar("T_ComparableEnum", bound="ComparableEnum")
|
|
|
|
|
|
class ComparableEnum(Enum):
|
|
def __lt__(self, other: Self) -> object:
|
|
if self.__class__ is not other.__class__:
|
|
return NotImplemented
|
|
|
|
return self.value.__lt__(other.value)
|
|
|
|
def __le__(self, other: Self) -> object:
|
|
if self.__class__ is not other.__class__:
|
|
return NotImplemented
|
|
|
|
return self.value.__le__(other.value)
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
if self.__class__ is not other.__class__:
|
|
return False
|
|
|
|
return cast(bool, self.value.__eq__(cast(Self, other).value))
|
|
|
|
def __ne__(self, other: object) -> bool:
|
|
if self.__class__ is not other.__class__:
|
|
return True
|
|
|
|
return cast(bool, self.value.__ne__(cast(Self, other).value))
|
|
|
|
def __gt__(self, other: Self) -> object:
|
|
if self.__class__ is not other.__class__:
|
|
return NotImplemented
|
|
|
|
return self.value.__gt__(other.value)
|
|
|
|
def __ge__(self, other: Self) -> object:
|
|
if self.__class__ is not other.__class__:
|
|
return NotImplemented
|
|
|
|
return self.value.__ge__(other.value)
|
|
|
|
|
|
def is_attribute_local(o: object, name: str) -> bool:
|
|
return name in getattr(o, "__dict__", ()) or name in getattr(o, "__slots__", ())
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def patch_request_handler(
|
|
api: ApiProtocol | type[ApiProtocol],
|
|
handler: Callable[..., Awaitable[Message | None]],
|
|
request_type: ProtocolMessageTypes | None = None,
|
|
) -> Iterator[None]:
|
|
if request_type is None:
|
|
request_type = ProtocolMessageTypes[handler.__name__]
|
|
|
|
metadata = ApiMetadata.copy(api.metadata)
|
|
original_request = metadata.message_type_to_request.pop(request_type)
|
|
decorator = metadata.request(
|
|
peer_required=original_request.peer_required,
|
|
bytes_required=original_request.bytes_required,
|
|
execute_task=original_request.execute_task,
|
|
reply_types=original_request.reply_types,
|
|
request_type=original_request.request_type,
|
|
)
|
|
decorator(handler)
|
|
|
|
was_local = is_attribute_local(api, "metadata")
|
|
original = api.metadata
|
|
# when an instance is passed, this is intentionally assigning to an instance
|
|
# counter to the class variable hint
|
|
api.metadata = metadata # type: ignore[misc]
|
|
try:
|
|
yield
|
|
finally:
|
|
if was_local:
|
|
# when an instance is passed, this is intentionally assigning to an instance
|
|
# counter to the class variable hint
|
|
api.metadata = original # type: ignore[misc]
|
|
else:
|
|
del api.metadata
|