v1.5.4: test infrastructure + bug fixes + CI/CD + deploy gate
- 6 new test files: route discovery, strict permissions, role switch, CRUD operations, 500 error protection - permission_config.py: central permission definitions for all 110+ routes - Bug fixes: is_active checkbox in users_edit, int(id) 500 crash, missing validation in users_create/users_delete - Pre-commit hooks: lint + static routing tests - GitHub Actions CI: full pytest suite on push - Deploy gate: pytest runs before deploy, aborts on failure - All 3 deploy scripts: --skip-tests flag support
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
name: Tests
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main, master, develop]
|
||||
pull_request:
|
||||
branches: [main, master]
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
services:
|
||||
postgres:
|
||||
image: postgres:15
|
||||
env:
|
||||
POSTGRES_DB: aegisone
|
||||
POSTGRES_USER: aegisone
|
||||
POSTGRES_PASSWORD: aegisone_pass
|
||||
ports:
|
||||
- 5432:5432
|
||||
options: >-
|
||||
--health-cmd pg_isready
|
||||
--health-interval 5s
|
||||
--health-timeout 5s
|
||||
--health-retries 5
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install dependencies
|
||||
working-directory: py_service
|
||||
run: |
|
||||
pip install -r requirements.txt
|
||||
pip install ruff
|
||||
|
||||
- name: Pre-test lint & static analysis
|
||||
working-directory: py_service
|
||||
run: |
|
||||
ruff check .
|
||||
pytest tests/test_routes_defined.py -v --no-header -q
|
||||
python -c "from tests.test_00_discover_routes import collect_app_routes; r=collect_app_routes(); print(f'Discovered {len(r)} routes')"
|
||||
|
||||
- name: Start server
|
||||
working-directory: py_service
|
||||
run: |
|
||||
uvicorn app.main:app --host 0.0.0.0 --port 8000 &
|
||||
sleep 3
|
||||
curl -sf http://localhost:8000/health || (echo "Server failed to start" && exit 1)
|
||||
|
||||
- name: Run tests
|
||||
working-directory: py_service
|
||||
run: |
|
||||
pytest -v --tb=short -x
|
||||
|
||||
- name: Test summary
|
||||
if: always()
|
||||
run: echo "CI test run complete"
|
||||
@@ -0,0 +1,15 @@
|
||||
repos:
|
||||
- repo: local
|
||||
hooks:
|
||||
- id: pytest-routes
|
||||
name: pytest (route definition check)
|
||||
description: Verify all routes are defined and covered in permission_config
|
||||
entry: pytest py_service/tests/test_routes_defined.py py_service/tests/test_00_discover_routes.py -v --no-header -q
|
||||
language: system
|
||||
pass_filenames: false
|
||||
always_run: true
|
||||
- id: ruff
|
||||
name: ruff linter
|
||||
entry: ruff check py_service/
|
||||
language: system
|
||||
pass_filenames: false
|
||||
@@ -0,0 +1,123 @@
|
||||
#!/bin/bash
|
||||
# ==========================================
|
||||
# Deploy update — быстрый деплой фиксов
|
||||
# Usage: bash other/deploy_update.sh [--skip-tests] [patch|minor|major]
|
||||
# ==========================================
|
||||
set -e
|
||||
|
||||
SKIP_TESTS=false
|
||||
BUMP="patch"
|
||||
for arg in "$@"; do
|
||||
case "$arg" in
|
||||
--skip-tests) SKIP_TESTS=true ;;
|
||||
patch|minor|major) BUMP="$arg" ;;
|
||||
esac
|
||||
done
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
|
||||
SERVER="angel@81.177.141.34"
|
||||
ARCHIVE="$SCRIPT_DIR/aegisone-update.tar.gz"
|
||||
BOT="/opt/projects/aegisone-py/max_bot"
|
||||
PY="/opt/projects/aegisone-py"
|
||||
|
||||
echo "=========================================="
|
||||
echo " AegisOne — Deploy Update"
|
||||
echo "=========================================="
|
||||
|
||||
# ==========================================
|
||||
# Gate: run tests before deploy
|
||||
# ==========================================
|
||||
if [ "$SKIP_TESTS" = false ]; then
|
||||
echo ""
|
||||
echo "[CHECK] Running tests..."
|
||||
cd "$REPO_ROOT/py_service"
|
||||
if curl -sf http://localhost:8000/health > /dev/null 2>&1; then
|
||||
if ! pytest -v --tb=short -x; then
|
||||
echo ""
|
||||
echo "❌ Tests failed. Deploy aborted."
|
||||
echo " Fix errors or use --skip-tests to force deploy."
|
||||
exit 1
|
||||
fi
|
||||
echo " ✅ Tests passed"
|
||||
else
|
||||
echo " ⚠️ Server not reachable at localhost:8000"
|
||||
echo " Start the service first, or use --skip-tests"
|
||||
exit 1
|
||||
fi
|
||||
else
|
||||
echo " ⚠️ Tests skipped (--skip-tests flag)"
|
||||
fi
|
||||
|
||||
# Bump version & update changelog
|
||||
cd "$REPO_ROOT"
|
||||
python py_service/scripts/bump_version.py "$BUMP" '{"features":["Обновление сервисного портала"],"fixes":["Исправления и улучшения"]}'
|
||||
VERSION=$(cat py_service/version.txt)
|
||||
echo "Version: $VERSION"
|
||||
|
||||
echo ""
|
||||
echo "[1/5] Uploading archive..."
|
||||
scp "$ARCHIVE" "$SERVER:/tmp/aegisone-update.tar.gz"
|
||||
|
||||
echo ""
|
||||
echo "[2/5] Applying update on server..."
|
||||
ssh "$SERVER" << 'SSHEOF'
|
||||
set -e
|
||||
cd /tmp
|
||||
rm -rf update
|
||||
mkdir update && tar xzf aegisone-update.tar.gz -C update
|
||||
PY="/opt/projects/aegisone-py"
|
||||
|
||||
# SQL migration (если есть)
|
||||
MIG=$(ls update/py_service/sql/v*.sql 2>/dev/null | head -1)
|
||||
if [ -n "$MIG" ]; then
|
||||
PG=$(docker ps --filter "name=aegisone-postgres" --format "{{.ID}}" | head -1)
|
||||
if [ -n "$PG" ]; then
|
||||
docker cp "$MIG" "$PG:/tmp/mig.sql"
|
||||
docker exec "$PG" psql -U aegisone -d aegisone -f /tmp/mig.sql 2>&1 | grep -v "^ALTER\|^CREATE\|^INSERT\|^ON CONFLICT" || true
|
||||
echo " ✅ Migration applied"
|
||||
fi
|
||||
fi
|
||||
|
||||
# Copy all py_service files
|
||||
rsync -a --delete update/py_service/ "$PY/"
|
||||
echo " ✅ py_service files copied"
|
||||
|
||||
# Copy max_bot files (если есть)
|
||||
if [ -d "update/max_bot" ]; then
|
||||
rsync -a --delete update/max_bot/ "$PY/max_bot/"
|
||||
echo " ✅ max_bot files copied"
|
||||
fi
|
||||
|
||||
rm -rf /tmp/update
|
||||
SSHEOF
|
||||
|
||||
echo ""
|
||||
echo "[3/5] Rebuilding max_bot container..."
|
||||
ssh "$SERVER" "cd $BOT && docker compose up -d --build 2>&1 | tail -5"
|
||||
|
||||
echo ""
|
||||
echo "[4/5] Rebuilding py_service container..."
|
||||
ssh "$SERVER" "cd $PY && docker compose up -d --build 2>&1 | tail -5"
|
||||
|
||||
echo ""
|
||||
echo "[5/5] Health check..."
|
||||
sleep 5
|
||||
ssh "$SERVER" '
|
||||
echo -n " max_bot: "
|
||||
curl -s -o /dev/null -w "%{http_code}" http://localhost:8002/health || echo "FAIL"
|
||||
echo -n " py_service: "
|
||||
curl -s -o /dev/null -w "%{http_code}" http://localhost:8000/health || echo "FAIL"
|
||||
'
|
||||
|
||||
# Commit changelog
|
||||
cd "$REPO_ROOT"
|
||||
git add py_service/version.txt py_service/CHANGELOG.md
|
||||
git diff --cached --quiet || git commit -m "v$VERSION: changelog [skip ci]"
|
||||
|
||||
echo ""
|
||||
echo "=========================================="
|
||||
echo " ✅ Update complete: v$VERSION"
|
||||
echo " Bot: https://max.aegisone.ru"
|
||||
echo " Portal: https://service.aegisone.ru"
|
||||
echo "=========================================="
|
||||
@@ -1,5 +1,47 @@
|
||||
# Changelog — AegisOne Service Portal
|
||||
|
||||
## 1.5.4 (22.05.2026)
|
||||
|
||||
## 1.5.3 (22.05.2026)
|
||||
### Новые функции
|
||||
- П.8: Роли: обновлён список MENU_ITEMS под новую структуру (7 секций: Панель, Управление, Работа, Чат-бот Max, Контент, Внешние системы, Настройки)
|
||||
- П.8: Коэффициенты и Тест формул объединены в одну страницу **Формулы** (/service/formulas) с 2 табами
|
||||
- П.8: Чат-бот Max в Настройках — 6 табов (Общие, Шаблоны ответов, Функции, Категории, База знаний, Тестовый прогон)
|
||||
- П.8: Расширены дефолтные права engineer (charts, bot_analytics, bot_conversations)
|
||||
- Переключатель ролей в navbar: owner может переключиться на engineer/technician, engineer — на technician
|
||||
- Тема: текст (Светлая/Тёмная/Авто) перенесён во всплывающую подсказку (title) на кнопке
|
||||
- Тесты (pytest): проверка регистрации всех роутов smoke-тест для сервера
|
||||
### Исправления и улучшения
|
||||
- Все старые отдельные страницы (formula_test, coefficients, bot_responses, bot_features, bot_categories, bot_kb, bot_test_runner) продолжают работать через прямые роуты
|
||||
- Исправлен баг: BotTicket FK → bot_users (NoReferencedTableError при старте)
|
||||
|
||||
## 1.5.2 (21.05.2026)
|
||||
### Новые функции
|
||||
- П.8: Настройка портала (/service/portal-settings) — Yandex GPT ключ, Яндекс.Диск OAuth
|
||||
- П.8: Настройка Ролей (/service/role-settings) — owner управляет видимостью пунктов меню для engineer/technician
|
||||
- Ролевые права меню: sidebar скрывает пункты на основе role_menu_permissions
|
||||
### Исправления и улучшения
|
||||
- max_bot: Исправлен NameError в api_bot_features (добавлен импорт BotFeature)
|
||||
- py_service: Добавлен PermissionsMiddleware для загрузки прав меню в каждый шаблон
|
||||
- SQL: Дефолтные права engineer (10 пунктов) и technician (4 пункта) создаются при пустой таблице
|
||||
|
||||
## 1.5.1 (21.05.2026)
|
||||
### Новые функции
|
||||
- П.8: Рассылка с предпросмотром — выбор получателей (чекбоксы), живой предпросмотр, черновик, история рассылок
|
||||
- П.8: Фильтр согласия ФЗ-152 — рассылка только с `consent_given = true`
|
||||
### Исправления и улучшения
|
||||
- П.8: Аналитика — "Всего пользователей" теперь показывает только согласных, карточка согласий удалена
|
||||
- П.8: Пользователи — сортировка по всем колонкам (клик на заголовок), пагинация, фильтр по согласию
|
||||
- П.8: Пользователи — строки без согласия выделены красным фоном
|
||||
- max_bot: Broadcast API — preview (draft) + send режимы, recipients endpoint, history endpoint
|
||||
- py_service: Прокси-роуты для recipients/history
|
||||
|
||||
## 1.5.0 (21.05.2026)
|
||||
### Новые функции
|
||||
- Интеграция бота Max "София": управление ботом в сервисном портале
|
||||
- 11 новых страниц: настройки, шаблоны, функции, категории, база знаний, диалоги, пользователи, тестовый прогон, аналитика
|
||||
- SQL миграция v1.5.0: bot_tickets, bot_features.sla_ticket
|
||||
|
||||
## 1.4.0 (20.05.2026)
|
||||
|
||||
## 1.3.0 (20.05.2026)
|
||||
|
||||
@@ -43,7 +43,13 @@ def now_str():
|
||||
return datetime.now().strftime("%d.%m.%Y %H:%M")
|
||||
|
||||
def ctx(request, user, **kw):
|
||||
c = {"request": request, "user": user, "now": now_str()}
|
||||
original_user = request.session.get("original_user")
|
||||
available_roles = {"owner": ["owner", "engineer", "technician"], "engineer": ["engineer", "technician"]}
|
||||
c = {"request": request, "user": user, "now": now_str(),
|
||||
"role_permissions": getattr(request.state, 'role_permissions', {}),
|
||||
"role_override_active": original_user is not None,
|
||||
"original_role": original_user["role"] if original_user else None,
|
||||
"available_roles": available_roles.get(user["role"], [user["role"]]) if user else []}
|
||||
c.update(kw)
|
||||
return c
|
||||
|
||||
@@ -65,28 +71,39 @@ async def users_list(request: Request, db: AsyncSession = Depends(get_db), user:
|
||||
@router.post("/service/users/create")
|
||||
async def users_create(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
form = await request.form()
|
||||
login = form.get("login")
|
||||
password = form.get("password")
|
||||
if not login or not password:
|
||||
return RedirectResponse(url="/service/users", status_code=302)
|
||||
from app.auth import hash_password
|
||||
new_user = User(
|
||||
login=form.get("login"),
|
||||
password_hash=hash_password(form.get("password")),
|
||||
login=login,
|
||||
password_hash=hash_password(password),
|
||||
full_name=form.get("full_name"), role=form.get("role", "technician"),
|
||||
phone=form.get("phone", ""), email=form.get("email", ""),
|
||||
is_active=form.get("is_active", "1") == "1",
|
||||
)
|
||||
db.add(new_user); await db.flush()
|
||||
await audit_log(db, user["id"], "user_create", f"Создан пользователь {form.get('login')}", request.client.host or "")
|
||||
await audit_log(db, user["id"], "user_create", f"Создан пользователь {login}", request.client.host or "")
|
||||
return RedirectResponse(url="/service/users", status_code=302)
|
||||
|
||||
@router.post("/service/users/edit")
|
||||
async def users_edit(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
form = await request.form()
|
||||
u = await db.get(User, int(form.get("id")))
|
||||
user_id = form.get("id")
|
||||
if not user_id:
|
||||
return RedirectResponse(url="/service/users", status_code=302)
|
||||
try:
|
||||
u = await db.get(User, int(user_id))
|
||||
except (TypeError, ValueError):
|
||||
return RedirectResponse(url="/service/users", status_code=302)
|
||||
if not u: return RedirectResponse(url="/service/users", status_code=302)
|
||||
u.full_name = form.get("full_name", u.full_name)
|
||||
u.role = form.get("role", u.role)
|
||||
u.phone = form.get("phone", u.phone)
|
||||
u.email = form.get("email", u.email)
|
||||
u.is_active = form.get("is_active") == "1"
|
||||
if "is_active" in form:
|
||||
u.is_active = form["is_active"] == "1"
|
||||
from app.auth import hash_password
|
||||
pwd = form.get("password", "")
|
||||
if pwd: u.password_hash = hash_password(pwd)
|
||||
@@ -96,7 +113,14 @@ async def users_edit(request: Request, db: AsyncSession = Depends(get_db), user:
|
||||
|
||||
@router.post("/service/users/delete")
|
||||
async def users_delete(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
u = await db.get(User, int((await request.form()).get("id")))
|
||||
form = await request.form()
|
||||
user_id = form.get("id")
|
||||
if not user_id:
|
||||
return RedirectResponse(url="/service/users", status_code=302)
|
||||
try:
|
||||
u = await db.get(User, int(user_id))
|
||||
except (TypeError, ValueError):
|
||||
return RedirectResponse(url="/service/users", status_code=302)
|
||||
if u and u.role != "owner": await db.delete(u); await db.flush()
|
||||
return RedirectResponse(url="/service/users", status_code=302)
|
||||
|
||||
@@ -466,6 +490,34 @@ async def formula_test_page(request: Request, db: AsyncSession = Depends(get_db)
|
||||
return templates.TemplateResponse("pages/formula_test.html", ctx(request, user=user, formulas=formula_list, selected=selected, active_page="coefficients", title="Тест формул"))
|
||||
|
||||
|
||||
@router.get("/service/formulas")
|
||||
async def formulas_page(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
coeffs = (await db.execute(select(FormulaCoefficient).order_by(FormulaCoefficient.key))).scalars().all()
|
||||
groups = defaultdict(list)
|
||||
for c in coeffs:
|
||||
prefix = c.formula_ref.split(".")[0] if c.formula_ref else "_other"
|
||||
groups[prefix].append(c)
|
||||
grouped = {}
|
||||
for prefix in sorted(groups.keys()):
|
||||
info = FORMULA_GROUPS.get(prefix, {"name": prefix, "formula": "", "result": ""})
|
||||
grouped[prefix] = {"info": info, "coeffs": groups[prefix]}
|
||||
|
||||
formula_list = []
|
||||
for prefix in sorted(groups.keys()):
|
||||
info = FORMULA_GROUPS.get(prefix, {"name": prefix, "formula": "", "result": ""})
|
||||
sb = FORMULA_SANDBOX.get(prefix)
|
||||
formula_list.append({
|
||||
"key": prefix,
|
||||
"name": info["name"],
|
||||
"formula": info["formula"],
|
||||
"result": info["result"],
|
||||
"sandbox": sb,
|
||||
"coeffs": groups[prefix],
|
||||
})
|
||||
selected = formula_list[0]["key"] if formula_list else ""
|
||||
return templates.TemplateResponse("pages/formulas.html", ctx(request, user=user, formulas=formula_list, selected=selected, groups=grouped, active_page="formulas", title="Формулы"))
|
||||
|
||||
|
||||
@router.post("/service/coefficients/test/{formula_name}")
|
||||
async def formula_test_calc(request: Request, formula_name: str, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
form = await request.form()
|
||||
@@ -760,7 +812,7 @@ async def passports_list(request: Request, db: AsyncSession = Depends(get_db), u
|
||||
# ===================== CHECKLIST =====================
|
||||
|
||||
@router.get("/service/checklist")
|
||||
async def checklist_list(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("technician"))):
|
||||
async def checklist_list(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("technician", "owner"))):
|
||||
result = await db.execute(
|
||||
select(Task).options(joinedload(Task.object)).where(
|
||||
and_(Task.assigned_to == user["id"], Task.status.in_(["open","in_progress"]))
|
||||
@@ -769,7 +821,7 @@ async def checklist_list(request: Request, db: AsyncSession = Depends(get_db), u
|
||||
return templates.TemplateResponse("pages/checklist.html", ctx(request, user=user, tasks=result.unique().scalars().all(), active_page="checklist", title="Чек-лист"))
|
||||
|
||||
@router.post("/service/checklist/complete")
|
||||
async def checklist_complete(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("technician"))):
|
||||
async def checklist_complete(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("technician", "owner"))):
|
||||
form = await request.form()
|
||||
t = await db.get(Task, int(form.get("task_id")))
|
||||
if t and t.assigned_to == user["id"]:
|
||||
@@ -1078,4 +1130,263 @@ async def cases_delete(request: Request, db: AsyncSession = Depends(get_db), use
|
||||
return RedirectResponse(url="/service/cases", status_code=302)
|
||||
|
||||
|
||||
# ===================== BOT SETTINGS =====================
|
||||
|
||||
@router.get("/service/bot-settings")
|
||||
async def bot_settings(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_settings.html", ctx(request, user=user, active_page="bot-settings", title="Настройки бота"))
|
||||
|
||||
@router.get("/service/bot-settings/responses")
|
||||
async def bot_responses(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_responses.html", ctx(request, user=user, active_page="bot-responses", title="Шаблоны ответов бота"))
|
||||
|
||||
@router.get("/service/bot-settings/features")
|
||||
async def bot_features(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_features.html", ctx(request, user=user, active_page="bot-features", title="Функции бота"))
|
||||
|
||||
@router.get("/service/bot-settings/categories")
|
||||
async def bot_categories(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_categories.html", ctx(request, user=user, active_page="bot-categories", title="Категории обращений"))
|
||||
|
||||
@router.get("/service/bot-settings/kb")
|
||||
async def bot_kb(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_kb.html", ctx(request, user=user, active_page="bot-kb", title="База знаний бота"))
|
||||
|
||||
@router.get("/service/bot-settings/conversations")
|
||||
async def bot_conversations(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_conversations.html", ctx(request, user=user, active_page="bot-conversations", title="Диалоги бота"))
|
||||
|
||||
@router.get("/service/bot-settings/users")
|
||||
async def bot_users(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_users.html", ctx(request, user=user, active_page="bot-users", title="Пользователи бота"))
|
||||
|
||||
@router.get("/service/bot-settings/test-runner")
|
||||
async def bot_test_runner(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_test_runner.html", ctx(request, user=user, active_page="bot-test-runner", title="Тестовый прогон бота"))
|
||||
|
||||
@router.get("/service/bot-settings/analytics")
|
||||
async def bot_analytics(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/bot_analytics.html", ctx(request, user=user, active_page="bot-analytics", title="Аналитика бота"))
|
||||
|
||||
|
||||
@router.get("/service/portal-settings")
|
||||
async def portal_settings_page(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/portal_settings.html", ctx(request, user=user, active_page="portal-settings", title="Настройка портала"))
|
||||
|
||||
|
||||
@router.get("/service/role-settings")
|
||||
async def role_settings_page(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
return templates.TemplateResponse("pages/role_settings.html", ctx(request, user=user, active_page="role-settings", title="Настройка Ролей"))
|
||||
|
||||
|
||||
# ---- Role permissions API ----
|
||||
@router.get("/service/api/role-permissions")
|
||||
async def api_role_permissions_get(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
from app.models.models import RoleMenuPermission
|
||||
from sqlalchemy import select
|
||||
result = {"engineer": [], "technician": []}
|
||||
rows = (await db.execute(select(RoleMenuPermission))).scalars().all()
|
||||
for row in rows:
|
||||
result.setdefault(row.role, []).append({"menu_key": row.menu_key, "visible": row.visible})
|
||||
return JSONResponse(result)
|
||||
|
||||
|
||||
@router.post("/service/api/role-permissions/save")
|
||||
async def api_role_permissions_save(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
from app.models.models import RoleMenuPermission
|
||||
from sqlalchemy import delete
|
||||
body = await request.json()
|
||||
role = body.get("role")
|
||||
permissions = body.get("permissions", [])
|
||||
if role not in ("engineer", "technician"):
|
||||
return JSONResponse({"ok": False, "error": "Invalid role"}, status_code=400)
|
||||
await db.execute(delete(RoleMenuPermission).where(RoleMenuPermission.role == role))
|
||||
for p in permissions:
|
||||
db.add(RoleMenuPermission(role=role, menu_key=p["menu_key"], visible=p["visible"]))
|
||||
await db.commit()
|
||||
return JSONResponse({"ok": True})
|
||||
|
||||
|
||||
# ===================== ROLE SWITCH =====================
|
||||
|
||||
@router.post("/service/api/role/switch")
|
||||
async def api_role_switch(request: Request, user: dict = Depends(get_current_user)):
|
||||
if not user:
|
||||
return JSONResponse({"ok": False, "error": "Not authenticated"}, status_code=401)
|
||||
body = await request.json()
|
||||
target_role = body.get("role")
|
||||
current_role = user["role"]
|
||||
|
||||
allowed_roles = {"owner": ["owner", "engineer", "technician"], "engineer": ["engineer", "technician"]}
|
||||
if current_role not in allowed_roles or target_role not in allowed_roles.get(current_role, []):
|
||||
return JSONResponse({"ok": False, "error": "Forbidden"}, status_code=403)
|
||||
|
||||
if "original_user" not in request.session:
|
||||
request.session["original_user"] = dict(user)
|
||||
request.session["user"] = {**user, "role": target_role}
|
||||
return JSONResponse({"ok": True, "role": target_role, "original_role": request.session["original_user"]["role"]})
|
||||
|
||||
|
||||
@router.post("/service/api/role/reset")
|
||||
async def api_role_reset(request: Request, user: dict = Depends(get_current_user)):
|
||||
if not user:
|
||||
return JSONResponse({"ok": False, "error": "Not authenticated"}, status_code=401)
|
||||
if "original_user" in request.session:
|
||||
request.session["user"] = request.session["original_user"]
|
||||
del request.session["original_user"]
|
||||
return JSONResponse({"ok": True, "role": request.session["user"]["role"]})
|
||||
|
||||
|
||||
# ===================== BOT API (proxied from max_bot) =====================
|
||||
|
||||
import aiohttp
|
||||
|
||||
MAX_BOT_URL = "http://127.0.0.1:8002"
|
||||
|
||||
@router.get("/service/api/bot/settings")
|
||||
async def api_bot_settings(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/settings") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/settings/save")
|
||||
async def api_bot_settings_save(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/settings/save", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/templates")
|
||||
async def api_bot_templates(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/templates") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/templates/save")
|
||||
async def api_bot_templates_save(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/templates/save", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/features")
|
||||
async def api_bot_features(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/features") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/features/toggle")
|
||||
async def api_bot_features_toggle(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/features/toggle", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/categories")
|
||||
async def api_bot_categories(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/categories") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/categories/create")
|
||||
async def api_bot_categories_create(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/categories/create", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/categories/edit")
|
||||
async def api_bot_categories_edit(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/categories/edit", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/categories/delete")
|
||||
async def api_bot_categories_delete(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/categories/delete", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/kb")
|
||||
async def api_bot_kb(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/kb") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/kb/create")
|
||||
async def api_bot_kb_create(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/kb/create", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/kb/edit")
|
||||
async def api_bot_kb_edit(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/kb/edit", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/kb/delete")
|
||||
async def api_bot_kb_delete(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/kb/delete", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/conversations")
|
||||
async def api_bot_conversations(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/conversations") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/users")
|
||||
async def api_bot_users(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner")),
|
||||
page: int = Query(1), limit: int = Query(20),
|
||||
sort_by: str = Query("created_at"), sort_dir: str = Query("desc"),
|
||||
consent: str = Query(None)):
|
||||
params = f"?page={page}&limit={limit}&sort_by={sort_by}&sort_dir={sort_dir}"
|
||||
if consent:
|
||||
params += f"&consent={consent}"
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/users{params}") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/analytics")
|
||||
async def api_bot_analytics(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/analytics") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/test-run")
|
||||
async def api_bot_test_run(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/test-run", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/broadcast/recipients")
|
||||
async def api_bot_broadcast_recipients(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner")),
|
||||
q: str = Query(None), page: int = Query(1), limit: int = Query(200)):
|
||||
params = f"?page={page}&limit={limit}"
|
||||
if q:
|
||||
params += f"&q={q}"
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/broadcast/recipients{params}") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.get("/service/api/bot/broadcast/history")
|
||||
async def api_bot_broadcast_history(db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner")),
|
||||
page: int = Query(1), limit: int = Query(20)):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(f"{MAX_BOT_URL}/api/bot/broadcast/history?page={page}&limit={limit}") as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
|
||||
@router.post("/service/api/bot/broadcast")
|
||||
async def api_bot_broadcast(request: Request, db: AsyncSession = Depends(get_db), user: dict = Depends(require_role("owner"))):
|
||||
body = await request.json()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(f"{MAX_BOT_URL}/api/bot/broadcast", json=body) as resp:
|
||||
return JSONResponse(await resp.json())
|
||||
@@ -0,0 +1,75 @@
|
||||
#!/usr/bin/env bash
|
||||
# Deploy script for AegisOne Service Portal
|
||||
# Usage: ./deploy.sh [--skip-tests] [patch|minor|major]
|
||||
set -e
|
||||
|
||||
SKIP_TESTS=false
|
||||
BUMP="patch"
|
||||
for arg in "$@"; do
|
||||
case "$arg" in
|
||||
--skip-tests) SKIP_TESTS=true ;;
|
||||
patch|minor|major) BUMP="$arg" ;;
|
||||
esac
|
||||
done
|
||||
|
||||
DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
|
||||
# Gate: run tests before deploy
|
||||
if [ "$SKIP_TESTS" = false ]; then
|
||||
echo ""
|
||||
echo "[CHECK] Running tests..."
|
||||
cd "$DIR"
|
||||
if curl -sf http://localhost:8000/health > /dev/null 2>&1; then
|
||||
if ! pytest -v --tb=short -x; then
|
||||
echo ""
|
||||
echo "❌ Tests failed. Deploy aborted."
|
||||
echo " Fix errors or use --skip-tests to force deploy."
|
||||
exit 1
|
||||
fi
|
||||
echo " ✅ Tests passed"
|
||||
else
|
||||
echo " ⚠️ Server not reachable at localhost:8000"
|
||||
echo " Start the service first, or use --skip-tests"
|
||||
exit 1
|
||||
fi
|
||||
fi
|
||||
|
||||
echo "=== AegisOne Service Portal Deploy ==="
|
||||
echo ""
|
||||
|
||||
# Bump version
|
||||
cd "$DIR"
|
||||
python scripts/bump_version.py "$BUMP" '{"features":["Обновление сервисного портала"],"fixes":["Исправления и улучшения"]}'
|
||||
VERSION=$(cat version.txt)
|
||||
echo "Version: $VERSION"
|
||||
|
||||
# Create archive
|
||||
cd ..
|
||||
ARCHIVE="other/aegisone-py-deploy.tar.gz"
|
||||
tar czf "$ARCHIVE" py_service/ --exclude='py_service/__pycache__' --exclude='py_service/.venv' --exclude='py_service/.git' --exclude='py_service/app/__pycache__' --exclude='py_service/app/**/__pycache__'
|
||||
echo "Archive created: $ARCHIVE"
|
||||
|
||||
# Upload
|
||||
scp "$ARCHIVE" angel@81.177.141.34:/opt/projects/
|
||||
echo "Uploaded to server"
|
||||
|
||||
# Deploy on server
|
||||
ssh angel@81.177.141.34 << 'SSHEOF'
|
||||
set -e
|
||||
cd /opt/projects
|
||||
rm -rf aegisone-py-backup 2>/dev/null || true
|
||||
mv aegisone-py aegisone-py-backup 2>/dev/null || true
|
||||
mkdir -p aegisone-py
|
||||
tar xzf aegisone-py-deploy.tar.gz -C aegisone-py --strip-components=1
|
||||
cd aegisone-py
|
||||
docker compose down
|
||||
docker compose build --no-cache --network host
|
||||
docker compose up -d
|
||||
echo "Service restarted"
|
||||
SSHEOF
|
||||
|
||||
# Commit changelog
|
||||
git add py_service/version.txt py_service/CHANGELOG.md
|
||||
git diff --cached --quiet || git commit -m "v$VERSION: changelog [skip ci]"
|
||||
|
||||
echo "=== Deploy complete: v$VERSION ==="
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Pytest configuration for service portal tests.
|
||||
|
||||
Provides authenticated httpx clients for testing against a running server.
|
||||
Set TEST_SERVER_URL env var to override the default http://localhost:8000.
|
||||
"""
|
||||
|
||||
import os
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
SERVER_URL = os.environ.get("TEST_SERVER_URL", "http://localhost:8000")
|
||||
PASSWORD = "AegisOne2024!"
|
||||
|
||||
|
||||
def _login(role: str) -> httpx.Client:
|
||||
"""Create an authenticated httpx client for the given role."""
|
||||
client = httpx.Client(base_url=SERVER_URL, follow_redirects=False, timeout=10)
|
||||
resp = client.post("/service/login", data={"login": role, "password": PASSWORD})
|
||||
assert resp.status_code == 302, f"{role} login failed: {resp.status_code} {resp.text[:200]}"
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def client():
|
||||
"""Unauthenticated httpx client."""
|
||||
with httpx.Client(base_url=SERVER_URL, follow_redirects=False, timeout=10) as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def owner_client(_check_server):
|
||||
"""Authenticated httpx client as owner."""
|
||||
return _login("owner")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def engineer_client(_check_server):
|
||||
"""Authenticated httpx client as engineer."""
|
||||
return _login("engineer")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def technician_client(_check_server):
|
||||
"""Authenticated httpx client as technician."""
|
||||
return _login("technician")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def api_client(_check_server):
|
||||
"""Authenticated httpx client as owner for API tests."""
|
||||
return _login("owner")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def _check_server():
|
||||
"""Sanity check — server must be reachable before running tests."""
|
||||
with httpx.Client(base_url=SERVER_URL, timeout=5) as c:
|
||||
resp = c.get("/health")
|
||||
if resp.status_code != 200:
|
||||
pytest.fail(f"Server at {SERVER_URL} is not reachable (health: {resp.status_code})")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def role_client(client, owner_client, engineer_client, technician_client):
|
||||
"""Dict mapping role names to authenticated httpx clients."""
|
||||
return {
|
||||
"anonymous": client,
|
||||
"owner": owner_client,
|
||||
"engineer": engineer_client,
|
||||
"technician": technician_client,
|
||||
}
|
||||
@@ -0,0 +1,218 @@
|
||||
"""Central permission configuration for all service routes.
|
||||
|
||||
Every route must be listed here. test_00_discover_routes.py enforces this.
|
||||
|
||||
Format:
|
||||
ROUTE_PERMISSIONS[(method, path)] = (role1, role2, ...)
|
||||
- "anonymous" means no auth required (public)
|
||||
- "authenticated" means any logged-in user
|
||||
|
||||
Adding a new route? Add it here first — the discover test will fail otherwise.
|
||||
"""
|
||||
|
||||
ROLES = ("owner", "engineer", "technician")
|
||||
|
||||
ROUTE_PERMISSIONS: dict[tuple[str, str], tuple[str, ...]] = {
|
||||
# === PUBLIC ===
|
||||
("GET", "/health"): ("anonymous",),
|
||||
("GET", "/service/login"): ("anonymous",),
|
||||
("POST", "/service/login"): ("anonymous",),
|
||||
("GET", "/"): ("anonymous",),
|
||||
|
||||
# === AUTH ===
|
||||
("GET", "/service/logout"): ("authenticated",),
|
||||
|
||||
# === DASHBOARD ===
|
||||
("GET", "/service/dashboard"): ("owner", "engineer", "technician"),
|
||||
|
||||
# === USERS ===
|
||||
("GET", "/service/users"): ("owner",),
|
||||
("POST", "/service/users/create"): ("owner",),
|
||||
("POST", "/service/users/edit"): ("owner",),
|
||||
("POST", "/service/users/delete"): ("owner",),
|
||||
|
||||
# === CUSTOMERS ===
|
||||
("GET", "/service/customers"): ("owner", "engineer"),
|
||||
("POST", "/service/customers/create"): ("owner", "engineer"),
|
||||
("POST", "/service/customers/edit"): ("owner", "engineer"),
|
||||
("POST", "/service/customers/delete"): ("owner",),
|
||||
|
||||
# === OBJECTS ===
|
||||
("GET", "/service/objects"): ("owner", "engineer"),
|
||||
("POST", "/service/objects/create"): ("owner", "engineer"),
|
||||
("POST", "/service/objects/edit"): ("owner", "engineer"),
|
||||
("POST", "/service/objects/delete"): ("owner",),
|
||||
|
||||
# === ASSIGNMENTS ===
|
||||
("GET", "/service/assignments"): ("owner",),
|
||||
("POST", "/service/assignments/create"): ("owner",),
|
||||
("POST", "/service/assignments/unassign"): ("owner",),
|
||||
|
||||
# === SLA ===
|
||||
("GET", "/service/sla"): ("owner", "engineer"),
|
||||
("POST", "/service/sla/create"): ("owner", "engineer"),
|
||||
("POST", "/service/sla/edit"): ("owner", "engineer"),
|
||||
|
||||
# === CEO ===
|
||||
("GET", "/service/ceo"): ("owner",),
|
||||
("POST", "/service/ceo/calc"): ("owner",),
|
||||
("POST", "/service/ceo/kpi"): ("owner",),
|
||||
|
||||
# === COEFFICIENTS ===
|
||||
("GET", "/service/coefficients"): ("owner",),
|
||||
("POST", "/service/coefficients/edit"): ("owner",),
|
||||
("POST", "/service/coefficients/toggle"): ("owner",),
|
||||
("POST", "/service/coefficients/create"): ("owner",),
|
||||
("POST", "/service/coefficients/delete"): ("owner",),
|
||||
("GET", "/service/coefficients/test"): ("owner",),
|
||||
("POST", "/service/coefficients/test/{formula_name}"): ("owner",),
|
||||
|
||||
# === FORMULAS ===
|
||||
("GET", "/service/formulas"): ("owner",),
|
||||
|
||||
# === TASKS ===
|
||||
("GET", "/service/tasks"): ("owner", "engineer", "technician"),
|
||||
("POST", "/service/tasks/create"): ("owner", "engineer"),
|
||||
("POST", "/service/tasks/close"): ("owner", "engineer", "technician"),
|
||||
|
||||
# === REPORTS ===
|
||||
("GET", "/service/reports"): ("owner", "engineer", "technician"),
|
||||
("POST", "/service/reports/create"): ("owner", "engineer", "technician"),
|
||||
|
||||
# === INCIDENTS ===
|
||||
("GET", "/service/incidents"): ("owner", "engineer", "technician"),
|
||||
("POST", "/service/incidents/create"): ("owner", "engineer"),
|
||||
("POST", "/service/incidents/resolve"): ("owner", "engineer", "technician"),
|
||||
|
||||
# === QUESTIONNAIRE ===
|
||||
("GET", "/service/questionnaire"): ("owner", "engineer"),
|
||||
("GET", "/service/questionnaire/step/{step}"): ("owner", "engineer"),
|
||||
("POST", "/service/questionnaire/save/{step}"): ("owner", "engineer"),
|
||||
|
||||
# === QUESTIONNAIRE CONFIG ===
|
||||
("GET", "/service/questionnaire-config"): ("owner",),
|
||||
("POST", "/service/questionnaire-config/create"): ("owner",),
|
||||
("POST", "/service/questionnaire-config/edit"): ("owner",),
|
||||
("POST", "/service/questionnaire-config/toggle"): ("owner",),
|
||||
("POST", "/service/questionnaire-config/delete"): ("owner",),
|
||||
|
||||
# === PASSPORTS ===
|
||||
("GET", "/service/passports"): ("owner", "engineer"),
|
||||
|
||||
# === CHECKLIST ===
|
||||
("GET", "/service/checklist"): ("owner", "technician"),
|
||||
("POST", "/service/checklist/complete"): ("owner", "technician"),
|
||||
|
||||
# === BLOG ===
|
||||
("GET", "/service/blog"): ("owner",),
|
||||
("POST", "/service/blog/create"): ("owner",),
|
||||
("POST", "/service/blog/edit"): ("owner",),
|
||||
("POST", "/service/blog/delete"): ("owner",),
|
||||
|
||||
# === QUICK MENU ===
|
||||
("GET", "/service/quick-menu"): ("owner", "engineer"),
|
||||
("POST", "/service/quick-menu/save"): ("owner", "engineer"),
|
||||
("GET", "/service/api/quick-menu"): ("authenticated",),
|
||||
("GET", "/service/quick-menu/tech"): ("engineer",),
|
||||
("POST", "/service/quick-menu/tech-save"): ("engineer",),
|
||||
|
||||
# === IDEAS ===
|
||||
("GET", "/service/ideas"): ("owner",),
|
||||
("POST", "/service/ideas/create"): ("owner",),
|
||||
("POST", "/service/ideas/edit"): ("owner",),
|
||||
("POST", "/service/ideas/status"): ("owner",),
|
||||
("POST", "/service/ideas/delete"): ("owner",),
|
||||
|
||||
# === CHARTS ===
|
||||
("GET", "/service/charts"): ("owner",),
|
||||
("GET", "/service/api/charts/data"): ("owner",),
|
||||
|
||||
# === CASES ===
|
||||
("GET", "/service/cases"): ("owner",),
|
||||
("POST", "/service/cases/create"): ("owner",),
|
||||
("POST", "/service/cases/edit"): ("owner",),
|
||||
("POST", "/service/cases/delete"): ("owner",),
|
||||
|
||||
# === BOT SETTINGS (pages) ===
|
||||
("GET", "/service/bot-settings"): ("owner",),
|
||||
("GET", "/service/bot-settings/responses"): ("owner",),
|
||||
("GET", "/service/bot-settings/features"): ("owner",),
|
||||
("GET", "/service/bot-settings/categories"): ("owner",),
|
||||
("GET", "/service/bot-settings/kb"): ("owner",),
|
||||
("GET", "/service/bot-settings/conversations"): ("owner",),
|
||||
("GET", "/service/bot-settings/users"): ("owner",),
|
||||
("GET", "/service/bot-settings/test-runner"): ("owner",),
|
||||
("GET", "/service/bot-settings/analytics"): ("owner",),
|
||||
|
||||
# === PORTAL / ROLE SETTINGS ===
|
||||
("GET", "/service/portal-settings"): ("owner",),
|
||||
("GET", "/service/role-settings"): ("owner",),
|
||||
|
||||
# === ROLE PERMISSIONS API ===
|
||||
("GET", "/service/api/role-permissions"): ("owner",),
|
||||
("POST", "/service/api/role-permissions/save"): ("owner",),
|
||||
|
||||
# === ROLE SWITCH API ===
|
||||
("POST", "/service/api/role/switch"): ("owner", "engineer"),
|
||||
("POST", "/service/api/role/reset"): ("authenticated",),
|
||||
|
||||
# === BOT API (proxied) ===
|
||||
("GET", "/service/api/bot/settings"): ("owner",),
|
||||
("POST", "/service/api/bot/settings/save"): ("owner",),
|
||||
("GET", "/service/api/bot/templates"): ("owner",),
|
||||
("POST", "/service/api/bot/templates/save"): ("owner",),
|
||||
("GET", "/service/api/bot/features"): ("owner",),
|
||||
("POST", "/service/api/bot/features/toggle"): ("owner",),
|
||||
("GET", "/service/api/bot/categories"): ("owner",),
|
||||
("POST", "/service/api/bot/categories/create"): ("owner",),
|
||||
("POST", "/service/api/bot/categories/edit"): ("owner",),
|
||||
("POST", "/service/api/bot/categories/delete"): ("owner",),
|
||||
("GET", "/service/api/bot/kb"): ("owner",),
|
||||
("POST", "/service/api/bot/kb/create"): ("owner",),
|
||||
("POST", "/service/api/bot/kb/edit"): ("owner",),
|
||||
("POST", "/service/api/bot/kb/delete"): ("owner",),
|
||||
("GET", "/service/api/bot/conversations"): ("owner",),
|
||||
("GET", "/service/api/bot/users"): ("owner",),
|
||||
("GET", "/service/api/bot/analytics"): ("owner",),
|
||||
("POST", "/service/api/bot/test-run"): ("owner",),
|
||||
("GET", "/service/api/bot/broadcast/recipients"): ("owner",),
|
||||
("GET", "/service/api/bot/broadcast/history"): ("owner",),
|
||||
("POST", "/service/api/bot/broadcast"): ("owner",),
|
||||
|
||||
# === OTHER API (main.py) ===
|
||||
("GET", "/service/api/shs"): ("owner", "engineer", "technician"),
|
||||
("GET", "/service/api/tasks"): ("owner", "engineer", "technician"),
|
||||
("GET", "/service/api/changelog"): ("authenticated",),
|
||||
|
||||
# === DOCUMENTS ===
|
||||
("GET", "/service/documents/"): ("owner", "engineer", "technician"),
|
||||
("GET", "/service/documents/admin/permissions"): ("owner",),
|
||||
("POST", "/service/documents/admin/permissions"): ("owner",),
|
||||
("POST", "/service/documents/admin/create"): ("owner",),
|
||||
("GET", "/service/documents/tech-access"): ("engineer",),
|
||||
("POST", "/service/documents/tech-access"): ("engineer",),
|
||||
("GET", "/service/documents/{slug}"): ("owner", "engineer", "technician"),
|
||||
("GET", "/service/documents/{slug}/edit"): ("owner", "engineer"),
|
||||
("POST", "/service/documents/{slug}/edit"): ("owner", "engineer"),
|
||||
("GET", "/service/documents/{slug}/download"): ("owner",),
|
||||
}
|
||||
|
||||
# Routes with path parameters — skipped in strict permission test
|
||||
# but still verified to exist in discover test.
|
||||
DYNAMIC_ROUTES: set[tuple[str, str]] = {
|
||||
("POST", "/service/coefficients/test/{formula_name}"),
|
||||
("GET", "/service/questionnaire/step/{step}"),
|
||||
("POST", "/service/questionnaire/save/{step}"),
|
||||
("GET", "/service/documents/{slug}"),
|
||||
("GET", "/service/documents/{slug}/edit"),
|
||||
("POST", "/service/documents/{slug}/edit"),
|
||||
("GET", "/service/documents/{slug}/download"),
|
||||
}
|
||||
|
||||
# Routes excluded from discover test (internal framework routes)
|
||||
EXCLUDED_ROUTES: set[tuple[str, str]] = {
|
||||
("GET", "/docs"),
|
||||
("GET", "/docs/oauth2-redirect"),
|
||||
("GET", "/openapi.json"),
|
||||
("GET", "/redoc"),
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
"""Auto-discover all routes and verify they're covered in permission_config.
|
||||
|
||||
This test runs without a server — it inspects the FastAPI app directly.
|
||||
When a new route is added, this test will FAIL until it's added to
|
||||
permission_config.ROUTE_PERMISSIONS.
|
||||
"""
|
||||
|
||||
from tests.permission_config import ROUTE_PERMISSIONS, DYNAMIC_ROUTES, EXCLUDED_ROUTES
|
||||
|
||||
|
||||
def collect_app_routes():
|
||||
"""Return set of (method, path) from the main app and all routers."""
|
||||
from app.main import app
|
||||
routes = set()
|
||||
for route in app.routes:
|
||||
if hasattr(route, "methods") and hasattr(route, "path"):
|
||||
for method in route.methods:
|
||||
method = method.upper()
|
||||
if method in ("GET", "POST", "PUT", "DELETE", "PATCH"):
|
||||
routes.add((method, route.path))
|
||||
return routes
|
||||
|
||||
|
||||
def test_all_routes_covered():
|
||||
"""Every app route must have a permission entry."""
|
||||
app_routes = collect_app_routes() - EXCLUDED_ROUTES
|
||||
config_routes = set(ROUTE_PERMISSIONS.keys())
|
||||
missing = app_routes - config_routes
|
||||
assert not missing, (
|
||||
f"Routes missing from permission_config.ROUTE_PERMISSIONS:\n"
|
||||
+ "\n".join(f" {m} {p}" for m, p in sorted(missing))
|
||||
+ "\n\nAdd them to py_service/tests/permission_config.py"
|
||||
)
|
||||
|
||||
|
||||
def test_no_stale_permissions():
|
||||
"""No stale entries in permission config."""
|
||||
app_routes = collect_app_routes() - EXCLUDED_ROUTES
|
||||
config_routes = set(ROUTE_PERMISSIONS.keys())
|
||||
stale = config_routes - app_routes
|
||||
assert not stale, (
|
||||
f"ROUTE_PERMISSIONS has stale entries (route no longer exists):\n"
|
||||
+ "\n".join(f" {m} {p}" for m, p in sorted(stale))
|
||||
)
|
||||
|
||||
|
||||
def test_dynamic_routes_also_covered():
|
||||
"""All DYNAMIC_ROUTES must also be in ROUTE_PERMISSIONS."""
|
||||
for entry in DYNAMIC_ROUTES:
|
||||
assert entry in ROUTE_PERMISSIONS, (
|
||||
f"{entry[0]} {entry[1]} is in DYNAMIC_ROUTES but missing from ROUTE_PERMISSIONS"
|
||||
)
|
||||
|
||||
|
||||
def test_all_roles_valid():
|
||||
"""Allowed roles must be one of: owner, engineer, technician, anonymous, authenticated."""
|
||||
valid_roles = {"owner", "engineer", "technician", "anonymous", "authenticated"}
|
||||
for (method, path), allowed_roles in ROUTE_PERMISSIONS.items():
|
||||
for role in allowed_roles:
|
||||
assert role in valid_roles, (
|
||||
f"{method} {path}: invalid role '{role}' (valid: {valid_roles})"
|
||||
)
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Strict permission tests — every role × every route with exact expected codes.
|
||||
|
||||
ALLOWED → status must be 200 or 302 (never 403/404/500)
|
||||
DENIED → status must be 403 (strict — not 302!)
|
||||
ANONYMOUS → protected routes get 302 (redirect to login)
|
||||
→ public routes get 200
|
||||
|
||||
Dynamic routes (with path params like {slug}) are skipped here.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from tests.permission_config import ROUTE_PERMISSIONS, DYNAMIC_ROUTES, ROLES
|
||||
|
||||
|
||||
def _generate_test_cases():
|
||||
"""Yield (role, method, path, expected) tuples."""
|
||||
for (method, path), allowed_roles in ROUTE_PERMISSIONS.items():
|
||||
if (method, path) in DYNAMIC_ROUTES:
|
||||
continue
|
||||
for role in ROLES:
|
||||
if role in allowed_roles or "authenticated" in allowed_roles:
|
||||
yield (role, method, path, "allow")
|
||||
else:
|
||||
yield (role, method, path, "deny")
|
||||
|
||||
# Anonymous
|
||||
if "anonymous" in allowed_roles:
|
||||
yield ("anonymous", method, path, "allow")
|
||||
else:
|
||||
yield ("anonymous", method, path, "deny")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("role,method,path,expected", list(_generate_test_cases()))
|
||||
def test_access(role_client, role, method, path, expected):
|
||||
client = role_client[role]
|
||||
resp = client.request(method, path)
|
||||
|
||||
if expected == "allow":
|
||||
assert resp.status_code not in (403, 404), (
|
||||
f"[{role}] {method} {path}: got {resp.status_code}, expected non-403/404"
|
||||
)
|
||||
if resp.status_code == 500:
|
||||
pytest.fail(f"[{role}] {method} {path}: 500 Internal Server Error")
|
||||
else:
|
||||
if role == "anonymous":
|
||||
assert resp.status_code in (302, 403), (
|
||||
f"[{role}] {method} {path}: got {resp.status_code}, expected 302/403"
|
||||
)
|
||||
else:
|
||||
assert resp.status_code == 403, (
|
||||
f"[{role}] {method} {path}: got {resp.status_code}, expected 403"
|
||||
)
|
||||
@@ -0,0 +1,142 @@
|
||||
"""Test role switching lifecycle — owner→engineer→reset, etc.
|
||||
|
||||
These tests verify that the session remains consistent after role switches.
|
||||
The critical check: after a reset, the original role's access is fully restored.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
from tests.conftest import SERVER_URL, PASSWORD
|
||||
|
||||
|
||||
class TestRoleSwitch:
|
||||
|
||||
def _login(self, role: str) -> httpx.Client:
|
||||
"""Create a fresh authenticated client."""
|
||||
c = httpx.Client(base_url=SERVER_URL, follow_redirects=False, timeout=10)
|
||||
resp = c.post("/service/login", data={"login": role, "password": PASSWORD})
|
||||
assert resp.status_code == 302
|
||||
return c
|
||||
|
||||
def _switch_role(self, client: httpx.Client, target: str):
|
||||
resp = client.post("/service/api/role/switch", json={"role": target})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["ok"] is True
|
||||
return data
|
||||
|
||||
def _reset_role(self, client: httpx.Client):
|
||||
resp = client.post("/service/api/role/reset")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["ok"] is True
|
||||
return data
|
||||
|
||||
def test_owner_to_engineer_and_back(self):
|
||||
"""Owner switches to engineer, then resets — full access restored."""
|
||||
client = self._login("owner")
|
||||
|
||||
# Owner can access owner pages
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 200
|
||||
|
||||
# Switch to engineer
|
||||
data = self._switch_role(client, "engineer")
|
||||
assert data["role"] == "engineer"
|
||||
assert data["original_role"] == "owner"
|
||||
|
||||
# Engineer cannot access owner pages
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 403, "Engineer should be denied owner pages"
|
||||
|
||||
# Engineer CAN access shared pages
|
||||
r = client.get("/service/customers")
|
||||
assert r.status_code == 200
|
||||
|
||||
# Reset to owner
|
||||
data = self._reset_role(client)
|
||||
assert data["role"] == "owner"
|
||||
|
||||
# CRITICAL: Owner access restored
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 200, "Owner access not restored after reset!"
|
||||
|
||||
# Dashboard still works
|
||||
r = client.get("/service/dashboard")
|
||||
assert r.status_code == 200
|
||||
|
||||
def test_owner_to_technician_and_back(self):
|
||||
"""Owner switches directly to technician."""
|
||||
client = self._login("owner")
|
||||
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 200
|
||||
|
||||
self._switch_role(client, "technician")
|
||||
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 403
|
||||
|
||||
r = client.get("/service/tasks")
|
||||
assert r.status_code == 200
|
||||
|
||||
self._reset_role(client)
|
||||
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 200
|
||||
|
||||
def test_engineer_to_technician_and_back(self):
|
||||
"""Engineer switches to technician, then resets."""
|
||||
client = self._login("engineer")
|
||||
|
||||
# Engineer can access engineer pages
|
||||
r = client.get("/service/customers")
|
||||
assert r.status_code == 200
|
||||
|
||||
self._switch_role(client, "technician")
|
||||
|
||||
# Technician cannot access engineer pages
|
||||
r = client.get("/service/customers")
|
||||
assert r.status_code == 403
|
||||
|
||||
r = client.get("/service/checklist")
|
||||
assert r.status_code == 200
|
||||
|
||||
self._reset_role(client)
|
||||
|
||||
# Engineer access restored
|
||||
r = client.get("/service/customers")
|
||||
assert r.status_code == 200
|
||||
|
||||
def test_double_switch_owner_engineer_technician_then_reset(self):
|
||||
"""Multi-step: owner→engineer→technician→reset→owner."""
|
||||
client = self._login("owner")
|
||||
|
||||
self._switch_role(client, "engineer")
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 403
|
||||
|
||||
# Switch deeper while already impersonating
|
||||
self._switch_role(client, "technician")
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 403
|
||||
r = client.get("/service/tasks")
|
||||
assert r.status_code == 200
|
||||
|
||||
# Single reset should restore owner
|
||||
self._reset_role(client)
|
||||
r = client.get("/service/users")
|
||||
assert r.status_code == 200
|
||||
|
||||
def test_forbidden_switch_technician_cannot_switch(self):
|
||||
"""Technician cannot use role switch (not in allowed_roles)."""
|
||||
client = self._login("technician")
|
||||
resp = client.post("/service/api/role/switch", json={"role": "owner"})
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_reset_without_switch_does_not_error(self):
|
||||
"""Reset without an active override should still work."""
|
||||
client = self._login("owner")
|
||||
resp = client.post("/service/api/role/reset")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["role"] == "owner"
|
||||
@@ -0,0 +1,175 @@
|
||||
"""CRUD operation tests — create, edit, delete for all entities.
|
||||
|
||||
Each test creates a unique entity, verifies the operation returns a redirect,
|
||||
and cleans up where possible. No HTML parsing — just HTTP status checks.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import time
|
||||
|
||||
|
||||
class TestUserCRUD:
|
||||
ID = None
|
||||
|
||||
def test_create_user(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
resp = owner_client.post("/service/users/create", data={
|
||||
"login": f"test_user_{ts}",
|
||||
"password": "TestPass123!",
|
||||
"full_name": f"Test User {ts}",
|
||||
"role": "technician",
|
||||
"phone": "+7-123-456-7890",
|
||||
"email": f"test_{ts}@example.com",
|
||||
"is_active": "1",
|
||||
})
|
||||
assert resp.status_code == 302, f"Create user failed: {resp.status_code}"
|
||||
# Follow redirect should succeed
|
||||
assert resp.headers.get("location") == "/service/users"
|
||||
|
||||
def test_edit_user(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
# First create a user to edit
|
||||
resp = owner_client.post("/service/users/create", data={
|
||||
"login": f"edit_user_{ts}",
|
||||
"password": "TestPass123!",
|
||||
"full_name": f"Edit User {ts}",
|
||||
"role": "technician",
|
||||
"is_active": "1",
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
location = resp.headers.get("location", "")
|
||||
# Extract ID from redirect or use default — the redirect goes to /service/users
|
||||
# We need the user ID. We'll need to find it from the list page.
|
||||
# For now, edit by fetching the users page and extracting an ID
|
||||
resp = owner_client.get("/service/users")
|
||||
assert resp.status_code == 200
|
||||
body = resp.text
|
||||
import re
|
||||
# Find our user in the HTML to get their ID
|
||||
# Pattern: <tr data-id="(\d+)">.*edit_user_{ts}
|
||||
match = re.search(rf"data-id=['\"](\d+)['\"][^>]*>.*?edit_user_{ts}", body, re.DOTALL)
|
||||
if not match:
|
||||
# Try alternative pattern: look for the login in a more flexible way
|
||||
match = re.search(rf"edit_user_{ts}.*?data-id=['\"](\d+)['\"]", body, re.DOTALL)
|
||||
if match:
|
||||
user_id = match.group(1)
|
||||
else:
|
||||
# Fallback: find any user ID with role technician
|
||||
match = re.search(r"data-id=['\"](\d+)['\"]", body)
|
||||
if not match:
|
||||
pytest.skip("Could not find a user to edit")
|
||||
user_id = match.group(1)
|
||||
|
||||
resp = owner_client.post("/service/users/edit", data={
|
||||
"id": user_id,
|
||||
"full_name": f"Updated User {ts}",
|
||||
"role": "technician",
|
||||
"phone": "+7-999-888-7766",
|
||||
"email": f"updated_{ts}@example.com",
|
||||
"is_active": "1",
|
||||
})
|
||||
assert resp.status_code == 302, f"Edit user returned {resp.status_code}"
|
||||
|
||||
def test_delete_user(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
# Create user to delete
|
||||
resp = owner_client.post("/service/users/create", data={
|
||||
"login": f"del_user_{ts}",
|
||||
"password": "TestPass123!",
|
||||
"full_name": f"Delete User {ts}",
|
||||
"role": "technician",
|
||||
"is_active": "1",
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
|
||||
# Find user ID
|
||||
resp = owner_client.get("/service/users")
|
||||
body = resp.text
|
||||
import re
|
||||
match = re.search(rf"data-id=['\"](\d+)['\"][^>]*>.*?del_user_{ts}", body, re.DOTALL)
|
||||
if not match:
|
||||
match = re.search(rf"del_user_{ts}.*?data-id=['\"](\d+)['\"]", body, re.DOTALL)
|
||||
if match:
|
||||
user_id = match.group(1)
|
||||
resp = owner_client.post("/service/users/delete", data={"id": user_id})
|
||||
assert resp.status_code == 302
|
||||
|
||||
|
||||
class TestCustomerCRUD:
|
||||
|
||||
def test_create_customer(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
inn = str(1000000000 + (ts % 8999999999))
|
||||
resp = owner_client.post("/service/customers/create", data={
|
||||
"name": f"Test Customer {ts}",
|
||||
"inn": inn[:10],
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
|
||||
def test_edit_customer(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
inn = str(1000000000 + (ts % 8999999999))
|
||||
resp = owner_client.post("/service/customers/create", data={
|
||||
"name": f"Edit Customer {ts}",
|
||||
"inn": inn[:10],
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
|
||||
resp = owner_client.get("/service/customers")
|
||||
assert resp.status_code == 200
|
||||
body = resp.text
|
||||
import re
|
||||
match = re.search(rf"data-id=['\"](\d+)['\"]", body)
|
||||
if match:
|
||||
cid = match.group(1)
|
||||
resp = owner_client.post("/service/customers/edit", data={
|
||||
"id": cid,
|
||||
"name": f"Updated Customer {ts}",
|
||||
"inn": inn[:10],
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
|
||||
|
||||
class TestBlogCRUD:
|
||||
|
||||
def test_create_blog_post(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
resp = owner_client.post("/service/blog/create", data={
|
||||
"title": f"Test Blog Post {ts}",
|
||||
"content": f"Content for test blog post {ts}",
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
|
||||
def test_edit_blog_post(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
resp = owner_client.post("/service/blog/create", data={
|
||||
"title": f"Edit Blog Post {ts}",
|
||||
"content": f"Content {ts}",
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
|
||||
resp = owner_client.get("/service/blog")
|
||||
assert resp.status_code == 200
|
||||
body = resp.text
|
||||
import re
|
||||
match = re.search(rf"Edit Blog Post {ts}.*?data-id=['\"](\d+)['\"]", body, re.DOTALL)
|
||||
if match:
|
||||
post_id = match.group(1)
|
||||
resp = owner_client.post("/service/blog/edit", data={
|
||||
"id": post_id,
|
||||
"title": f"Updated Blog Post {ts}",
|
||||
"content": f"Updated content {ts}",
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
|
||||
|
||||
class TestQuestionnaireConfigCRUD:
|
||||
|
||||
def test_create_questionnaire_config(self, owner_client):
|
||||
ts = int(time.time() * 1000)
|
||||
resp = owner_client.post("/service/questionnaire-config/create", data={
|
||||
"key": f"test_key_{ts}",
|
||||
"label": f"Test Label {ts}",
|
||||
"question_text": f"Test question? {ts}",
|
||||
})
|
||||
assert resp.status_code == 302
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Detect 500 Internal Server Errors — test POST routes with edge-case inputs.
|
||||
|
||||
These tests send malformed or empty data to ensure routes handle errors
|
||||
gracefully (redirect or 400/422) instead of crashing with 500.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
MALFORMED_POST_CASES = [
|
||||
# USERS
|
||||
("POST", "/service/users/create", {}),
|
||||
("POST", "/service/users/create", {"login": "no_password"}),
|
||||
("POST", "/service/users/create", {"password": "no_login"}),
|
||||
("POST", "/service/users/edit", {}),
|
||||
("POST", "/service/users/edit", {"id": ""}),
|
||||
("POST", "/service/users/edit", {"id": "not_a_number"}),
|
||||
("POST", "/service/users/delete", {}),
|
||||
("POST", "/service/users/delete", {"id": ""}),
|
||||
|
||||
# CUSTOMERS
|
||||
("POST", "/service/customers/create", {}),
|
||||
("POST", "/service/customers/create", {"name": "no_inn"}),
|
||||
("POST", "/service/customers/create", {"inn": "bad_inn"}),
|
||||
("POST", "/service/customers/edit", {}),
|
||||
("POST", "/service/customers/delete", {}),
|
||||
|
||||
# OBJECTS
|
||||
("POST", "/service/objects/create", {}),
|
||||
("POST", "/service/objects/edit", {}),
|
||||
("POST", "/service/objects/delete", {}),
|
||||
|
||||
# ASSIGNMENTS
|
||||
("POST", "/service/assignments/create", {}),
|
||||
("POST", "/service/assignments/unassign", {}),
|
||||
|
||||
# SLA
|
||||
("POST", "/service/sla/create", {}),
|
||||
("POST", "/service/sla/edit", {}),
|
||||
|
||||
# CEO
|
||||
("POST", "/service/ceo/calc", {}),
|
||||
("POST", "/service/ceo/kpi", {}),
|
||||
|
||||
# COEFFICIENTS
|
||||
("POST", "/service/coefficients/edit", {}),
|
||||
("POST", "/service/coefficients/toggle", {}),
|
||||
("POST", "/service/coefficients/create", {}),
|
||||
("POST", "/service/coefficients/delete", {}),
|
||||
|
||||
# TASKS
|
||||
("POST", "/service/tasks/create", {}),
|
||||
("POST", "/service/tasks/close", {}),
|
||||
|
||||
# REPORTS
|
||||
("POST", "/service/reports/create", {}),
|
||||
|
||||
# INCIDENTS
|
||||
("POST", "/service/incidents/create", {}),
|
||||
("POST", "/service/incidents/resolve", {}),
|
||||
|
||||
# BLOG
|
||||
("POST", "/service/blog/create", {}),
|
||||
("POST", "/service/blog/edit", {}),
|
||||
("POST", "/service/blog/delete", {}),
|
||||
|
||||
# IDEAS
|
||||
("POST", "/service/ideas/create", {}),
|
||||
("POST", "/service/ideas/edit", {}),
|
||||
("POST", "/service/ideas/status", {}),
|
||||
("POST", "/service/ideas/delete", {}),
|
||||
|
||||
# CASES
|
||||
("POST", "/service/cases/create", {}),
|
||||
("POST", "/service/cases/edit", {}),
|
||||
("POST", "/service/cases/delete", {}),
|
||||
|
||||
# QUESTIONNAIRE CONFIG
|
||||
("POST", "/service/questionnaire-config/create", {}),
|
||||
("POST", "/service/questionnaire-config/edit", {}),
|
||||
("POST", "/service/questionnaire-config/toggle", {}),
|
||||
("POST", "/service/questionnaire-config/delete", {}),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method,path,data", MALFORMED_POST_CASES)
|
||||
def test_no_500_on_malformed_input(owner_client, method, path, data):
|
||||
"""Malformed input should NOT cause a 500."""
|
||||
resp = owner_client.request(method, path, data=data)
|
||||
assert resp.status_code != 500, (
|
||||
f"{method} {path} with {data}: got 500 Internal Server Error"
|
||||
)
|
||||
@@ -0,0 +1,258 @@
|
||||
"""Test that all service portal routes return appropriate status codes.
|
||||
This helps catch 404 (missing routes), 500 (server errors), and auth issues.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# Routes accessible without authentication (should return 200 or 302 redirect)
|
||||
PUBLIC_ROUTES = [
|
||||
("GET", "/service/login"),
|
||||
("GET", "/health"),
|
||||
]
|
||||
|
||||
# All protected GET routes (require auth)
|
||||
PROTECTED_GET_ROUTES = [
|
||||
"/service/dashboard",
|
||||
"/service/users",
|
||||
"/service/customers",
|
||||
"/service/objects",
|
||||
"/service/assignments",
|
||||
"/service/sla",
|
||||
"/service/ceo",
|
||||
"/service/coefficients",
|
||||
"/service/coefficients/test",
|
||||
"/service/formulas",
|
||||
"/service/tasks",
|
||||
"/service/reports",
|
||||
"/service/incidents",
|
||||
"/service/questionnaire",
|
||||
"/service/questionnaire-config",
|
||||
"/service/passports",
|
||||
"/service/checklist",
|
||||
"/service/blog",
|
||||
"/service/cases",
|
||||
"/service/ideas",
|
||||
"/service/charts",
|
||||
"/service/portal-settings",
|
||||
"/service/role-settings",
|
||||
"/service/bot-settings",
|
||||
"/service/bot-settings/responses",
|
||||
"/service/bot-settings/features",
|
||||
"/service/bot-settings/categories",
|
||||
"/service/bot-settings/kb",
|
||||
"/service/bot-settings/conversations",
|
||||
"/service/bot-settings/users",
|
||||
"/service/bot-settings/test-runner",
|
||||
"/service/bot-settings/analytics",
|
||||
"/service/quick-menu",
|
||||
"/service/documents/",
|
||||
]
|
||||
|
||||
API_GET_ROUTES = [
|
||||
"/service/api/shs",
|
||||
"/service/api/tasks",
|
||||
"/service/api/changelog",
|
||||
"/service/api/quick-menu",
|
||||
"/service/api/role-permissions",
|
||||
"/service/api/charts/data",
|
||||
"/service/api/bot/settings",
|
||||
"/service/api/bot/templates",
|
||||
"/service/api/bot/features",
|
||||
"/service/api/bot/categories",
|
||||
"/service/api/bot/kb",
|
||||
"/service/api/bot/conversations",
|
||||
"/service/api/bot/users",
|
||||
"/service/api/bot/analytics",
|
||||
"/service/api/bot/broadcast/history",
|
||||
"/service/api/bot/broadcast/recipients",
|
||||
]
|
||||
|
||||
|
||||
class TestPublicRoutes:
|
||||
def test_login_page(self, client):
|
||||
resp = client.get("/service/login")
|
||||
assert resp.status_code in (200, 302)
|
||||
|
||||
def test_health(self, client):
|
||||
resp = client.get("/health")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_root_redirect(self, client):
|
||||
resp = client.get("/", follow_redirects=False)
|
||||
assert resp.status_code == 302
|
||||
|
||||
|
||||
class TestAuthRedirect:
|
||||
"""Unauthenticated requests should redirect to login."""
|
||||
|
||||
@pytest.mark.parametrize("method,path", PUBLIC_ROUTES)
|
||||
def test_public(self, client, method, path):
|
||||
resp = getattr(client, method.lower())(path, follow_redirects=False)
|
||||
assert resp.status_code in (200, 302), f"{method} {path} returned {resp.status_code}"
|
||||
|
||||
@pytest.mark.parametrize("path", PROTECTED_GET_ROUTES)
|
||||
def test_protected_redirect(self, client, path):
|
||||
resp = client.get(path, follow_redirects=False)
|
||||
assert resp.status_code == 302, f"GET {path} returned {resp.status_code} (expected 302)"
|
||||
|
||||
@pytest.mark.parametrize("path", API_GET_ROUTES)
|
||||
def test_api_no_auth(self, client, path):
|
||||
"""API routes bypass auth middleware, but may fail due to missing DB."""
|
||||
resp = client.get(path)
|
||||
# Should not be 404 — either 200, 500 (DB error), or 403
|
||||
assert resp.status_code != 404, f"GET {path} returned 404 (route missing)"
|
||||
|
||||
|
||||
class TestOwnerAccess:
|
||||
"""Owner should have access to all pages."""
|
||||
|
||||
@pytest.mark.parametrize("path", PROTECTED_GET_ROUTES)
|
||||
def test_all_pages(self, owner_client, path):
|
||||
resp = owner_client.get(path, follow_redirects=False)
|
||||
# 200 = success, 302 = redirect (some pages redirect), 500 = server error
|
||||
assert resp.status_code in (200, 302, 500), f"GET {path} returned {resp.status_code}"
|
||||
if resp.status_code == 500:
|
||||
pytest.fail(f"GET {path} returned 500 Internal Server Error")
|
||||
|
||||
@pytest.mark.parametrize("path", API_GET_ROUTES)
|
||||
def test_api_routes(self, owner_client, path):
|
||||
resp = owner_client.get(path, follow_redirects=False)
|
||||
assert resp.status_code != 404, f"GET {path} returned 404 (route missing)"
|
||||
|
||||
|
||||
class TestEngineerAccess:
|
||||
"""Engineer should have access to permitted pages only."""
|
||||
|
||||
ENGINEER_ALLOWED = [
|
||||
"/service/dashboard",
|
||||
"/service/customers",
|
||||
"/service/objects",
|
||||
"/service/sla",
|
||||
"/service/questionnaire",
|
||||
"/service/passports",
|
||||
"/service/tasks",
|
||||
"/service/reports",
|
||||
"/service/incidents",
|
||||
"/service/documents/",
|
||||
]
|
||||
|
||||
ENGINEER_FORBIDDEN = [
|
||||
"/service/users",
|
||||
"/service/assignments",
|
||||
"/service/ceo",
|
||||
"/service/coefficients",
|
||||
"/service/coefficients/test",
|
||||
"/service/formulas",
|
||||
"/service/questionnaire-config",
|
||||
"/service/blog",
|
||||
"/service/cases",
|
||||
"/service/ideas",
|
||||
"/service/charts",
|
||||
"/service/portal-settings",
|
||||
"/service/role-settings",
|
||||
"/service/bot-settings",
|
||||
"/service/bot-settings/responses",
|
||||
"/service/bot-settings/features",
|
||||
"/service/bot-settings/categories",
|
||||
"/service/bot-settings/kb",
|
||||
"/service/bot-settings/users",
|
||||
"/service/bot-settings/test-runner",
|
||||
"/service/bot-settings/analytics",
|
||||
"/service/bot-settings/conversations",
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize("path", ENGINEER_ALLOWED)
|
||||
def test_allowed(self, engineer_client, path):
|
||||
resp = engineer_client.get(path, follow_redirects=False)
|
||||
assert resp.status_code in (200, 302, 500), f"GET {path} returned {resp.status_code}"
|
||||
if resp.status_code == 500:
|
||||
pytest.fail(f"GET {path} returned 500 Internal Server Error")
|
||||
|
||||
@pytest.mark.parametrize("path", ENGINEER_FORBIDDEN)
|
||||
def test_forbidden(self, engineer_client, path):
|
||||
resp = engineer_client.get(path, follow_redirects=False)
|
||||
assert resp.status_code in (302, 403), f"GET {path} returned {resp.status_code} (expected 302/403)"
|
||||
|
||||
|
||||
class TestTechnicianAccess:
|
||||
"""Technician should have very limited access."""
|
||||
|
||||
TECHNICIAN_ALLOWED = [
|
||||
"/service/dashboard",
|
||||
"/service/tasks",
|
||||
"/service/reports",
|
||||
"/service/incidents",
|
||||
"/service/checklist",
|
||||
"/service/documents/",
|
||||
]
|
||||
|
||||
TECHNICIAN_FORBIDDEN = [
|
||||
"/service/users",
|
||||
"/service/customers",
|
||||
"/service/objects",
|
||||
"/service/assignments",
|
||||
"/service/sla",
|
||||
"/service/ceo",
|
||||
"/service/coefficients",
|
||||
"/service/formulas",
|
||||
"/service/questionnaire",
|
||||
"/service/questionnaire-config",
|
||||
"/service/passports",
|
||||
"/service/blog",
|
||||
"/service/cases",
|
||||
"/service/ideas",
|
||||
"/service/charts",
|
||||
"/service/portal-settings",
|
||||
"/service/role-settings",
|
||||
"/service/bot-settings",
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize("path", TECHNICIAN_ALLOWED)
|
||||
def test_allowed(self, technician_client, path):
|
||||
resp = technician_client.get(path, follow_redirects=False)
|
||||
assert resp.status_code in (200, 302, 500), f"GET {path} returned {resp.status_code}"
|
||||
|
||||
@pytest.mark.parametrize("path", TECHNICIAN_FORBIDDEN)
|
||||
def test_forbidden(self, technician_client, path):
|
||||
resp = technician_client.get(path, follow_redirects=False)
|
||||
assert resp.status_code in (302, 403), f"GET {path} returned {resp.status_code} (expected 302/403)"
|
||||
|
||||
|
||||
class TestSidebarRoutes:
|
||||
"""Every sidebar link should resolve without 404."""
|
||||
|
||||
SIDEBAR_LINKS = [
|
||||
"/service/dashboard",
|
||||
"/service/charts",
|
||||
"/service/ceo",
|
||||
"/service/customers",
|
||||
"/service/objects",
|
||||
"/service/sla",
|
||||
"/service/questionnaire",
|
||||
"/service/passports",
|
||||
"/service/users",
|
||||
"/service/assignments",
|
||||
"/service/documents/",
|
||||
"/service/tasks",
|
||||
"/service/reports",
|
||||
"/service/incidents",
|
||||
"/service/checklist",
|
||||
"/service/documents/tech-access",
|
||||
"/service/bot-settings/analytics",
|
||||
"/service/bot-settings/users",
|
||||
"/service/bot-settings/conversations",
|
||||
"/service/blog",
|
||||
"/service/cases",
|
||||
"/service/questionnaire-config",
|
||||
"/service/formulas",
|
||||
"/service/bot-settings",
|
||||
"/service/portal-settings",
|
||||
"/service/role-settings",
|
||||
"/service/ideas",
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize("path", SIDEBAR_LINKS)
|
||||
def test_sidebar_link(self, owner_client, path):
|
||||
resp = owner_client.get(path, follow_redirects=False)
|
||||
assert resp.status_code != 404, f"Sidebar link {path} returned 404 (missing route)"
|
||||
@@ -0,0 +1,184 @@
|
||||
"""Test that all expected routes are defined in the service_pages router.
|
||||
This catches missing routes (potential 404s) purely from code inspection.
|
||||
No database or server is needed.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from app.routers.service_pages import router
|
||||
|
||||
|
||||
def get_registered_routes():
|
||||
"""Return set of (method, path) tuples from the service_pages router."""
|
||||
routes = set()
|
||||
for route in router.routes:
|
||||
if hasattr(route, "methods") and hasattr(route, "path"):
|
||||
for method in route.methods:
|
||||
routes.add((method, route.path))
|
||||
return routes
|
||||
|
||||
|
||||
# All routes that SHOULD be registered in service_pages.py
|
||||
EXPECTED_ROUTES = [
|
||||
("GET", "/service/users"),
|
||||
("POST", "/service/users/create"),
|
||||
("POST", "/service/users/edit"),
|
||||
("POST", "/service/users/delete"),
|
||||
("GET", "/service/customers"),
|
||||
("POST", "/service/customers/create"),
|
||||
("POST", "/service/customers/edit"),
|
||||
("POST", "/service/customers/delete"),
|
||||
("GET", "/service/objects"),
|
||||
("POST", "/service/objects/create"),
|
||||
("POST", "/service/objects/edit"),
|
||||
("POST", "/service/objects/delete"),
|
||||
("GET", "/service/assignments"),
|
||||
("POST", "/service/assignments/create"),
|
||||
("POST", "/service/assignments/unassign"),
|
||||
("GET", "/service/sla"),
|
||||
("POST", "/service/sla/create"),
|
||||
("POST", "/service/sla/edit"),
|
||||
("GET", "/service/ceo"),
|
||||
("POST", "/service/ceo/calc"),
|
||||
("POST", "/service/ceo/kpi"),
|
||||
("GET", "/service/coefficients"),
|
||||
("POST", "/service/coefficients/edit"),
|
||||
("POST", "/service/coefficients/toggle"),
|
||||
("POST", "/service/coefficients/create"),
|
||||
("POST", "/service/coefficients/delete"),
|
||||
("GET", "/service/coefficients/test"),
|
||||
("GET", "/service/formulas"),
|
||||
("GET", "/service/tasks"),
|
||||
("POST", "/service/tasks/create"),
|
||||
("POST", "/service/tasks/close"),
|
||||
("GET", "/service/reports"),
|
||||
("POST", "/service/reports/create"),
|
||||
("GET", "/service/incidents"),
|
||||
("POST", "/service/incidents/create"),
|
||||
("POST", "/service/incidents/resolve"),
|
||||
("GET", "/service/questionnaire"),
|
||||
("GET", "/service/questionnaire/step/{step}"),
|
||||
("POST", "/service/questionnaire/save/{step}"),
|
||||
("GET", "/service/questionnaire-config"),
|
||||
("POST", "/service/questionnaire-config/create"),
|
||||
("POST", "/service/questionnaire-config/edit"),
|
||||
("POST", "/service/questionnaire-config/toggle"),
|
||||
("POST", "/service/questionnaire-config/delete"),
|
||||
("GET", "/service/passports"),
|
||||
("GET", "/service/checklist"),
|
||||
("POST", "/service/checklist/complete"),
|
||||
("GET", "/service/blog"),
|
||||
("POST", "/service/blog/create"),
|
||||
("POST", "/service/blog/edit"),
|
||||
("POST", "/service/blog/delete"),
|
||||
("GET", "/service/api/quick-menu"),
|
||||
("GET", "/service/quick-menu"),
|
||||
("POST", "/service/quick-menu/save"),
|
||||
("GET", "/service/quick-menu/tech"),
|
||||
("POST", "/service/quick-menu/tech-save"),
|
||||
("GET", "/service/ideas"),
|
||||
("POST", "/service/ideas/create"),
|
||||
("POST", "/service/ideas/edit"),
|
||||
("POST", "/service/ideas/status"),
|
||||
("POST", "/service/ideas/delete"),
|
||||
("GET", "/service/charts"),
|
||||
("GET", "/service/api/charts/data"),
|
||||
("GET", "/service/cases"),
|
||||
("POST", "/service/cases/create"),
|
||||
("POST", "/service/cases/edit"),
|
||||
("POST", "/service/cases/delete"),
|
||||
("GET", "/service/bot-settings"),
|
||||
("GET", "/service/bot-settings/responses"),
|
||||
("GET", "/service/bot-settings/features"),
|
||||
("GET", "/service/bot-settings/categories"),
|
||||
("GET", "/service/bot-settings/kb"),
|
||||
("GET", "/service/bot-settings/conversations"),
|
||||
("GET", "/service/bot-settings/users"),
|
||||
("GET", "/service/bot-settings/test-runner"),
|
||||
("GET", "/service/bot-settings/analytics"),
|
||||
("GET", "/service/portal-settings"),
|
||||
("GET", "/service/role-settings"),
|
||||
("GET", "/service/api/role-permissions"),
|
||||
("POST", "/service/api/role-permissions/save"),
|
||||
("POST", "/service/api/role/switch"),
|
||||
("POST", "/service/api/role/reset"),
|
||||
("GET", "/service/api/bot/settings"),
|
||||
("POST", "/service/api/bot/settings/save"),
|
||||
("GET", "/service/api/bot/templates"),
|
||||
("POST", "/service/api/bot/templates/save"),
|
||||
("GET", "/service/api/bot/features"),
|
||||
("POST", "/service/api/bot/features/toggle"),
|
||||
("GET", "/service/api/bot/categories"),
|
||||
("POST", "/service/api/bot/categories/create"),
|
||||
("POST", "/service/api/bot/categories/edit"),
|
||||
("POST", "/service/api/bot/categories/delete"),
|
||||
("GET", "/service/api/bot/kb"),
|
||||
("POST", "/service/api/bot/kb/create"),
|
||||
("POST", "/service/api/bot/kb/edit"),
|
||||
("POST", "/service/api/bot/kb/delete"),
|
||||
("GET", "/service/api/bot/conversations"),
|
||||
("GET", "/service/api/bot/users"),
|
||||
("GET", "/service/api/bot/analytics"),
|
||||
("POST", "/service/api/bot/test-run"),
|
||||
("GET", "/service/api/bot/broadcast/recipients"),
|
||||
("GET", "/service/api/bot/broadcast/history"),
|
||||
("POST", "/service/api/bot/broadcast"),
|
||||
]
|
||||
|
||||
|
||||
class TestRoutesDefined:
|
||||
def test_routes_exist(self):
|
||||
actual = get_registered_routes()
|
||||
missing = []
|
||||
for method, path in EXPECTED_ROUTES:
|
||||
if (method, path) not in actual:
|
||||
missing.append(f"{method} {path}")
|
||||
assert not missing, f"Missing routes:\n" + "\n".join(missing)
|
||||
|
||||
def test_no_unexpected_missing_links(self):
|
||||
"""Every sidebar link should have a corresponding route."""
|
||||
from app.main import app as main_app
|
||||
# Collect all GET routes from both main app and service_pages router
|
||||
all_destinations = set()
|
||||
for route in list(router.routes) + list(main_app.routes):
|
||||
if hasattr(route, "methods") and hasattr(route, "path") and "GET" in route.methods:
|
||||
all_destinations.add(route.path)
|
||||
sidebar_links = [
|
||||
"/service/dashboard",
|
||||
"/service/charts",
|
||||
"/service/ceo",
|
||||
"/service/customers",
|
||||
"/service/objects",
|
||||
"/service/sla",
|
||||
"/service/questionnaire",
|
||||
"/service/passports",
|
||||
"/service/users",
|
||||
"/service/assignments",
|
||||
"/service/documents/",
|
||||
"/service/tasks",
|
||||
"/service/reports",
|
||||
"/service/incidents",
|
||||
"/service/checklist",
|
||||
"/service/documents/tech-access",
|
||||
"/service/bot-settings/analytics",
|
||||
"/service/bot-settings/users",
|
||||
"/service/bot-settings/conversations",
|
||||
"/service/blog",
|
||||
"/service/cases",
|
||||
"/service/questionnaire-config",
|
||||
"/service/formulas",
|
||||
"/service/bot-settings",
|
||||
"/service/portal-settings",
|
||||
"/service/role-settings",
|
||||
"/service/ideas",
|
||||
]
|
||||
for link in sidebar_links:
|
||||
assert link in all_destinations, f"Sidebar link {link} has no GET route registered"
|
||||
|
||||
|
||||
class TestCoefficientsTestCalc:
|
||||
"""The formula test POST route uses path parameter {formula_name}."""
|
||||
|
||||
def test_formula_test_route_pattern(self):
|
||||
actual = get_registered_routes()
|
||||
assert ("POST", "/service/coefficients/test/{formula_name}") in actual, \
|
||||
"Missing POST route pattern for formula tests"
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Smoke test — проверяет что сервер отвечает на все основные маршруты.
|
||||
Запуск: TEST_SERVER_URL=http://localhost:8000 pytest tests/test_smoke_server.py -v
|
||||
|
||||
Работает против запущенного сервера (локально или удалённо).
|
||||
Не требует БД — просто проверяет что ответ не 404 и не 502.
|
||||
"""
|
||||
|
||||
import os
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
SERVER_URL = os.environ.get("TEST_SERVER_URL", "http://localhost:8000")
|
||||
|
||||
# All GET routes from the sidebar + common pages
|
||||
ROUTES_TO_CHECK = [
|
||||
"/service/dashboard",
|
||||
"/service/charts",
|
||||
"/service/ceo",
|
||||
"/service/customers",
|
||||
"/service/objects",
|
||||
"/service/sla",
|
||||
"/service/questionnaire",
|
||||
"/service/passports",
|
||||
"/service/users",
|
||||
"/service/assignments",
|
||||
"/service/tasks",
|
||||
"/service/reports",
|
||||
"/service/incidents",
|
||||
"/service/checklist",
|
||||
"/service/documents/",
|
||||
"/service/documents/tech-access",
|
||||
"/service/blog",
|
||||
"/service/cases",
|
||||
"/service/ideas",
|
||||
"/service/questionnaire-config",
|
||||
"/service/formulas",
|
||||
"/service/coefficients",
|
||||
"/service/coefficients/test",
|
||||
"/service/bot-settings",
|
||||
"/service/bot-settings/responses",
|
||||
"/service/bot-settings/features",
|
||||
"/service/bot-settings/categories",
|
||||
"/service/bot-settings/kb",
|
||||
"/service/bot-settings/conversations",
|
||||
"/service/bot-settings/users",
|
||||
"/service/bot-settings/test-runner",
|
||||
"/service/bot-settings/analytics",
|
||||
"/service/portal-settings",
|
||||
"/service/role-settings",
|
||||
"/service/quick-menu",
|
||||
]
|
||||
|
||||
# API routes (should work with or without auth)
|
||||
API_ROUTES = [
|
||||
"/service/api/shs",
|
||||
"/service/api/changelog",
|
||||
"/service/api/quick-menu",
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def client():
|
||||
with httpx.Client(base_url=SERVER_URL, follow_redirects=False, timeout=10) as c:
|
||||
yield c
|
||||
|
||||
|
||||
class TestServerReachable:
|
||||
def test_health(self, client):
|
||||
resp = client.get("/health")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_login_page(self, client):
|
||||
resp = client.get("/service/login")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_root_redirect(self, client):
|
||||
resp = client.get("/")
|
||||
assert resp.status_code == 200 or resp.status_code == 302
|
||||
|
||||
|
||||
class TestAuthPagesRedirect:
|
||||
"""Without auth, should redirect to login (302) or show login page (200)."""
|
||||
|
||||
@pytest.mark.parametrize("path", ROUTES_TO_CHECK)
|
||||
def test_redirect_to_login(self, client, path):
|
||||
resp = client.get(path, follow_redirects=False)
|
||||
assert resp.status_code in (200, 302, 403), \
|
||||
f"GET {path}: got {resp.status_code} (expected 302 redirect or 200/403)"
|
||||
|
||||
|
||||
class TestApiRoutes:
|
||||
"""API routes bypass auth middleware — should not 404."""
|
||||
|
||||
@pytest.mark.parametrize("path", API_ROUTES)
|
||||
def test_api_not_404(self, client, path):
|
||||
resp = client.get(path)
|
||||
assert resp.status_code != 404, f"API {path} returned 404 (route missing)"
|
||||
@@ -1 +1 @@
|
||||
1.4.0
|
||||
1.5.4
|
||||
|
||||
Reference in New Issue
Block a user