Task2

.gitignore

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

README.md

@@ -1 +1,15 @@
-# WebServiceDevelopment
+# WebServiceDevelopment
+
+Это репозиторий, в котором будут храниться решения ДЗ по курсу
+разработки веб-сервисов на питоне. Домашними заданиями является работа
+над проектом. Пока что в качестве проекта сделать "датаграм" (типо
+"телеграм", но "датаграм", потому что на курсе по Java нам предлагали
+различать UDP и TCP так, что UDP -- это датаграмма, а TCP -- телеграмма).
+
+## Тестирование
+Перед тестированием установите:
+`pip install email-validator`
+
+Запуск юнит тестов: `python -m pytest tests/unit_tests.py`
+
+Запуск интеграционных тестов: `python -m pytest tests/integration_tests.py`

databases/users.py

@@ -0,0 +1,39 @@
+from typing import Dict, Optional
+
+
+class UserMeta:
+    def __init__(self, email: str, name: str):
+        self.email = email
+        self.name = name
+
+
+__id_pwd: Dict[int, str] = {}
+__id_user_meta: Dict[int, UserMeta] = {}
+
+
+def get_user_by_auth(email: str, password: str) -> Optional[UserMeta]:
+    for (id, user_meta) in __id_user_meta.items():
+        if user_meta.email == email:
+            if __id_pwd[id] != password:
+                return None
+            return user_meta
+    return None
+
+
+def is_email_occupied(email: str) -> bool:
+    for (_, user_meta) in __id_user_meta.items():
+        if user_meta.email == email:
+            return True
+    return False
+
+
+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)
+
+
+def get_num_of_users():
+    return len(__id_user_meta)

main.py

@@ -0,0 +1,39 @@
+from fastapi import FastAPI, HTTPException
+from models.users import UserCreateModel, UserAuthModel
+from databases.users import get_user_by_auth, is_email_occupied, add_user, get_num_of_users
+
+datagram_app = FastAPI()
+
+user_list = [
+    {
+        'mail': 'abc@yandex.ru',
+        'password': 'abc_pwd'
+    },
+    {
+        'mail': 'qwerty@gmail.com',
+        'password': 'qwerty'
+    }
+]
+
+
+@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

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

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")