mirror of
https://github.com/neondatabase/neon.git
synced 2025-12-23 06:09:59 +00:00
5bd8e2363a
This will help to keep us from using deprecated Python features going forward. Signed-off-by: Tristan Partin <tristan@neon.tech>
62 lines
1.5 KiB
Python
62 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from functools import cached_property
|
|
|
|
import pytest
|
|
|
|
|
|
class PgStatTable:
|
|
table: str
|
|
columns: list[str]
|
|
additional_query: str
|
|
|
|
def __init__(self, table: str, columns: list[str], filter_query: str = ""):
|
|
self.table = table
|
|
self.columns = columns
|
|
self.additional_query = filter_query
|
|
|
|
@cached_property
|
|
def query(self) -> str:
|
|
return f"SELECT {','.join(self.columns)} FROM {self.table} {self.additional_query}"
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def pg_stats_rw() -> list[PgStatTable]:
|
|
return [
|
|
PgStatTable(
|
|
"pg_stat_database",
|
|
["tup_returned", "tup_fetched", "tup_inserted", "tup_updated", "tup_deleted"],
|
|
"WHERE datname='postgres'",
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def pg_stats_ro() -> list[PgStatTable]:
|
|
return [
|
|
PgStatTable(
|
|
"pg_stat_database", ["tup_returned", "tup_fetched"], "WHERE datname='postgres'"
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def pg_stats_wo() -> list[PgStatTable]:
|
|
return [
|
|
PgStatTable(
|
|
"pg_stat_database",
|
|
["tup_inserted", "tup_updated", "tup_deleted"],
|
|
"WHERE datname='postgres'",
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def pg_stats_wal() -> list[PgStatTable]:
|
|
return [
|
|
PgStatTable(
|
|
"pg_stat_wal",
|
|
["wal_records", "wal_fpi", "wal_bytes", "wal_buffers_full", "wal_write"],
|
|
)
|
|
]
|