2023-09-28 09:41:33 +02:00
|
|
|
from pathlib import Path
|
|
|
|
from threading import Thread
|
|
|
|
from typing import Any, Iterator
|
2023-09-28 10:57:24 +02:00
|
|
|
|
|
|
|
import pytest
|
2023-09-28 09:41:33 +02:00
|
|
|
from click.testing import CliRunner
|
2023-09-28 10:57:24 +02:00
|
|
|
|
2023-09-28 09:41:33 +02:00
|
|
|
from . import PatroniAPI
|
2023-09-28 09:37:44 +02:00
|
|
|
|
|
|
|
|
2023-10-03 14:29:21 +02:00
|
|
|
@pytest.fixture(
|
|
|
|
params=[False, True],
|
|
|
|
ids=lambda v: "new-replica-state" if v else "old-replica-state",
|
|
|
|
)
|
|
|
|
def old_replica_state(request: Any) -> Any:
|
|
|
|
return request.param
|
2023-09-28 10:57:24 +02:00
|
|
|
|
|
|
|
|
2023-09-28 09:41:33 +02:00
|
|
|
@pytest.fixture(scope="session")
|
|
|
|
def datadir() -> Path:
|
|
|
|
return Path(__file__).parent / "json"
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
|
|
def patroni_api(
|
|
|
|
tmp_path_factory: pytest.TempPathFactory, datadir: Path
|
|
|
|
) -> Iterator[PatroniAPI]:
|
|
|
|
"""A fake HTTP server for the Patroni API serving files from a temporary
|
|
|
|
directory.
|
|
|
|
"""
|
|
|
|
httpd = PatroniAPI(tmp_path_factory.mktemp("api"), datadir=datadir)
|
|
|
|
t = Thread(target=httpd.serve_forever)
|
|
|
|
t.start()
|
|
|
|
yield httpd
|
|
|
|
httpd.shutdown()
|
|
|
|
t.join()
|
2023-09-28 09:41:33 +02:00
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def runner() -> CliRunner:
|
|
|
|
"""A CliRunner with stdout and stderr not mixed."""
|
|
|
|
return CliRunner(mix_stderr=False)
|