44 lines
2.0 KiB
Python
44 lines
2.0 KiB
Python
 
|
"""Тесты идут против приложения в контейнерах (docker compose up -d) с учётными данными из .env."""
|
||
|
|
import pathlib
|
||
|
|
import random
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
ENV = dict(l.split("=", 1) for l in pathlib.Path(__file__).resolve().parent.parent.joinpath(".env").read_text().split() if "=" in l)
|
||
|
|
BASE = f"http://127.0.0.1:{ENV['APP_PORT']}/api/v1"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(scope="session")
|
||
|
|
def client():
|
||
|
|
c = httpx.Client(base_url=BASE, timeout=30)
|
||
|
|
tok = c.post("/auth/login", json={"username": ENV["ADMIN_USERNAME"], "password": ENV["ADMIN_PASSWORD"]}).json()["access_token"]
|
||
|
|
c.headers["Authorization"] = f"Bearer {tok}"
|
||
|
|
return c
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def org(client):
|
||
|
|
"""Временная организация с дефолтным VRF; после теста удаляется вместе с содержимым."""
|
||
|
|
o = client.post("/organizations", json={"name": f"test-{uuid.uuid4().hex[:8]}", "inn": "".join(random.choices("0123456789", k=10))}).json()
|
||
|
|
o["vrf_id"] = client.get("/vrfs", params={"organization_id": o["id"]}).json()["items"][0]["id"]
|
||
|
|
yield o
|
||
|
|
for p in client.get("/prefixes", params={"organization_id": o["id"]}).json()["items"]:
|
||
|
|
client.delete(f"/prefixes/{p['id']}", params={"force": True})
|
||
|
|
for d in client.get("/devices", params={"organization_id": o["id"]}).json()["items"]:
|
||
|
|
client.delete(f"/devices/{d['id']}")
|
||
|
|
for v in client.get("/vrfs", params={"organization_id": o["id"]}).json()["items"]:
|
||
|
|
client.delete(f"/vrfs/{v['id']}")
|
||
|
|
client.delete(f"/organizations/{o['id']}")
|
||
|
|
|
||
|
|
|
||
|
|
DSN = f"postgresql://{ENV['POSTGRES_USER']}:{ENV['POSTGRES_PASSWORD']}@127.0.0.1:{ENV['DB_HOST_PORT']}/{ENV['POSTGRES_DB']}"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def db():
|
||
|
|
"""Прямое подключение к БД — для сценариев, недоступных через API (старые записи, временные пользователи)."""
|
||
|
|
import psycopg
|
||
|
|
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
|
|
yield conn
|