tests/test_users.py - main - grigorovich/ring

tests/test_users.py
338 строк13 KB
grigvlwork
21 сен 2025, 11:29
21 сен 2025, 11:294cbf7cb
import pytest
import pytest_asyncio
import asyncpg
import httpx
import os
import string

# from dotenv import load_dotenv
from datetime import datetime, timedelta, timezone
from httpx import AsyncClient
from backend.main import app

# load_dotenv(".env")

BASE_URL = "http://localhost:8000"  # порт, который проброшен в docker-compose

DB_CONFIG = {
    "user": os.getenv("POSTGRES_USER", "postgres"),
    "password": os.getenv("POSTGRES_PASSWORD", "postgres"),
    "database": os.getenv("POSTGRES_DB", "ring"),
    "host": "localhost",
    "port": "5432",
}


@pytest.mark.asyncio
async def test_register_success(async_client: AsyncClient, db_pool):
    username = "new_user"
    password = "strongpassword"
    device_name = "Pixel 7"

    response = await async_client.post(
        "/register",
        params={"username": username, "password": password, "device_name": device_name},
    )
    assert response.status_code == 200
    data = response.json()

    # Проверяем структуру ответа
    assert "id" in data
    assert data["username"] == username
    assert "access_token" in data
    assert "refresh_token" in data
    assert data["token_type"] == "bearer"

    # Проверяем, что сессия создалась в базе
    async with db_pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT * FROM sessions WHERE refresh_token = $1", data["refresh_token"]
        )
        assert row is not None
        assert row["user_id"] == data["id"]
        assert row["device_name"] == device_name
        assert isinstance(row["expires_at"], datetime)
        assert row["expires_at"] > datetime.now(timezone.utc)


# @pytest.mark.asyncio
# async def test_register_and_login_live():
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         # 1. регистрация нового пользователя
#         resp = await client.post(
#             "/register", params={"username": "bob", "password": "secret"}
#         )
#         assert resp.status_code == 200, resp.text
#         data = resp.json()
#         assert "id" in data
#         assert data["username"] == "bob"

#         # 2. успешный логин
#         resp = await client.post(
#             "/login", params={"username": "bob", "password": "secret"}
#         )
#         assert resp.status_code == 200, resp.text
#         login_data = resp.json()
#         assert "token" in login_data
#         assert "id" in login_data

#         # 3. неуспешный логин
#         resp = await client.post(
#             "/login", params={"username": "bob", "password": "wrong"}
#         )
#         assert resp.status_code == 401, resp.text
#         assert resp.json()["detail"] == "Invalid credentials"


# @pytest.mark.asyncio
# async def test_register_duplicate_username():
#     async with httpx.AsyncClient() as client:
#         # первый раз ок
#         r = await client.post(
#             f"{BASE_URL}/register", params={"username": "bob", "password": "pass"}
#         )
#         assert r.status_code == 200

#         # второй раз с тем же именем → ошибка
#         r = await client.post(
#             f"{BASE_URL}/register", params={"username": "bob", "password": "other"}
#         )
#         assert r.status_code == 400
#         assert r.json()["detail"] == "Username already exists"


# @pytest.mark.asyncio
# async def test_login_wrong_password():
#     async with httpx.AsyncClient() as client:
#         # создаём пользователя
#         await client.post(
#             f"{BASE_URL}/register", params={"username": "charlie", "password": "good"}
#         )

#         # неправильный пароль
#         r = await client.post(
#             f"{BASE_URL}/login", params={"username": "charlie", "password": "bad"}
#         )
#         assert r.status_code == 401
#         assert r.json()["detail"] == "Invalid credentials"


# @pytest.mark.asyncio
# async def test_login_nonexistent_user():
#     async with httpx.AsyncClient() as client:
#         r = await client.post(
#             f"{BASE_URL}/login", params={"username": "ghost", "password": "none"}
#         )
#         assert r.status_code == 401
#         assert r.json()["detail"] == "Invalid credentials"


# # Тесты для контактов


# @pytest.mark.asyncio
# async def test_create_invite_code():
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         # создаём тестового пользователя
#         resp = await client.post(
#             "/register", params={"username": "alice", "password": "secret"}
#         )
#         assert resp.status_code == 200, resp.text  # проверка успешной регистрации
#         user_data = resp.json()
#         assert "id" in user_data, user_data
#         user_id = user_data["id"]

#         # создаём invite-код
#         resp = await client.post("/invite/create", params={"owner_id": user_id})
#         assert resp.status_code == 200, resp.text  # проверка успешного создания кода

#         data = resp.json()
#         assert "code" in data, data
#         code = data["code"]

#         # проверяем длину кода
#         assert 6 <= len(code) <= 8
#         # проверяем, что код содержит только буквы и цифры
#         for c in code:
#             assert c in string.ascii_uppercase + string.digits


# @pytest.mark.asyncio
# async def test_invite_code_unique():
#     """Проверка уникальности invite-кода"""
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         resp = await client.post(
#             "/register", params={"username": "bob", "password": "secret"}
#         )
#         owner_id = resp.json()["id"]

#         codes = set()
#         # создаём несколько кодов
#         for _ in range(5):
#             r = await client.post("/invite/create", params={"owner_id": owner_id})
#             code = r.json()["code"]
#             assert code not in codes  # проверка уникальности
#             codes.add(code)


# @pytest.mark.asyncio
# async def test_use_invite_code_success():
#     """Проверка успешного использования кода"""
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         # регистрируем владельца и пользователя
#         resp_owner = await client.post(
#             "/register", params={"username": "charlie", "password": "secret"}
#         )
#         owner_id = resp_owner.json()["id"]

#         resp_user = await client.post(
#             "/register", params={"username": "dave", "password": "secret"}
#         )
#         user_id = resp_user.json()["id"]

#         # создаём код
#         r = await client.post("/invite/create", params={"owner_id": owner_id})
#         code = r.json()["code"]

#         # используем код
#         r = await client.post("/invite/use", params={"user_id": user_id, "code": code})
#         assert r.status_code == 200
#         data = r.json()
#         assert data["status"] == "ok"
#         assert data["contact_added"] == owner_id


# @pytest.mark.asyncio
# async def test_use_invite_code_success():
#     """Успешное использование кода"""
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         # Регистрируем владельца и пользователя
#         resp_owner = await client.post(
#             "/register", params={"username": "owner", "password": "secret"}
#         )
#         owner_id = resp_owner.json()["id"]

#         resp_user = await client.post(
#             "/register", params={"username": "user", "password": "secret"}
#         )
#         user_id = resp_user.json()["id"]

#         # Создаем код
#         r = await client.post("/invite/create", params={"owner_id": owner_id})
#         code = r.json()["code"]

#         # Используем код
#         r2 = await client.post("/invite/use", params={"user_id": user_id, "code": code})
#         assert r2.status_code == 200
#         data = r2.json()
#         assert data["contact_added"] == owner_id


# @pytest.mark.asyncio
# async def test_use_invite_code_already_used():
#     """Попытка использовать уже использованный код"""
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         resp_owner = await client.post(
#             "/register", params={"username": "owner2", "password": "secret"}
#         )
#         owner_id = resp_owner.json()["id"]

#         resp_user = await client.post(
#             "/register", params={"username": "user2", "password": "secret"}
#         )
#         user_id = resp_user.json()["id"]

#         r = await client.post("/invite/create", params={"owner_id": owner_id})
#         code = r.json()["code"]

#         # Используем код первый раз
#         await client.post("/invite/use", params={"user_id": user_id, "code": code})

#         # Пробуем использовать повторно
#         r2 = await client.post("/invite/use", params={"user_id": user_id, "code": code})
#         assert r2.status_code == 400


# @pytest.mark.asyncio
# async def test_use_invite_code_nonexistent():
#     """Использование несуществующего кода"""
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         resp_user = await client.post(
#             "/register", params={"username": "user3", "password": "secret"}
#         )
#         user_id = resp_user.json()["id"]

#         r = await client.post(
#             "/invite/use", params={"user_id": user_id, "code": "FAKECODE"}
#         )
#         assert r.status_code == 400


# @pytest.mark.asyncio
# async def test_use_invite_code_expired():
#     """Использование истёкшего кода"""
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         resp_owner = await client.post(
#             "/register", params={"username": "owner4", "password": "secret"}
#         )
#         owner_id = resp_owner.json()["id"]

#         resp_user = await client.post(
#             "/register", params={"username": "user4", "password": "secret"}
#         )
#         user_id = resp_user.json()["id"]

#         # Создаем код вручную с истекшей датой
#         expires_at = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat()
#         await client.post(
#             "/invite/create", params={"owner_id": owner_id, "expires_at": expires_at}
#         )
#         # Получаем код из базы (можно через эндпойнт списка кодов или через фикстуру)
#         # Допустим, тут возвращаем FAKE_EXPIRED_CODE
#         code = "FAKE_EXPIRED_CODE"

#         r = await client.post("/invite/use", params={"user_id": user_id, "code": code})
#         assert r.status_code == 400


# @pytest.mark.asyncio
# async def test_use_invite_code_contact_uniqueness():
#     """Проверка, что контакты не дублируются при повторном использовании кода другим пользователем"""
#     async with httpx.AsyncClient(base_url=BASE_URL) as client:
#         # Создаем владельца
#         resp_owner = await client.post(
#             "/register", params={"username": "owner5", "password": "secret"}
#         )
#         owner_id = resp_owner.json()["id"]

#         # Создаем двух пользователей
#         resp_user1 = await client.post(
#             "/register", params={"username": "user5a", "password": "secret"}
#         )
#         user1_id = resp_user1.json()["id"]

#         resp_user2 = await client.post(
#             "/register", params={"username": "user5b", "password": "secret"}
#         )
#         user2_id = resp_user2.json()["id"]

#         # Создаем код
#         r = await client.post("/invite/create", params={"owner_id": owner_id})
#         code = r.json()["code"]

#         # Первый пользователь использует код
#         r1 = await client.post(
#             "/invite/use", params={"user_id": user1_id, "code": code}
#         )
#         assert r1.status_code == 200

#         # Второй пользователь пытается использовать тот же код
#         r2 = await client.post(
#             "/invite/use", params={"user_id": user2_id, "code": code}
#         )
#         assert r2.status_code == 400  # код уже использован

#         # Проверяем, что контакты владельца не дублируются
#         r_contacts = await client.get(f"/contacts/{owner_id}")
#         contacts = r_contacts.json()["contacts"]
#         assert contacts.count(user1_id) == 1