PROD.2026/RekomenciBackend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add edit resume

Browse Source
This commit is contained in:
ivankirpichnikov
2025-11-22 03:31:17 +03:00
parent d9a3c39980
commit a2a9b8f8c1
9 changed files with 288 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/web_api/conftest.py
+10 -5
View File
@@ -1,11 +1,13 @@
import asyncio
import logging
import os
from collections.abc import AsyncIterable, AsyncIterator
from pathlib import Path
from typing import Any
from typing import cast
import pytest
from dishka import AsyncContainer
from fastapi import FastAPI
from template_project.web_api.configuration import load_configuration
from template_project.web_api.entry_point import make_server
@@ -14,20 +16,23 @@ from tests.web_api.ioc import make_ioc
@pytest.fixture(scope="session")
async def dishka_container(backend: Any) -> AsyncIterable[AsyncContainer]:
async def dishka_container(backend: FastAPI) -> AsyncIterable[AsyncContainer]:
path = Path(os.environ["CONFIGURATION_PATH"])
configuration = load_configuration(path)
ioc = make_ioc(configuration)
ioc = make_ioc(configuration, backend)
yield ioc
await ioc.close()
@pytest.fixture(scope="session")
async def backend() -> AsyncIterator[None]:
async def backend() -> AsyncIterator[FastAPI]:
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
configuration = load_configuration(Path(os.environ["CONFIGURATION_PATH"]))
server = make_server(configuration)
asyncio.create_task(server.serve()) # type: ignore[unused-awaitable]
yield
yield cast(FastAPI, server.config.app)
@pytest.fixture
tests/web_api/e2e/test_resume.py
+127 -2
View File
@@ -1,10 +1,15 @@
from typing import Final
from dirty_equals import IsDict, IsUUID
from dirty_equals import IsDict, IsPartialDict, IsUUID
from dishka import FromDishka
from uuid_utils.compat import uuid7
from tests.web_api.helpers import is_not_found_response, is_success_response, is_unauthorized_response
from tests.web_api.helpers import (
is_forbidden_response,
is_not_found_response,
is_success_response,
is_unauthorized_response,
)
from tests.web_api.ioc import DatabaseClearer, inject
from tests.web_api.test_api_gateway import TestApiGateway
@@ -129,3 +134,123 @@ async def test_not_found_get_resume(
resume_id=str(uuid7()),
)
assert is_not_found_response(response)
@inject
async def test_success_edit_resume(
unique_email: str,
test_api_gateway: FromDishka[TestApiGateway],
database_clearer: FromDishka[DatabaseClearer],
) -> None:
await database_clearer.clear()
response_sign_up = await test_api_gateway.sign_up_email(unique_email, DEFAULT_PASSWORD)
access_token = response_sign_up.json()["access_token"]
response = await test_api_gateway.create_resume(
access_token=access_token,
about_me="About me",
experience_type="noExperience",
key_skills=["i love lisp", "i love rust"],
position="Position",
)
resume_id = response.json()["resume_id"]
response = await test_api_gateway.edit_resume(
access_token=access_token,
resume_id=resume_id,
about_me="Updated about me",
experience_type="between1And3",
key_skills=["i love python"],
position="Updated Position",
)
assert is_success_response(response)
assert response.json() == IsPartialDict(
position="Updated Position",
about_me="Updated about me",
key_skills=["i love python"],
experience_type="between1And3",
)
@inject
async def test_unauthorized_edit_resume(
unique_email: str,
test_api_gateway: FromDishka[TestApiGateway],
database_clearer: FromDishka[DatabaseClearer],
) -> None:
await database_clearer.clear()
response_sign_up = await test_api_gateway.sign_up_email(unique_email, DEFAULT_PASSWORD)
access_token = response_sign_up.json()["access_token"]
response = await test_api_gateway.create_resume(
access_token=access_token,
about_me="About me",
experience_type="noExperience",
key_skills=["i love lisp", "i love rust"],
position="Position",
)
resume_id = response.json()["resume_id"]
response = await test_api_gateway.edit_resume(
access_token="...",
resume_id=resume_id,
about_me="Updated about me",
experience_type="between1And3",
key_skills=["i love python"],
position="Updated Position",
)
assert is_unauthorized_response(response)
@inject
async def test_not_found_edit_resume(
unique_email: str,
test_api_gateway: FromDishka[TestApiGateway],
database_clearer: FromDishka[DatabaseClearer],
) -> None:
await database_clearer.clear()
response_sign_up = await test_api_gateway.sign_up_email(unique_email, DEFAULT_PASSWORD)
access_token = response_sign_up.json()["access_token"]
response = await test_api_gateway.edit_resume(
access_token=access_token,
resume_id=str(uuid7()),
about_me="Updated about me",
experience_type="between1And3",
key_skills=["i love python"],
position="Updated Position",
)
assert is_not_found_response(response)
@inject
async def test_forbidden_edit_resume(
unique_email: str,
test_api_gateway: FromDishka[TestApiGateway],
database_clearer: FromDishka[DatabaseClearer],
) -> None:
await database_clearer.clear()
response_sign_up = await test_api_gateway.sign_up_email(unique_email, DEFAULT_PASSWORD)
response = await test_api_gateway.create_resume(
access_token=response_sign_up.json()["access_token"],
about_me="About me",
experience_type="noExperience",
key_skills=["i love lisp", "i love rust"],
position="Position",
)
response_other_sign_up = await test_api_gateway.sign_up_email("f" + unique_email, DEFAULT_PASSWORD)
response = await test_api_gateway.edit_resume(
access_token=response_other_sign_up.json()["access_token"],
resume_id=response.json()["resume_id"],
about_me="Updated about me",
experience_type="between1And3",
key_skills=["i love python"],
position="Updated Position",
)
assert is_forbidden_response(response)
tests/web_api/ioc.py
+9 -5
View File
@@ -1,4 +1,4 @@
from collections.abc import Callable
from collections.abc import AsyncIterable, Callable
from inspect import Parameter
from typing import Final
@@ -13,7 +13,8 @@ from dishka import (
)
from dishka.integrations.base import wrap_injection
from dishka.integrations.fastapi import FastapiProvider
from httpx import AsyncClient
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
@@ -71,11 +72,13 @@ class TestProvider(Provider):
database_clearer = provide(DatabaseClearer)
@provide
def http_client(self) -> AsyncClient:
return AsyncClient(base_url="http://backend:8080")
async def http_client(self, app: FastAPI) -> AsyncIterable[AsyncClient]:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
yield client
def make_ioc(configuration: Configuration) -> AsyncContainer:
def make_ioc(configuration: Configuration, app: FastAPI) -> AsyncContainer:
return make_async_container(
IdPProvider(),
FactoryProvider(),
@@ -97,6 +100,7 @@ def make_ioc(configuration: Configuration) -> AsyncContainer:
FirebaseConfiguration: configuration.firebase,
Configuration: configuration,
S3Config: configuration.s3,
FastAPI: app,
},
)
tests/web_api/test_api_gateway.py
+24
View File
@@ -78,3 +78,27 @@ class TestApiGateway:
f"/resume/{resume_id}",
headers=make_auth_headers(access_token),
)
async def edit_resume(
self,
access_token: str,
resume_id: str,
position: str | None = None,
about_me: str | None = None,
key_skills: list[str] | None = None,
experience_type: str | None = None,
) -> Response:
return await self._client.patch(
f"/resume/{resume_id}",
json={
key: value
for key, value in {
"position": position,
"about_me": about_me,
"key_skills": key_skills,
"experience_type": experience_type,
}.items()
if value is not None
},
headers=make_auth_headers(access_token),
)