diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..5e16df4 --- /dev/null +++ b/.github/workflows/tests.yml @@ -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" diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..870d1e6 --- /dev/null +++ b/.pre-commit-config.yaml @@ -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 diff --git a/other/deploy_update.sh b/other/deploy_update.sh new file mode 100644 index 0000000..6230ade --- /dev/null +++ b/other/deploy_update.sh @@ -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 "==========================================" diff --git a/py_service/CHANGELOG.md b/py_service/CHANGELOG.md index 5695508..fd3eebb 100644 --- a/py_service/CHANGELOG.md +++ b/py_service/CHANGELOG.md @@ -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) diff --git a/py_service/app/routers/service_pages.py b/py_service/app/routers/service_pages.py index 2d83e4f..726222f 100644 --- a/py_service/app/routers/service_pages.py +++ b/py_service/app/routers/service_pages.py @@ -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()) \ No newline at end of file diff --git a/py_service/deploy.sh b/py_service/deploy.sh new file mode 100644 index 0000000..36a7c72 --- /dev/null +++ b/py_service/deploy.sh @@ -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 ===" diff --git a/py_service/tests/__init__.py b/py_service/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/py_service/tests/conftest.py b/py_service/tests/conftest.py new file mode 100644 index 0000000..cc44b8b --- /dev/null +++ b/py_service/tests/conftest.py @@ -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, + } diff --git a/py_service/tests/permission_config.py b/py_service/tests/permission_config.py new file mode 100644 index 0000000..167c557 --- /dev/null +++ b/py_service/tests/permission_config.py @@ -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"), +} diff --git a/py_service/tests/test_00_discover_routes.py b/py_service/tests/test_00_discover_routes.py new file mode 100644 index 0000000..8741089 --- /dev/null +++ b/py_service/tests/test_00_discover_routes.py @@ -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})" + ) diff --git a/py_service/tests/test_01_permissions_strict.py b/py_service/tests/test_01_permissions_strict.py new file mode 100644 index 0000000..0bfe6d2 --- /dev/null +++ b/py_service/tests/test_01_permissions_strict.py @@ -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" + ) diff --git a/py_service/tests/test_02_session_role_switch.py b/py_service/tests/test_02_session_role_switch.py new file mode 100644 index 0000000..a452f46 --- /dev/null +++ b/py_service/tests/test_02_session_role_switch.py @@ -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" diff --git a/py_service/tests/test_03_crud_operations.py b/py_service/tests/test_03_crud_operations.py new file mode 100644 index 0000000..47cee83 --- /dev/null +++ b/py_service/tests/test_03_crud_operations.py @@ -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: .*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 diff --git a/py_service/tests/test_04_internal_errors.py b/py_service/tests/test_04_internal_errors.py new file mode 100644 index 0000000..c8dfcca --- /dev/null +++ b/py_service/tests/test_04_internal_errors.py @@ -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" + ) diff --git a/py_service/tests/test_routes.py b/py_service/tests/test_routes.py new file mode 100644 index 0000000..71c433f --- /dev/null +++ b/py_service/tests/test_routes.py @@ -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)" diff --git a/py_service/tests/test_routes_defined.py b/py_service/tests/test_routes_defined.py new file mode 100644 index 0000000..cd4bca7 --- /dev/null +++ b/py_service/tests/test_routes_defined.py @@ -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" diff --git a/py_service/tests/test_smoke_server.py b/py_service/tests/test_smoke_server.py new file mode 100644 index 0000000..c39b846 --- /dev/null +++ b/py_service/tests/test_smoke_server.py @@ -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)" diff --git a/py_service/version.txt b/py_service/version.txt index 88c5fb8..94fe62c 100644 --- a/py_service/version.txt +++ b/py_service/version.txt @@ -1 +1 @@ -1.4.0 +1.5.4