86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
from pathlib import Path
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
|
|
# Make `site/` importable as top-level modules: app, routes, db, operations, ...
|
|
SITE_DIR = Path(__file__).resolve().parents[1] / "site"
|
|
if str(SITE_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SITE_DIR))
|
|
|
|
# Путь к polygon: соседняя репа в корне autotest
|
|
POLYGON_DIR = Path(__file__).resolve().parents[2] / "polygon" / "site"
|
|
|
|
|
|
@pytest.fixture
|
|
def app_client(polygon_server):
|
|
"""Flask test client, направленный в локальный Polygon TEST."""
|
|
from app import app
|
|
|
|
old_polygon_endpoint = app.config.get("POLYGON_ENDPOINT", "")
|
|
app.config["POLYGON_ENDPOINT"] = f"{polygon_server}/api/v1/svc"
|
|
app.config.update(TESTING=True)
|
|
try:
|
|
with app.test_client() as client:
|
|
yield client
|
|
finally:
|
|
app.config["POLYGON_ENDPOINT"] = old_polygon_endpoint
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def polygon_server():
|
|
"""Запустить polygon как subprocess на весь pytest-session.
|
|
|
|
Используется для интеграционных тестов app-autotest ↔ polygon.
|
|
После всех тестов процесс убивается.
|
|
"""
|
|
if not POLYGON_DIR.is_dir():
|
|
pytest.skip("polygon repo not found")
|
|
|
|
with socket.socket() as port_socket:
|
|
port_socket.bind(("127.0.0.1", 0))
|
|
port = port_socket.getsockname()[1]
|
|
polygon_url = f"http://127.0.0.1:{port}"
|
|
env = os.environ.copy()
|
|
env["POLYGON_STANDS"] = "test:services/test"
|
|
|
|
proc = subprocess.Popen(
|
|
["python3", "-c", f"import app; app.app.run(host='127.0.0.1', port={port}, debug=False)"],
|
|
cwd=str(POLYGON_DIR),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
env=env,
|
|
)
|
|
|
|
# Ждём готовности (poll /health, timeout 10s)
|
|
deadline = time.time() + 10
|
|
while time.time() < deadline:
|
|
try:
|
|
r = requests.get(f"{polygon_url}/health", timeout=2)
|
|
if r.status_code == 200:
|
|
break
|
|
except requests.ConnectionError:
|
|
time.sleep(0.5)
|
|
else:
|
|
proc.terminate()
|
|
proc.wait()
|
|
pytest.fail("polygon did not start within 10s")
|
|
|
|
yield polygon_url
|
|
|
|
proc.terminate()
|
|
proc.wait()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_polygon(polygon_server):
|
|
"""Перед каждым интеграционным тестом — сброс состояния polygon."""
|
|
requests.post(f"{polygon_server}/test/api/v1/svc/_mock/reset", timeout=5)
|
|
|