Филиппов Денис Task3

.gitignore

@@ -0,0 +1,4 @@
+/.idea
+*/__pycache__
+/venv
+/.pytest_cash

README.md

@@ -1 +1,47 @@
-# WebServiceDevelopment
+# WebServiceDevelopment
+
+Это репозиторий, в котором будут храниться решения ДЗ по курсу
+разработки веб-сервисов на питоне. Домашними заданиями является работа
+над проектом. Пока что в качестве проекта сделать "датаграм" (типо
+"телеграм", но "датаграм", потому что на курсе по Java нам предлагали
+различать UDP и TCP так, что UDP -- это датаграмма, а TCP -- телеграмма).
+
+## Установка необходимых пакетов
+`pip install strawberry-graphql[debug-server]` -- установка библиотеки
+[strawberry](https://strawberry.rocks/).
+Ещё нужно установить `fastapi` и `pydantic`, позже добавлю инструкции.
+
+## Запуск
+`uvicorn main:datagram_app --reload` -- для запуска.
+
+http://127.0.0.1:8000/docs -- ручки
+
+http://127.0.0.1:8000/graphql -- работа с GraphQL
+
+## GraphQL
+Пример запроса:
+
+```
+{
+  users (year:2001) {
+    name
+    additionalInfo {
+      status
+      birthDate  {
+        year
+        day
+      }
+    }
+  }
+}
+```
+
+Все схемы и типы указаны в файле `strawbery_types.py`
+
+## Тестирование
+Перед тестированием установите:
+`pip install email-validator`
+
+Запуск юнит тестов: `python -m pytest tests/unit_tests.py`
+
+Запуск интеграционных тестов: `python -m pytest tests/integration_tests.py`

databases/users.py

@@ -0,0 +1,52 @@
+from typing import Dict, Optional
+from datetime import date
+
+
+class UserMeta:
+    class AdditionalInfo:
+        status: Optional[str]
+        birth_date: Optional[date]
+
+        def __init__(self, status: Optional[str] = None, birth_date: Optional[date] = None):
+            self.status = status
+            self.birth_date = birth_date
+
+    email: str
+    name: str
+    additional_info: AdditionalInfo
+
+    def __init__(self, email: str, name: str, additional_info: AdditionalInfo = AdditionalInfo()):
+        self.email = email
+        self.name = name
+        self.additional_info = additional_info
+
+
+__id_pwd: Dict[int, str] = {}
+__id_user_meta: Dict[int, UserMeta] = {}
+__email_id: Dict[str, int] = {}
+
+
+def get_user_by_auth(email: str, password: str) -> Optional[UserMeta]:
+    if not is_email_occupied(email):
+        return None
+    id = __email_id[email]
+    if __id_pwd[id] != password:
+        return None
+    return __id_user_meta[id]
+
+
+def is_email_occupied(email: str) -> bool:
+    return email in __email_id
+
+
+def add_user(email: str, name: str, password: str):
+    if is_email_occupied(email):
+        raise ValueError("User with this email exists")
+    id = len(__id_pwd)
+    __id_pwd[id] = password
+    __id_user_meta[id] = UserMeta(email, name)
+    __email_id[email] = id
+
+
+def get_num_of_users():
+    return len(__id_user_meta)

main.py

@@ -0,0 +1,38 @@
+from fastapi import FastAPI, HTTPException
+from pydantic_models.users import UserCreateModel, UserAuthModel
+from databases.users import get_user_by_auth, is_email_occupied, add_user, get_num_of_users
+from strawberry_types import Query
+from strawberry.asgi import GraphQL
+import strawberry
+
+
+datagram_app = FastAPI()
+
+
+@datagram_app.get("/statistics/")
+async def get_statistics(indicator: str):
+    if indicator != "num_of_users":
+        raise HTTPException(status_code=404, detail="Don't know this indicator")
+    get_num_of_users()
+
+
+@datagram_app.post("/users/registration", status_code=201)
+async def register_user(user: UserCreateModel):
+    if is_email_occupied(user.email):
+        raise HTTPException(status_code=403, detail="This mail has been already registered")
+    add_user(user.email, user.name, user.password.get_secret_value())
+    return "Registered successfully"
+
+
+@datagram_app.post("/users/auth")
+async def authorize(user_auth: UserAuthModel):
+    user = get_user_by_auth(user_auth.email, user_auth.password.get_secret_value())
+    if user is None:
+        raise HTTPException(status_code=403, detail="Wrong mail or password")
+    return user
+
+
+schema = strawberry.Schema(query=Query)
+graphql_app = GraphQL(schema)
+datagram_app.add_route("/graphql", graphql_app)
+datagram_app.add_websocket_route("/graphql", graphql_app)

pydantic_models/users.py

@@ -0,0 +1,21 @@
+from pydantic import BaseModel, EmailStr, SecretStr, validator
+from utils import validate_new_password
+
+
+class UserAuthModel(BaseModel):
+    email: EmailStr
+    password: SecretStr
+
+    @validator('password')
+    def password_acceptable(cls, v):
+        validate_new_password(v)
+        return v
+
+
+class UserCreateModel(UserAuthModel):
+    name: str
+
+    @validator('name')
+    def name_alphanumeric(cls, v):
+        assert v.isalnum(), 'must be alphanumeric'
+        return v

strawberry_types.py

@@ -0,0 +1,46 @@
+from typing import List
+import strawberry
+
+
+@strawberry.type
+class Date:
+    year: int
+    month: int
+    day: int
+
+
+@strawberry.type
+class AdditionalInfo:
+    status: str
+    birth_date: Date
+
+
+@strawberry.type
+class UserMeta:
+    email: str
+    name: str
+    additional_info: AdditionalInfo
+
+
+@strawberry.type
+class Query:
+    @strawberry.field
+    def users(self, year: int) -> List[UserMeta]:
+        return list(filter(lambda u: u.additional_info.birth_date.year == year, [
+            UserMeta(
+                "hello@world.ru",
+                "Mike",
+                AdditionalInfo(
+                    "I am a hero",
+                    Date(2001, 1, 1)
+                )
+            ),
+            UserMeta(
+                "hi@world.ru",
+                "Pol",
+                AdditionalInfo(
+                    "DevOps",
+                    Date(1001, 1, 1)
+                )
+            ),
+        ]))

tests/integration_tests.py

@@ -0,0 +1,58 @@
+import pytest
+from fastapi.testclient import TestClient
+
+from main import datagram_app
+
+client = TestClient(datagram_app)
+
+
+@pytest.fixture()
+def setup_and_teardown_db():
+    from databases.users import __id_pwd, __id_user_meta
+    save_pwd = __id_pwd.copy()
+    save_meta = __id_user_meta.copy()
+    __id_pwd.clear()
+    __id_user_meta.clear()
+
+    __id_pwd = save_pwd
+    __id_user_meta = save_meta
+
+
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ registration test ~~~~~~~~~~~~~~~~~~
+def test_registration_success(setup_and_teardown_db):
+    rs = client.post("/users/registration", json={"email": "a@yandex.ru", "name": "a", "password": "Aa1!"})
+    assert rs.status_code == 201
+
+
+def test_registration_wrong_pwd_format(setup_and_teardown_db):
+    rs = client.post("/users/registration", json={"email": "a@yandex.ru", "name": "a", "password": "a1!"})
+    assert rs.status_code == 422
+
+
+def test_registration_mail_occupied(setup_and_teardown_db):
+    rs = client.post("/users/registration", json={"email": "a@yandex.ru", "name": "a", "password": "Aa1!"})
+    assert rs.status_code == 201
+    rs = client.post("/users/registration", json={"email": "a@yandex.ru", "name": "a", "password": "Aa1!"})
+    assert rs.status_code == 403
+
+
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ auth test ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+def test_auth_success(setup_and_teardown_db):
+    rs = client.post("/users/registration", json={"email": "a@yandex.ru", "name": "a", "password": "Aa1!"})
+    assert rs.status_code == 201
+    rs = client.post("/users/auth", json={"email": "a@yandex.ru", "password": "Aa1!"})
+    assert rs.status_code == 200
+
+
+def test_auth_user_not_exist(setup_and_teardown_db):
+    rs = client.post("/users/registration", json={"email": "a@yandex.ru", "name": "a", "password": "Aa1!"})
+    assert rs.status_code == 201
+    rs = client.post("/users/auth", json={"email": "b@yandex.ru", "password": "Aa1!"})
+    assert rs.status_code == 403
+
+
+def test_auth_wrong_password(setup_and_teardown_db):
+    rs = client.post("/users/registration", json={"email": "a@yandex.ru", "name": "a", "password": "Aa1!"})
+    assert rs.status_code == 201
+    rs = client.post("/users/auth", json={"email": "a@yandex.ru", "password": "Ba1!"})
+    assert rs.status_code == 403

tests/unit_tests.py

@@ -0,0 +1,84 @@
+import pytest
+from utils import validate_new_password
+from databases.users import add_user, get_user_by_auth
+from pydantic import SecretStr
+
+
+# ~~~~~~~~~~~~~~~~~ validate password tests ~~~~~~~~~~~~~~~~~~~~
+def test_correct_password():
+    validate_new_password(SecretStr("Ab1!"))
+
+
+def test_miss_lowercase():
+    with pytest.raises(ValueError) as ve:
+        validate_new_password(SecretStr("A1!"))
+    assert ve.value.args[1] == r"[a-z]"
+
+
+def test_miss_uppercase():
+    with pytest.raises(ValueError) as ve:
+        validate_new_password(SecretStr("b1!"))
+    assert ve.value.args[1] == r"[A-Z]"
+
+
+def test_miss_digit():
+    with pytest.raises(ValueError) as ve:
+        validate_new_password(SecretStr("Ab!"))
+    assert ve.value.args[1] == r"[0-9]"
+
+
+def test_miss_special_char():
+    with pytest.raises(ValueError) as ve:
+        validate_new_password(SecretStr("Ab1"))
+    assert ve.value.args[1] == r"[!@#_.]"
+
+
+def test_wrong_char():
+    with pytest.raises(ValueError) as ve:
+        validate_new_password(SecretStr("Ab1!$"))
+    assert ve.value.args[1] == "wrong char"
+
+
+@pytest.fixture()
+def setup_and_teardown_db():
+    from databases.users import __id_pwd, __id_user_meta
+    save_pwd = __id_pwd.copy()
+    save_meta = __id_user_meta.copy()
+    __id_pwd.clear()
+    __id_user_meta.clear()
+
+    __id_pwd = save_pwd
+    __id_user_meta = save_meta
+
+
+# ~~~~~~~~~~~~~~~~~ add new user tests ~~~~~~~~~~~~~~~~~~~~
+def test_add_user_correct_params(setup_and_teardown_db):
+    add_user(email="a@yandex.ru", name="a", password="Aa1!")
+    add_user(email="b@yandex.ru", name="b", password="Bb2@")
+
+
+def test_add_user_with_existing_name(setup_and_teardown_db):
+    add_user(email="a@yandex.ru", name="a", password="Aa1!")
+    add_user(email="b@yandex.ru", name="a", password="Aa1!")
+
+
+def test_add_user_occupied_email(setup_and_teardown_db):
+    add_user(email="a@yandex.ru", name="a", password="Aa1!")
+    with pytest.raises(ValueError):
+        add_user(email="a@yandex.ru", name="b", password="Bb2@")
+
+
+# # ~~~~~~~~~~~~~~~~~ get user meta tests ~~~~~~~~~~~~~~~~~~~~
+def test_get_user_meta_correct_params(setup_and_teardown_db):
+    add_user(email="a@yandex.ru", name="a", password="Aa1!")
+    assert get_user_by_auth(email="a@yandex.ru", password="Aa1!") is not None
+
+
+def test_get_user_meta_occupied_email(setup_and_teardown_db):
+    add_user(email="a@yandex.ru", name="a", password="Aa1!")
+    assert get_user_by_auth(email="aa@yandex.ru", password="Aa1!") is None
+
+
+def test_get_user_meta_wrong_password(setup_and_teardown_db):
+    add_user(email="a@yandex.ru", name="a", password="Aa1!")
+    assert get_user_by_auth(email="a@yandex.ru", password="Aa1!!") is None

utils.py

@@ -0,0 +1,17 @@
+from pydantic import SecretStr
+import re
+
+
+def validate_new_password(pwd: SecretStr):
+    value = pwd.get_secret_value()
+    if re.search(r"[a-z]", value) is None:
+        raise ValueError("password must contain lowercase letter", r"[a-z]")
+    if re.search(r"[A-Z]", value) is None:
+        raise ValueError("password must contain uppercase letter", r"[A-Z]")
+    if re.search(r"[0-9]", value) is None:
+        raise ValueError("password must contain digit", r"[0-9]")
+    if re.search(r"[!@#_.]", value) is None:
+        raise ValueError("password must contain special character '!', '@', '#', '_' or '.'", r"[!@#_.]")
+    m = re.search(r"[^a-zA-Z0-9!@#_.]", value)
+    if m is not None:
+        raise ValueError("password contains forbidden character " + m.group(0), "wrong char")