- crm/tests: 27 тестов через настоящий server.main на sqlite-бэкенде (валидация заказов, цена только из каталога, honeypot, дедупликация, трекинг и отмена гостем, стоп-лист, аутентификация, гигиена текста, гейт расписания бара) - .gitea/workflows/ci.yml: test-джоба + deploy (только push в main, после тестов): rsync кода -> pre-flight docker compose config -> pg_dump бэкап БД (ротация 7) -> up -d --build -> health-check - деплой на neon-bar86.ru идёт деплой-ключом из secrets репозитория
This commit is contained in:
1 parent
da5b259d06
commit
881ef45be5
8 files changed
+469
No files matched your search
@@ -0,0 +1,115 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Поднимает CRM как настоящий процесс (server.main) на sqlite-бэкенде.
|
||||
|
||||
- CRM_DB=sqlite — тот же режим, что и локальная разработка; файл базы
|
||||
crm/data/crm.db (в CI каталог данных пересоздаётся с нуля — шаг в workflow).
|
||||
- tests/_server_main.py фиксирует барное время на пятнице 21:00 (бар открыт),
|
||||
иначе гейт «заказы по времени» делает тесты зависимыми от часа запуска CI.
|
||||
|
||||
Гейт расписания и гигиена гостевого текста тестируются отдельно — как чистые
|
||||
функции, без сервера (test_gate_and_text.py).
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
CRM_DIR = Path(__file__).resolve().parents[1]
|
||||
|
||||
# Пароли первого старта — только из окружения (F-2), дефолтные запрещены (F-13)
|
||||
os.environ.setdefault("CRM_OWNER_PASSWORD", "ci-owner-pass-1")
|
||||
os.environ.setdefault("CRM_ADMIN_PASSWORD", "ci-admin-pass-1")
|
||||
os.environ.setdefault("CRM_WAITER_PASSWORD", "ci-waiter-pass-1")
|
||||
os.environ["CRM_ALLOW_DEFAULT_PASSWORDS"] = "0"
|
||||
os.environ["CRM_SEED_DEMO_ORDERS"] = "0"
|
||||
# и в самом процессе тестов (юнит-тесты импортируют server/storage напрямую):
|
||||
# sqlite-ветка storage не требует psycopg
|
||||
os.environ["CRM_DB"] = "sqlite"
|
||||
|
||||
|
||||
def _free_port() -> int:
|
||||
s = socket.socket()
|
||||
s.bind(("127.0.0.1", 0))
|
||||
port = s.getsockname()[1]
|
||||
s.close()
|
||||
return port
|
||||
|
||||
|
||||
class Api:
|
||||
"""Минимальный HTTP-клиент: возвращает (status, json-тело) для любого ответа,
|
||||
включая 4xx/5xx — тестам нужен именно статус, а не исключение."""
|
||||
|
||||
def __init__(self, base: str):
|
||||
self.base = base
|
||||
|
||||
def request(self, method, path, body=None, token=None):
|
||||
data = None
|
||||
headers = {}
|
||||
if body is not None:
|
||||
data = json.dumps(body).encode("utf-8")
|
||||
headers["Content-Type"] = "application/json"
|
||||
if token:
|
||||
headers["X-Auth-Token"] = token
|
||||
req = urllib.request.Request(self.base + path, data=data, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
raw = resp.read()
|
||||
return resp.status, (json.loads(raw) if raw else None)
|
||||
except urllib.error.HTTPError as e:
|
||||
raw = e.read()
|
||||
try:
|
||||
return e.code, (json.loads(raw) if raw else None)
|
||||
except ValueError:
|
||||
return e.code, None
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def api():
|
||||
port = _free_port()
|
||||
env = {**os.environ, "CRM_DB": "sqlite", "CRM_PORT": str(port)}
|
||||
log = tempfile.TemporaryFile()
|
||||
proc = subprocess.Popen(
|
||||
[sys.executable, str(CRM_DIR / "tests" / "_server_main.py")],
|
||||
cwd=CRM_DIR, env=env, stdout=log, stderr=subprocess.STDOUT,
|
||||
)
|
||||
base = f"http://127.0.0.1:{port}"
|
||||
for _ in range(150):
|
||||
if proc.poll() is not None:
|
||||
log.seek(0)
|
||||
raise RuntimeError("CRM не поднялся:\n" + log.read().decode("utf-8", "replace"))
|
||||
try:
|
||||
with urllib.request.urlopen(base + "/api/healthz", timeout=2) as r:
|
||||
if r.status == 200:
|
||||
break
|
||||
except Exception:
|
||||
time.sleep(0.2)
|
||||
else:
|
||||
proc.terminate()
|
||||
raise RuntimeError("CRM не ответил на /api/healthz за 30 секунд")
|
||||
yield Api(base)
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
proc.wait(timeout=5)
|
||||
log.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def owner_token(api):
|
||||
status, body = api.request(
|
||||
"POST", "/api/login",
|
||||
{"login": "owner", "password": os.environ["CRM_OWNER_PASSWORD"]},
|
||||
)
|
||||
assert status == 200, body
|
||||
assert body["role"] == "owner"
|
||||
return body["token"]
|
||||
Reference in new issue
Block a user