merrow
881ef45be5
ci.yml (push) Invalid workflow file
- 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 репозитория
116 lines
4.2 KiB
Python
116 lines
4.2 KiB
Python
# -*- 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"]
|