Test using the TestClient (#160)

This commit is contained in:
Sebastián Ramírez
2020-04-20 20:31:29 +02:00
committed by GitHub
parent 41a2f15d8f
commit 43d5b49bd1
8 changed files with 82 additions and 103 deletions

View File

@@ -1,16 +1,16 @@
from typing import Dict from typing import Dict
import requests from fastapi.testclient import TestClient
from app.core.config import settings from app.core.config import settings
from app.tests.utils.utils import get_server_api
def test_celery_worker_test(superuser_token_headers: Dict[str, str]) -> None: def test_celery_worker_test(
server_api = get_server_api() client: TestClient, superuser_token_headers: Dict[str, str]
) -> None:
data = {"msg": "test"} data = {"msg": "test"}
r = requests.post( r = client.post(
f"{server_api}{settings.API_V1_STR}/utils/test-celery/", f"{settings.API_V1_STR}/utils/test-celery/",
json=data, json=data,
headers=superuser_token_headers, headers=superuser_token_headers,
) )

View File

@@ -1,18 +1,16 @@
import requests from fastapi.testclient import TestClient
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.core.config import settings from app.core.config import settings
from app.tests.utils.item import create_random_item from app.tests.utils.item import create_random_item
from app.tests.utils.utils import get_server_api
def test_create_item(superuser_token_headers: dict, db: Session) -> None: def test_create_item(
server_api = get_server_api() client: TestClient, superuser_token_headers: dict, db: Session
) -> None:
data = {"title": "Foo", "description": "Fighters"} data = {"title": "Foo", "description": "Fighters"}
response = requests.post( response = client.post(
f"{server_api}{settings.API_V1_STR}/items/", f"{settings.API_V1_STR}/items/", headers=superuser_token_headers, json=data,
headers=superuser_token_headers,
json=data,
) )
assert response.status_code == 200 assert response.status_code == 200
content = response.json() content = response.json()
@@ -22,12 +20,12 @@ def test_create_item(superuser_token_headers: dict, db: Session) -> None:
assert "owner_id" in content assert "owner_id" in content
def test_read_item(superuser_token_headers: dict, db: Session) -> None: def test_read_item(
client: TestClient, superuser_token_headers: dict, db: Session
) -> None:
item = create_random_item(db) item = create_random_item(db)
server_api = get_server_api() response = client.get(
response = requests.get( f"{settings.API_V1_STR}/items/{item.id}", headers=superuser_token_headers,
f"{server_api}{settings.API_V1_STR}/items/{item.id}",
headers=superuser_token_headers,
) )
assert response.status_code == 200 assert response.status_code == 200
content = response.json() content = response.json()

View File

@@ -1,31 +1,27 @@
from typing import Dict from typing import Dict
import requests from fastapi.testclient import TestClient
from app.core.config import settings from app.core.config import settings
from app.tests.utils.utils import get_server_api
def test_get_access_token() -> None: def test_get_access_token(client: TestClient) -> None:
server_api = get_server_api()
login_data = { login_data = {
"username": settings.FIRST_SUPERUSER, "username": settings.FIRST_SUPERUSER,
"password": settings.FIRST_SUPERUSER_PASSWORD, "password": settings.FIRST_SUPERUSER_PASSWORD,
} }
r = requests.post( r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
f"{server_api}{settings.API_V1_STR}/login/access-token", data=login_data
)
tokens = r.json() tokens = r.json()
assert r.status_code == 200 assert r.status_code == 200
assert "access_token" in tokens assert "access_token" in tokens
assert tokens["access_token"] assert tokens["access_token"]
def test_use_access_token(superuser_token_headers: Dict[str, str]) -> None: def test_use_access_token(
server_api = get_server_api() client: TestClient, superuser_token_headers: Dict[str, str]
r = requests.post( ) -> None:
f"{server_api}{settings.API_V1_STR}/login/test-token", r = client.post(
headers=superuser_token_headers, f"{settings.API_V1_STR}/login/test-token", headers=superuser_token_headers,
) )
result = r.json() result = r.json()
assert r.status_code == 200 assert r.status_code == 200

View File

@@ -1,19 +1,18 @@
from typing import Dict from typing import Dict
import requests from fastapi.testclient import TestClient
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app import crud from app import crud
from app.core.config import settings from app.core.config import settings
from app.schemas.user import UserCreate from app.schemas.user import UserCreate
from app.tests.utils.utils import get_server_api, random_email, random_lower_string from app.tests.utils.utils import random_email, random_lower_string
def test_get_users_superuser_me(superuser_token_headers: Dict[str, str]) -> None: def test_get_users_superuser_me(
server_api = get_server_api() client: TestClient, superuser_token_headers: Dict[str, str]
r = requests.get( ) -> None:
f"{server_api}{settings.API_V1_STR}/users/me", headers=superuser_token_headers r = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers)
)
current_user = r.json() current_user = r.json()
assert current_user assert current_user
assert current_user["is_active"] is True assert current_user["is_active"] is True
@@ -21,11 +20,10 @@ def test_get_users_superuser_me(superuser_token_headers: Dict[str, str]) -> None
assert current_user["email"] == settings.FIRST_SUPERUSER assert current_user["email"] == settings.FIRST_SUPERUSER
def test_get_users_normal_user_me(normal_user_token_headers: Dict[str, str]) -> None: def test_get_users_normal_user_me(
server_api = get_server_api() client: TestClient, normal_user_token_headers: Dict[str, str]
r = requests.get( ) -> None:
f"{server_api}{settings.API_V1_STR}/users/me", headers=normal_user_token_headers r = client.get(f"{settings.API_V1_STR}/users/me", headers=normal_user_token_headers)
)
current_user = r.json() current_user = r.json()
assert current_user assert current_user
assert current_user["is_active"] is True assert current_user["is_active"] is True
@@ -33,15 +31,14 @@ def test_get_users_normal_user_me(normal_user_token_headers: Dict[str, str]) ->
assert current_user["email"] == settings.EMAIL_TEST_USER assert current_user["email"] == settings.EMAIL_TEST_USER
def test_create_user_new_email(superuser_token_headers: dict, db: Session) -> None: def test_create_user_new_email(
server_api = get_server_api() client: TestClient, superuser_token_headers: dict, db: Session
) -> None:
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
data = {"email": username, "password": password} data = {"email": username, "password": password}
r = requests.post( r = client.post(
f"{server_api}{settings.API_V1_STR}/users/", f"{settings.API_V1_STR}/users/", headers=superuser_token_headers, json=data,
headers=superuser_token_headers,
json=data,
) )
assert 200 <= r.status_code < 300 assert 200 <= r.status_code < 300
created_user = r.json() created_user = r.json()
@@ -50,16 +47,16 @@ def test_create_user_new_email(superuser_token_headers: dict, db: Session) -> No
assert user.email == created_user["email"] assert user.email == created_user["email"]
def test_get_existing_user(superuser_token_headers: dict, db: Session) -> None: def test_get_existing_user(
server_api = get_server_api() client: TestClient, superuser_token_headers: dict, db: Session
) -> None:
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.user.create(db, obj_in=user_in) user = crud.user.create(db, obj_in=user_in)
user_id = user.id user_id = user.id
r = requests.get( r = client.get(
f"{server_api}{settings.API_V1_STR}/users/{user_id}", f"{settings.API_V1_STR}/users/{user_id}", headers=superuser_token_headers,
headers=superuser_token_headers,
) )
assert 200 <= r.status_code < 300 assert 200 <= r.status_code < 300
api_user = r.json() api_user = r.json()
@@ -69,40 +66,37 @@ def test_get_existing_user(superuser_token_headers: dict, db: Session) -> None:
def test_create_user_existing_username( def test_create_user_existing_username(
superuser_token_headers: dict, db: Session client: TestClient, superuser_token_headers: dict, db: Session
) -> None: ) -> None:
server_api = get_server_api()
username = random_email() username = random_email()
# username = email # username = email
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
crud.user.create(db, obj_in=user_in) crud.user.create(db, obj_in=user_in)
data = {"email": username, "password": password} data = {"email": username, "password": password}
r = requests.post( r = client.post(
f"{server_api}{settings.API_V1_STR}/users/", f"{settings.API_V1_STR}/users/", headers=superuser_token_headers, json=data,
headers=superuser_token_headers,
json=data,
) )
created_user = r.json() created_user = r.json()
assert r.status_code == 400 assert r.status_code == 400
assert "_id" not in created_user assert "_id" not in created_user
def test_create_user_by_normal_user(normal_user_token_headers: Dict[str, str]) -> None: def test_create_user_by_normal_user(
server_api = get_server_api() client: TestClient, normal_user_token_headers: Dict[str, str]
) -> None:
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
data = {"email": username, "password": password} data = {"email": username, "password": password}
r = requests.post( r = client.post(
f"{server_api}{settings.API_V1_STR}/users/", f"{settings.API_V1_STR}/users/", headers=normal_user_token_headers, json=data,
headers=normal_user_token_headers,
json=data,
) )
assert r.status_code == 400 assert r.status_code == 400
def test_retrieve_users(superuser_token_headers: dict, db: Session) -> None: def test_retrieve_users(
server_api = get_server_api() client: TestClient, superuser_token_headers: dict, db: Session
) -> None:
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
@@ -113,9 +107,7 @@ def test_retrieve_users(superuser_token_headers: dict, db: Session) -> None:
user_in2 = UserCreate(email=username2, password=password2) user_in2 = UserCreate(email=username2, password=password2)
crud.user.create(db, obj_in=user_in2) crud.user.create(db, obj_in=user_in2)
r = requests.get( r = client.get(f"{settings.API_V1_STR}/users/", headers=superuser_token_headers)
f"{server_api}{settings.API_V1_STR}/users/", headers=superuser_token_headers
)
all_users = r.json() all_users = r.json()
assert len(all_users) > 1 assert len(all_users) > 1

View File

@@ -1,29 +1,34 @@
from typing import Dict, Iterator from typing import Dict, Generator
import pytest import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.core.config import settings from app.core.config import settings
from app.db.session import SessionLocal from app.db.session import SessionLocal
from app.main import app
from app.tests.utils.user import authentication_token_from_email from app.tests.utils.user import authentication_token_from_email
from app.tests.utils.utils import get_server_api, get_superuser_token_headers from app.tests.utils.utils import get_superuser_token_headers
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def db() -> Iterator[Session]: def db() -> Generator:
yield SessionLocal() yield SessionLocal()
@pytest.fixture(scope="module") @pytest.fixture(scope="module")
def server_api() -> str: def client() -> Generator:
return get_server_api() with TestClient(app) as c:
yield c
@pytest.fixture(scope="module") @pytest.fixture(scope="module")
def superuser_token_headers() -> Dict[str, str]: def superuser_token_headers(client: TestClient) -> Dict[str, str]:
return get_superuser_token_headers() return get_superuser_token_headers(client)
@pytest.fixture(scope="module") @pytest.fixture(scope="module")
def normal_user_token_headers(db: Session) -> Dict[str, str]: def normal_user_token_headers(client: TestClient, db: Session) -> Dict[str, str]:
return authentication_token_from_email(email=settings.EMAIL_TEST_USER, db=db) return authentication_token_from_email(
client=client, email=settings.EMAIL_TEST_USER, db=db
)

View File

@@ -1,23 +1,21 @@
from typing import Dict from typing import Dict
import requests from fastapi.testclient import TestClient
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app import crud from app import crud
from app.core.config import settings from app.core.config import settings
from app.models.user import User from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate from app.schemas.user import UserCreate, UserUpdate
from app.tests.utils.utils import get_server_api, random_email, random_lower_string from app.tests.utils.utils import random_email, random_lower_string
def user_authentication_headers( def user_authentication_headers(
server_api: str, email: str, password: str *, client: TestClient, email: str, password: str
) -> Dict[str, str]: ) -> Dict[str, str]:
data = {"username": email, "password": password} data = {"username": email, "password": password}
r = requests.post( r = client.post(f"{settings.API_V1_STR}/login/access-token", data=data)
f"{server_api}{settings.API_V1_STR}/login/access-token", data=data
)
response = r.json() response = r.json()
auth_token = response["access_token"] auth_token = response["access_token"]
headers = {"Authorization": f"Bearer {auth_token}"} headers = {"Authorization": f"Bearer {auth_token}"}
@@ -32,7 +30,9 @@ def create_random_user(db: Session) -> User:
return user return user
def authentication_token_from_email(*, email: str, db: Session) -> Dict[str, str]: def authentication_token_from_email(
*, client: TestClient, email: str, db: Session
) -> Dict[str, str]:
""" """
Return a valid token for the user with given email. Return a valid token for the user with given email.
@@ -47,4 +47,4 @@ def authentication_token_from_email(*, email: str, db: Session) -> Dict[str, str
user_in_update = UserUpdate(password=password) user_in_update = UserUpdate(password=password)
user = crud.user.update(db, db_obj=user, obj_in=user_in_update) user = crud.user.update(db, db_obj=user, obj_in=user_in_update)
return user_authentication_headers(get_server_api(), email, password) return user_authentication_headers(client=client, email=email, password=password)

View File

@@ -2,7 +2,7 @@ import random
import string import string
from typing import Dict from typing import Dict
import requests from fastapi.testclient import TestClient
from app.core.config import settings from app.core.config import settings
@@ -15,22 +15,13 @@ def random_email() -> str:
return f"{random_lower_string()}@{random_lower_string()}.com" return f"{random_lower_string()}@{random_lower_string()}.com"
def get_server_api() -> str: def get_superuser_token_headers(client: TestClient) -> Dict[str, str]:
server_name = f"http://{settings.SERVER_NAME}"
return server_name
def get_superuser_token_headers() -> Dict[str, str]:
server_api = get_server_api()
login_data = { login_data = {
"username": settings.FIRST_SUPERUSER, "username": settings.FIRST_SUPERUSER,
"password": settings.FIRST_SUPERUSER_PASSWORD, "password": settings.FIRST_SUPERUSER_PASSWORD,
} }
r = requests.post( r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
f"{server_api}{settings.API_V1_STR}/login/access-token", data=login_data
)
tokens = r.json() tokens = r.json()
a_token = tokens["access_token"] a_token = tokens["access_token"]
headers = {"Authorization": f"Bearer {a_token}"} headers = {"Authorization": f"Bearer {a_token}"}
# superuser_token_headers = headers
return headers return headers

View File

@@ -3,7 +3,6 @@ import logging
from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed
from app.db.session import SessionLocal from app.db.session import SessionLocal
from app.tests.api.api_v1.test_login import test_get_access_token
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -23,8 +22,6 @@ def init() -> None:
# Try to create session to check if DB is awake # Try to create session to check if DB is awake
db = SessionLocal() db = SessionLocal()
db.execute("SELECT 1") db.execute("SELECT 1")
# Wait for API to be awake, run one simple tests to authenticate
test_get_access_token()
except Exception as e: except Exception as e:
logger.error(e) logger.error(e)
raise e raise e