1
0
mirror of https://github.com/ijaric/voice_assistant.git synced 2025-07-12 20:53:24 +00:00

feat: [#41] add tests from template

This commit is contained in:
ksieuk 2023-10-08 22:15:52 +03:00
parent e05eba76fd
commit 24992e5464
21 changed files with 336 additions and 0 deletions

View File

@ -9,6 +9,10 @@ NGINX_PORT=80
API_HOST=0.0.0.0
API_PORT=8000
TEST_API_PROTOCOL=http
TEST_API_HOST=api
TEST_API_PORT=8000
JWT_SECRET_KEY=v9LctjUWwol4XbvczPiLFMDtZ8aal7mm
JWT_ALGORITHM=HS256

View File

@ -0,0 +1,18 @@
FROM python:3.11
RUN apt-get update
WORKDIR /opt/app
COPY pyproject.toml ./
COPY poetry.lock ./
RUN apt-get update \
&& pip install poetry \
&& poetry config virtualenvs.create false \
&& poetry install --no-dev
COPY tests tests
COPY lib lib
CMD ["pytest"]

View File

@ -0,0 +1,56 @@
version: "3"
services:
postgres:
image: postgres:15.2
restart: always
environment:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: ${POSTGRES_NAME}
env_file:
- .env
expose:
- "${POSTGRES_PORT}"
volumes:
- postgres_data:/var/lib/postgresql/data/
networks:
- backend_network
api:
build:
context: .
container_name: api
image: fastapi_app
restart: always
entrypoint: ["/opt/app/entrypoint.sh"]
env_file:
- .env
expose:
- "${API_PORT}"
depends_on:
- postgres
networks:
- backend_network
- api_network
tests:
build:
context: .
dockerfile: "Dockerfile.tests"
env_file:
- .env
depends_on:
- postgres
- api
networks:
- api_network
volumes:
postgres_data:
networks:
api_network:
driver: bridge
backend_network:
driver: bridge

View File

@ -0,0 +1,70 @@
import asyncio
import typing
import fastapi
import httpx
import pytest_asyncio
import lib.app as lib_app
import tests.core.settings as tests_core_settings
import tests.functional.models as functional_models
@pytest_asyncio.fixture # type: ignore[reportUntypedFunctionDecorator]
async def http_client(
base_url: str = tests_core_settings.tests_settings.api.get_api_url,
) -> typing.AsyncGenerator[httpx.AsyncClient, typing.Any]:
session = httpx.AsyncClient(base_url=base_url)
yield session
await session.aclose()
@pytest_asyncio.fixture # type: ignore[reportUntypedFunctionDecorator]
async def make_request(http_client: httpx.AsyncClient):
async def inner(
api_method: str = "",
method: functional_models.MethodsEnum = functional_models.MethodsEnum.GET,
headers: dict[str, str] = tests_core_settings.tests_settings.api.headers,
body: dict[str, typing.Any] | None = None,
jwt_token: str | None = None,
) -> functional_models.HTTPResponse:
if jwt_token is not None:
headers["Authorization"] = f"Bearer {jwt_token}"
client_params = {"json": body, "headers": headers}
if method == functional_models.MethodsEnum.GET:
del client_params["json"]
response = await getattr(http_client, method.value)(api_method, **client_params)
return functional_models.HTTPResponse(
body=response.json(),
headers=response.headers,
status_code=response.status_code,
)
return inner
@pytest_asyncio.fixture(scope="session") # type: ignore[reportUntypedFunctionDecorator]
def app() -> fastapi.FastAPI:
settings = lib_app.Settings()
application = lib_app.Application.from_settings(settings)
fastapi_app = application._fastapi_app # type: ignore[reportPrivateUsage]
return fastapi_app
@pytest_asyncio.fixture # type: ignore[reportUntypedFunctionDecorator]
async def app_http_client(
app: fastapi.FastAPI,
base_url: str = tests_core_settings.tests_settings.api.get_api_url,
) -> typing.AsyncGenerator[httpx.AsyncClient, typing.Any]:
session = httpx.AsyncClient(app=app, base_url=base_url)
yield session
await session.aclose()
@pytest_asyncio.fixture(scope="session") # type: ignore[reportUntypedFunctionDecorator]
def event_loop():
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()

View File

@ -0,0 +1,5 @@
from .settings import *
__all__ = [
"tests_settings",
]

View File

@ -0,0 +1,17 @@
import pydantic
import pydantic_settings
import tests.core.split_settings as app_split_settings
class TestsSettings(pydantic_settings.BaseSettings):
api: app_split_settings.ApiSettings = pydantic.Field(default_factory=lambda: app_split_settings.ApiSettings())
postgres: app_split_settings.PostgresSettings = pydantic.Field(
default_factory=lambda: app_split_settings.PostgresSettings()
)
project: app_split_settings.ProjectSettings = pydantic.Field(
default_factory=lambda: app_split_settings.ProjectSettings()
)
tests_settings = TestsSettings()

View File

@ -0,0 +1,9 @@
from .api import *
from .postgres import *
from .project import *
__all__ = [
"ApiSettings",
"PostgresSettings",
"ProjectSettings",
]

View File

@ -0,0 +1,23 @@
import pydantic
import pydantic_settings
import lib.app.split_settings.utils as app_split_settings_utils
class ApiSettings(pydantic_settings.BaseSettings):
model_config = pydantic_settings.SettingsConfigDict(
env_file=app_split_settings_utils.ENV_PATH,
env_prefix="TEST_API_",
env_file_encoding="utf-8",
extra="ignore",
)
protocol: str = "http"
host: str = "0.0.0.0"
port: int = 8000
headers: dict[str, str] = {"Content-Type": "application/json"}
@pydantic.computed_field
@property
def get_api_url(self) -> str:
return f"{self.protocol}://{self.host}:{self.port}/api/v1"

View File

@ -0,0 +1,42 @@
import pydantic
import pydantic_settings
import lib.app.split_settings.utils as app_split_settings_utils
class PostgresSettings(pydantic_settings.BaseSettings):
model_config = pydantic_settings.SettingsConfigDict(
env_file=app_split_settings_utils.ENV_PATH,
env_prefix="POSTGRES_",
env_file_encoding="utf-8",
extra="ignore",
)
name: str = "test_database_name"
host: str = "localhost"
port: int = 5432
user: str = "app"
password: pydantic.SecretStr = pydantic.Field(
default=...,
validation_alias=pydantic.AliasChoices("password", "postgres_password"),
)
@property
def db_uri_async(self) -> str:
db_uri: str = "postgresql+asyncpg://{pg_user}:{pg_pass}@{pg_host}/{pg_dbname}".format(
pg_user=self.user,
pg_pass=self.password.get_secret_value(),
pg_host=self.host,
pg_dbname=self.name,
)
return db_uri
@property
def db_uri_sync(self) -> str:
db_uri: str = "postgresql://{pg_user}:{pg_pass}@{pg_host}/{pg_dbname}".format(
pg_user=self.user,
pg_pass=self.password.get_secret_value(),
pg_host=self.host,
pg_dbname=self.name,
)
return db_uri

View File

@ -0,0 +1,15 @@
import pydantic
import pydantic_settings
import lib.app.split_settings.utils as app_split_settings_utils
class ProjectSettings(pydantic_settings.BaseSettings):
model_config = pydantic_settings.SettingsConfigDict(
env_file=app_split_settings_utils.ENV_PATH,
env_file_encoding="utf-8",
extra="ignore",
)
debug: bool = False
jwt_secret_key: pydantic.SecretStr = pydantic.Field(default=..., validation_alias="jwt_secret_key")

View File

@ -0,0 +1,4 @@
import pathlib
BASE_PATH = pathlib.Path(__file__).parent.parent.parent.parent.parent.resolve()
ENV_PATH = BASE_PATH / ".env"

View File

@ -0,0 +1,7 @@
from .http import *
__all__ = [
"HTTPResponse",
"MakeResponseCallableType",
"MethodsEnum",
]

View File

@ -0,0 +1,35 @@
import dataclasses
import enum
import typing
import multidict
import tests.core.settings as functional_settings
class MethodsEnum(enum.Enum):
GET = "get"
POST = "post"
PUT = "put"
DELETE = "delete"
PATCH = "patch"
@dataclasses.dataclass
class HTTPResponse:
body: dict[str, typing.Any] | str
headers: multidict.CIMultiDictProxy[str]
status_code: int
class MakeResponseCallableType(typing.Protocol):
async def __call__(
self,
api_method: str = "",
url: str = functional_settings.tests_settings.api.get_api_url,
method: MethodsEnum = MethodsEnum.GET,
headers: dict[str, str] = functional_settings.tests_settings.api.headers,
body: dict[str, typing.Any] | None = None,
jwt_token: str | None = None,
) -> HTTPResponse:
...

View File

@ -0,0 +1,17 @@
import http
import pytest
import tests.functional.models as tests_functional_models
pytestmark = [pytest.mark.asyncio]
async def test_health(
make_request: tests_functional_models.MakeResponseCallableType,
):
response = await make_request(
method=tests_functional_models.MethodsEnum.GET,
api_method=f"/health/",
)
assert response.status_code == http.HTTPStatus.OK

View File

View File

@ -0,0 +1,3 @@
[pytest]
log_format = %(asctime)s %(levelname)s %(message)s
log_date_format = %Y-%m-%d %H:%M:%S

View File

View File

View File

@ -0,0 +1,11 @@
import http
import httpx
import pytest
pytestmark = [pytest.mark.asyncio]
async def test_health(app_http_client: httpx.AsyncClient) -> None:
response = await app_http_client.get("/health/")
assert response.status_code == http.HTTPStatus.OK