From 394f1e36b054bc0b12697394435abc200492b45b Mon Sep 17 00:00:00 2001 From: Erick Duarte Date: Wed, 15 Oct 2025 18:56:02 -0300 Subject: [PATCH] format --- app.py | 16 ++++--- conftest.py | 4 +- .../send_notification_usecase.py | 24 +++++----- system_notification/config/config.py | 6 +-- .../jwt_auth_controller_decorator.py | 24 +++++++--- .../domain/notification_factory_caller.py | 26 +++++++---- .../domain/notifications/base_notification.py | 8 ++-- .../factories/slack_notification_factory.py | 18 +++++--- .../notifications/notification_target.py | 2 +- .../domain/protocols/controller_protocol.py | 10 +++-- .../protocols/factory_caller_protocol.py | 12 +++-- .../domain/protocols/jwt_adapter_protocol.py | 2 +- .../notification_factory_protocol.py | 8 +++- .../domain/protocols/notification_sender.py | 4 +- .../client/aiohttp_adapter/aiohttp_adapter.py | 23 ++++++---- .../http/client/aiohttp_adapter/protocols.py | 2 +- .../controller/health_check_controller.py | 12 +++-- .../send_notification_controller.py | 40 ++++++++++------- .../infra/http/server/fastapi_http_server.py | 9 ++-- .../infra/http/server/flask_http_server.py | 14 +++--- .../infra/jwt/jose_jwt_adapter.py | 6 +-- .../slack_notification_handler.py | 30 ++++++++----- .../test_send_notification_usecase.py | 36 ++++++++++----- .../domain/decorators/test_auth_decorator.py | 30 ++++++++----- .../notifications/test_base_notification.py | 33 +++++++++----- .../notifications/test_notification_target.py | 6 +-- tests/domain/test_factory_caller.py | 26 ++++++----- .../infra/http/client/test_aiohttp_adapter.py | 18 ++++---- .../controller/api_notification_serializer.py | 14 ++++-- .../test_send_notification_controller.py | 42 +++++++++++------- tests/infra/jwt/test_jose_jw_adapter.py | 16 +++---- .../test_slack_notification_handler.py | 20 ++++++--- .../infra/server/test_fastapi_http_server.py | 44 ++++++++++++------- 33 files changed, 368 insertions(+), 217 deletions(-) diff --git a/app.py b/app.py index b0ed8f3..0642ff9 100644 --- a/app.py +++ b/app.py @@ -19,17 +19,23 @@ from system_notification.infra.http.controller.send_notification_controller import ( SendNotificationController, ) -from system_notification.infra.http.server.fastapi_http_server import FastApiHttpServer -from system_notification.infra.http.server.flask_http_server import FlaskHttpServer +from system_notification.infra.http.server.fastapi_http_server import ( + FastApiHttpServer, +) +from system_notification.infra.http.server.flask_http_server import ( + FlaskHttpServer, +) from tests.infra.http.controller.api_notification_serializer import ( ApiNotificationSerializer, ) -server = FastApiHttpServer(app=FastAPI(root_path=f"/{os.environ.get('STAGE', '')}")) +server = FastApiHttpServer( + app=FastAPI(root_path=f"/{os.environ.get('STAGE', '')}") +) # server = FlaskHttpServer() factory_caller = NotificationFactoryCaller() factory_caller.add_factory( - SlackNotificationFactory(slack_token=SETTINGS.get("SLACK_API_TOKEN", "")) + SlackNotificationFactory(slack_token=SETTINGS.get('SLACK_API_TOKEN', '')) ) uc = SendNotificationUseCase(factory_caller=factory_caller) send_notification_controller = SendNotificationController( @@ -41,5 +47,5 @@ http_server=server, ) handler = Mangum(server._app) -if __name__ == "__main__": +if __name__ == '__main__': server.serve(port=5000) diff --git a/conftest.py b/conftest.py index 5ba592c..6f470dc 100755 --- a/conftest.py +++ b/conftest.py @@ -1,6 +1,6 @@ def pytest_itemcollected(item): node = item.obj if node.__doc__: - test_title_parts = item._nodeid.split("::") + test_title_parts = item._nodeid.split('::') test_title_parts[-1] = node.__doc__.strip() - item._nodeid = "::".join(test_title_parts) + item._nodeid = '::'.join(test_title_parts) diff --git a/system_notification/application/send_notification_usecase/send_notification_usecase.py b/system_notification/application/send_notification_usecase/send_notification_usecase.py index db4245c..d360043 100644 --- a/system_notification/application/send_notification_usecase/send_notification_usecase.py +++ b/system_notification/application/send_notification_usecase/send_notification_usecase.py @@ -5,10 +5,15 @@ NotificationTarget, ) from system_notification.domain.protocols import NotificationSender -from system_notification.domain.protocols.factory_caller_protocol import FactoryCaller -from system_notification.domain.protocols.notification_protocol import Notification -from system_notification.domain.exceptions.notification_error import TargetNotFound - +from system_notification.domain.protocols.factory_caller_protocol import ( + FactoryCaller, +) +from system_notification.domain.protocols.notification_protocol import ( + Notification, +) +from system_notification.domain.exceptions.notification_error import ( + TargetNotFound, +) @dataclass @@ -18,7 +23,7 @@ class SendNotificationInput: target: List[NotificationTarget] = field(default_factory=list) priority: Literal[0, 1, 2, 3] = 0 placeholders: Dict[str, str] = field(default_factory=dict) - icon: str = "" + icon: str = '' @dataclass @@ -37,10 +42,7 @@ async def execute( output: List[SendNotificationOutput] = [] if not input.target: raise TargetNotFound( - { - "is_sent": False, - "detail": "missing at least one target" - } + {'is_sent': False, 'detail': 'missing at least one target'} ) for target in input.target: sender: Optional[ @@ -59,6 +61,8 @@ async def execute( notification.icon = input.icon await sender.send(notification) output.append( - SendNotificationOutput(status=notification.status, target=target) + SendNotificationOutput( + status=notification.status, target=target + ) ) return output diff --git a/system_notification/config/config.py b/system_notification/config/config.py index 772f8a8..35d5aa9 100644 --- a/system_notification/config/config.py +++ b/system_notification/config/config.py @@ -3,11 +3,11 @@ from dynaconf import Dynaconf settings = Dynaconf( - settings_files=["settings.toml", ".secrets.toml"], + settings_files=['settings.toml', '.secrets.toml'], envvar_prefix=False, ) -ENVIRON_TYPE = settings.get("environ_type", "dev") -SETTINGS = settings.get(ENVIRON_TYPE) or defaultdict(lambda: "missing data") +ENVIRON_TYPE = settings.get('environ_type', 'dev') +SETTINGS = settings.get(ENVIRON_TYPE) or defaultdict(lambda: 'missing data') # `envvar_prefix` = export envvars with `export DYNACONF_FOO=bar`. diff --git a/system_notification/domain/decorators/jwt_auth_controller_decorator.py b/system_notification/domain/decorators/jwt_auth_controller_decorator.py index 8311aa6..479f71d 100644 --- a/system_notification/domain/decorators/jwt_auth_controller_decorator.py +++ b/system_notification/domain/decorators/jwt_auth_controller_decorator.py @@ -2,9 +2,15 @@ from typing import Any, Callable from system_notification.domain.protocols.controller_protocol import Controller -from system_notification.domain.protocols.jwt_adapter_protocol import JWTAdapter -from system_notification.infra.http.server.helpers.http_request import HttpRequest -from system_notification.infra.http.server.helpers.http_response import HttpResponse +from system_notification.domain.protocols.jwt_adapter_protocol import ( + JWTAdapter, +) +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) +from system_notification.infra.http.server.helpers.http_response import ( + HttpResponse, +) class JWTAuthControllerDecorator: @@ -20,16 +26,20 @@ def __call__(self, controller: Callable, *args, **kwargs) -> Callable: self._undecorated_callable = controller @functools.wraps(controller) - async def wrapper(controller_instance: Callable, request: HttpRequest) -> Any: - authorization_list = request.headers.get("authorization", "").split(" ") + async def wrapper( + controller_instance: Callable, request: HttpRequest + ) -> Any: + authorization_list = request.headers.get( + 'authorization', '' + ).split(' ') if len(authorization_list) <= 1 or not self._jwt_adapter.is_valid( data=authorization_list[1] ): return HttpResponse( status_code=401, body={ - "message": "Not Authorized", - "detail": "invalid API token", + 'message': 'Not Authorized', + 'detail': 'invalid API token', }, ) self._is_authenticated = True diff --git a/system_notification/domain/notification_factory_caller.py b/system_notification/domain/notification_factory_caller.py index 7b7794c..930dd11 100644 --- a/system_notification/domain/notification_factory_caller.py +++ b/system_notification/domain/notification_factory_caller.py @@ -1,15 +1,21 @@ import contextlib from typing import Dict, Literal, Optional -from system_notification.domain.exceptions.notification_error import TargetNotFound +from system_notification.domain.exceptions.notification_error import ( + TargetNotFound, +) from system_notification.domain.notifications.notification_target import ( NotificationTarget, ) from system_notification.domain.protocols.notification_factory_protocol import ( NotificationFactory, ) -from system_notification.domain.protocols.notification_protocol import Notification -from system_notification.domain.protocols.notification_sender import NotificationSender +from system_notification.domain.protocols.notification_protocol import ( + Notification, +) +from system_notification.domain.protocols.notification_sender import ( + NotificationSender, +) class NotificationFactoryCaller: @@ -19,18 +25,20 @@ def __init__(self) -> None: def add_factory(self, factory: NotificationFactory) -> None: self._factories[factory.target_type] = factory - async def get_sender(self, target: NotificationTarget) -> Optional[NotificationSender]: + async def get_sender( + self, target: NotificationTarget + ) -> Optional[NotificationSender]: with contextlib.suppress(KeyError): factory = self._factories[target.type] return factory.make_sender() raise TargetNotFound( { - "is_sent": False, - "destin": { - "target_type": target.type, - "target": target.target, + 'is_sent': False, + 'destin': { + 'target_type': target.type, + 'target': target.target, }, - "detail": "unknown_handler", + 'detail': 'unknown_handler', } ) diff --git a/system_notification/domain/notifications/base_notification.py b/system_notification/domain/notifications/base_notification.py index 28cc759..548fd5b 100644 --- a/system_notification/domain/notifications/base_notification.py +++ b/system_notification/domain/notifications/base_notification.py @@ -7,7 +7,7 @@ NotificationTarget, ) -T = TypeVar("T", covariant=True) +T = TypeVar('T', covariant=True) @dataclass @@ -15,13 +15,13 @@ class BaseNotification: title: str content: str priority: Literal[0, 1, 2, 3] = 0 - icon: str = "" + icon: str = '' def __post_init__(self) -> None: self._is_sent: bool = False self._vars: Dict[str, str] = {} self._target: Optional[NotificationTarget] = None - self._status: str = "filled" + self._status: str = 'filled' if not self.priority or type(self.priority) != int: self.priority = 0 @@ -46,7 +46,7 @@ def status(self) -> str: @status.setter def status(self, status: str) -> None: - if status.lower() in ("queued", "sent"): + if status.lower() in ('queued', 'sent'): self._is_sent = True self._status = status diff --git a/system_notification/domain/notifications/factories/slack_notification_factory.py b/system_notification/domain/notifications/factories/slack_notification_factory.py index ad2cdc9..19ab92d 100644 --- a/system_notification/domain/notifications/factories/slack_notification_factory.py +++ b/system_notification/domain/notifications/factories/slack_notification_factory.py @@ -1,15 +1,21 @@ from typing import Literal -from system_notification.domain.notifications.base_notification import BaseNotification -from system_notification.domain.protocols.notification_protocol import Notification -from system_notification.domain.protocols.notification_sender import NotificationSender +from system_notification.domain.notifications.base_notification import ( + BaseNotification, +) +from system_notification.domain.protocols.notification_protocol import ( + Notification, +) +from system_notification.domain.protocols.notification_sender import ( + NotificationSender, +) from system_notification.infra.notification_handlers.slack_notification_handler import ( SlackNotificationHandler, ) class SlackNotificationFactory: - target_type = "slack_channel" + target_type = 'slack_channel' def __init__(self, slack_token: str) -> None: self._slack_token = slack_token @@ -17,7 +23,9 @@ def __init__(self, slack_token: str) -> None: def make_notificaton( self, title: str, content: str, priority: Literal[0, 1, 2, 3] = 0 ) -> Notification: - return BaseNotification(title=title, content=content, priority=priority) + return BaseNotification( + title=title, content=content, priority=priority + ) def make_sender(self) -> NotificationSender: return SlackNotificationHandler(token=self._slack_token) diff --git a/system_notification/domain/notifications/notification_target.py b/system_notification/domain/notifications/notification_target.py index d73874a..a30039a 100644 --- a/system_notification/domain/notifications/notification_target.py +++ b/system_notification/domain/notifications/notification_target.py @@ -15,4 +15,4 @@ def target(self) -> str: return str(self._target).lower() def __str__(self) -> str: - return f"{self.type}:{self.target}" + return f'{self.type}:{self.target}' diff --git a/system_notification/domain/protocols/controller_protocol.py b/system_notification/domain/protocols/controller_protocol.py index 6b5d59e..6d4d486 100644 --- a/system_notification/domain/protocols/controller_protocol.py +++ b/system_notification/domain/protocols/controller_protocol.py @@ -1,14 +1,18 @@ from typing import Protocol -from system_notification.infra.http.server.helpers.http_request import HttpRequest -from system_notification.infra.http.server.helpers.http_response import HttpResponse +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) +from system_notification.infra.http.server.helpers.http_response import ( + HttpResponse, +) class HttpServer(Protocol): def serve(self, port: int = 8000) -> None: ... - def on(self, method: str, url: str, controller: "Controller") -> None: + def on(self, method: str, url: str, controller: 'Controller') -> None: ... diff --git a/system_notification/domain/protocols/factory_caller_protocol.py b/system_notification/domain/protocols/factory_caller_protocol.py index a85a103..853e0bf 100644 --- a/system_notification/domain/protocols/factory_caller_protocol.py +++ b/system_notification/domain/protocols/factory_caller_protocol.py @@ -6,15 +6,21 @@ from system_notification.domain.protocols.notification_factory_protocol import ( NotificationFactory, ) -from system_notification.domain.protocols.notification_protocol import Notification -from system_notification.domain.protocols.notification_sender import NotificationSender +from system_notification.domain.protocols.notification_protocol import ( + Notification, +) +from system_notification.domain.protocols.notification_sender import ( + NotificationSender, +) class FactoryCaller(Protocol): def add_factory(self, factory: NotificationFactory) -> None: pass - async def get_sender(self, target: NotificationTarget) -> Optional[NotificationSender]: + async def get_sender( + self, target: NotificationTarget + ) -> Optional[NotificationSender]: pass async def get_notification( diff --git a/system_notification/domain/protocols/jwt_adapter_protocol.py b/system_notification/domain/protocols/jwt_adapter_protocol.py index 039e284..824491f 100644 --- a/system_notification/domain/protocols/jwt_adapter_protocol.py +++ b/system_notification/domain/protocols/jwt_adapter_protocol.py @@ -1,6 +1,6 @@ from typing import Dict, Protocol, TypeVar -JWT_TYPES = TypeVar("JWT_TYPES", str, bytes, int, float) +JWT_TYPES = TypeVar('JWT_TYPES', str, bytes, int, float) class JWTAdapter(Protocol): diff --git a/system_notification/domain/protocols/notification_factory_protocol.py b/system_notification/domain/protocols/notification_factory_protocol.py index 50c1954..ea440eb 100644 --- a/system_notification/domain/protocols/notification_factory_protocol.py +++ b/system_notification/domain/protocols/notification_factory_protocol.py @@ -1,7 +1,11 @@ from typing import Literal, Protocol, Tuple -from system_notification.domain.protocols.notification_protocol import Notification -from system_notification.domain.protocols.notification_sender import NotificationSender +from system_notification.domain.protocols.notification_protocol import ( + Notification, +) +from system_notification.domain.protocols.notification_sender import ( + NotificationSender, +) class NotificationFactory(Protocol): diff --git a/system_notification/domain/protocols/notification_sender.py b/system_notification/domain/protocols/notification_sender.py index 3912bbd..de18e08 100644 --- a/system_notification/domain/protocols/notification_sender.py +++ b/system_notification/domain/protocols/notification_sender.py @@ -1,6 +1,8 @@ from typing import Protocol, runtime_checkable -from system_notification.domain.protocols.notification_protocol import Notification +from system_notification.domain.protocols.notification_protocol import ( + Notification, +) @runtime_checkable diff --git a/system_notification/infra/http/client/aiohttp_adapter/aiohttp_adapter.py b/system_notification/infra/http/client/aiohttp_adapter/aiohttp_adapter.py index bde1dba..87f8ae8 100644 --- a/system_notification/infra/http/client/aiohttp_adapter/aiohttp_adapter.py +++ b/system_notification/infra/http/client/aiohttp_adapter/aiohttp_adapter.py @@ -13,7 +13,7 @@ def __init__(self, client_session: Optional[ClientSession] = None) -> None: self._client_session = client_session async def _make_async(self, data): - setattr(data, "status_code", data.status) + setattr(data, 'status_code', data.status) json_data = {} with contextlib.suppress(aiohttp.client_exceptions.ContentTypeError): json_data = await data.json() @@ -25,7 +25,12 @@ async def async_json(): return data async def __request( - self, session: aiohttp.ClientSession, method: str, url: str, json, files=None + self, + session: aiohttp.ClientSession, + method: str, + url: str, + json, + files=None, ) -> aiohttp.ClientResponse: session = self._client_session or session if files: @@ -43,19 +48,21 @@ async def _request( self, method: str, url: str, json=None, files=None ) -> aiohttp.ClientResponse: async with aiohttp.ClientSession() as session: - return await self.__request(session, method, url, json, files=files) + return await self.__request( + session, method, url, json, files=files + ) def get(self, url: str) -> Coroutine: - return self._request("GET", url) + return self._request('GET', url) def post(self, url: str, json, files=None) -> Coroutine: - return self._request("POST", url, json, files=files) + return self._request('POST', url, json, files=files) def put(self, url: str, json) -> Coroutine: - return self._request("PUT", url, json) + return self._request('PUT', url, json) def patch(self, url: str, json) -> Coroutine: - return self._request("PATCH", url, json or {}) + return self._request('PATCH', url, json or {}) def delete(self, url: str) -> Coroutine: - return self._request("DELETE", url) + return self._request('DELETE', url) diff --git a/system_notification/infra/http/client/aiohttp_adapter/protocols.py b/system_notification/infra/http/client/aiohttp_adapter/protocols.py index b29c587..eff79ee 100644 --- a/system_notification/infra/http/client/aiohttp_adapter/protocols.py +++ b/system_notification/infra/http/client/aiohttp_adapter/protocols.py @@ -8,7 +8,7 @@ def request(self, *args, **kwargs) -> Any: async def read(self, *args, **kwargs) -> Any: return - async def __aenter__(self) -> "ClientSession": + async def __aenter__(self) -> 'ClientSession': return self async def __aexit__(self, *args, **kwargs) -> None: diff --git a/system_notification/infra/http/controller/health_check_controller.py b/system_notification/infra/http/controller/health_check_controller.py index be17617..029fdf7 100644 --- a/system_notification/infra/http/controller/health_check_controller.py +++ b/system_notification/infra/http/controller/health_check_controller.py @@ -1,6 +1,10 @@ from system_notification.domain.protocols.controller_protocol import HttpServer -from system_notification.infra.http.server.helpers.http_request import HttpRequest -from system_notification.infra.http.server.helpers.http_response import HttpResponse +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) +from system_notification.infra.http.server.helpers.http_response import ( + HttpResponse, +) class HealthCheckController: @@ -11,7 +15,7 @@ def __init__( http_server: HttpServer, ) -> None: self.http_server = http_server - self.http_server.on(method="GET", url="/healthz", controller=self) + self.http_server.on(method='GET', url='/healthz', controller=self) async def handle(self, request: HttpRequest) -> HttpResponse: - return HttpResponse(body="OK") + return HttpResponse(body='OK') diff --git a/system_notification/infra/http/controller/send_notification_controller.py b/system_notification/infra/http/controller/send_notification_controller.py index 90d19fd..e238d01 100644 --- a/system_notification/infra/http/controller/send_notification_controller.py +++ b/system_notification/infra/http/controller/send_notification_controller.py @@ -7,13 +7,19 @@ SendNotificationUseCase, ) from system_notification.domain.decorators import JWTAuthControllerDecorator -from system_notification.domain.exceptions.notification_error import TargetNotFound +from system_notification.domain.exceptions.notification_error import ( + TargetNotFound, +) from system_notification.domain.notifications.notification_target import ( NotificationTarget, ) from system_notification.domain.protocols.controller_protocol import HttpServer -from system_notification.infra.http.server.helpers.http_request import HttpRequest -from system_notification.infra.http.server.helpers.http_response import HttpResponse +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) +from system_notification.infra.http.server.helpers.http_response import ( + HttpResponse, +) from system_notification.infra.jwt.jose_jwt_adapter import JoseJWTAdapter from tests.infra.http.controller.api_notification_serializer import ( ApiNotificationSerializer, @@ -32,7 +38,7 @@ def __init__( self.http_server = http_server self.send_notification = send_notifcation_usecase self.serializer = serializer - self.http_server.on("POST", "/notification", self) + self.http_server.on('POST', '/notification', self) @JWTAuthControllerDecorator(JoseJWTAdapter()) async def handle(self, request: HttpRequest) -> HttpResponse: @@ -43,36 +49,36 @@ async def handle(self, request: HttpRequest) -> HttpResponse: try: return await self._handle(request) except TargetNotFound as err: - return HttpResponse(status_code=400, body={"error": err.args}) + return HttpResponse(status_code=400, body={'error': err.args}) # except Exception as err: # print("exception::", err, err.__traceback__.) # return HttpResponse(status_code=500, body={"error": err.args}) async def _handle(self, request: HttpRequest) -> HttpResponse: - notification = request.body.get("data", {}) + notification = request.body.get('data', {}) targets: List[NotificationTarget] = [ NotificationTarget( - destin.get("type"), - destin.get("target"), + destin.get('type'), + destin.get('target'), ) - for destin in notification.get("destin", []) + for destin in notification.get('destin', []) ] input = SendNotificationInput( - title=notification.get("title"), - content=notification.get("content"), - priority=notification.get("priority", 0), + title=notification.get('title'), + content=notification.get('content'), + priority=notification.get('priority', 0), target=targets, - placeholders=notification.get("placeholders", {}), - icon=notification.get("icon"), + placeholders=notification.get('placeholders', {}), + icon=notification.get('icon'), ) output_list = await self.send_notification.execute(input) return HttpResponse( status_code=202, body={ - "data": [ + 'data': [ { - "status": output.status, - "sent_to": asdict(output.target), + 'status': output.status, + 'sent_to': asdict(output.target), } for output in output_list ] diff --git a/system_notification/infra/http/server/fastapi_http_server.py b/system_notification/infra/http/server/fastapi_http_server.py index eab8024..7001092 100644 --- a/system_notification/infra/http/server/fastapi_http_server.py +++ b/system_notification/infra/http/server/fastapi_http_server.py @@ -7,7 +7,9 @@ from fastapi.responses import JSONResponse from system_notification.domain.protocols import Controller -from system_notification.infra.http.server.helpers.http_request import HttpRequest +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) def get_query_params_as_dict(request: Request) -> Dict[Any, Any]: @@ -31,7 +33,8 @@ def serve(self, port: int = 8000) -> None: def on(self, method: str, url: str, controller: Controller) -> None: async def fastapi_controller( - request: Request, params: Dict[Any, Any] = Depends(get_query_params_as_dict) + request: Request, + params: Dict[Any, Any] = Depends(get_query_params_as_dict), ) -> Any: application_request = HttpRequest( headers={**request.headers}, @@ -44,7 +47,7 @@ async def fastapi_controller( status_code=response.status_code, ) - def asdf(name: str = "any valid name") -> str: + def asdf(name: str = 'any valid name') -> str: return name self._app.add_api_route( diff --git a/system_notification/infra/http/server/flask_http_server.py b/system_notification/infra/http/server/flask_http_server.py index 37a511a..7e3f97a 100644 --- a/system_notification/infra/http/server/flask_http_server.py +++ b/system_notification/infra/http/server/flask_http_server.py @@ -6,7 +6,9 @@ from flask import Flask, Request, Response, request from system_notification.domain.protocols.controller_protocol import Controller -from system_notification.infra.http.server.helpers.http_request import HttpRequest +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) def get_query_params_as_dict(request: Request) -> Dict[Any, Any]: @@ -20,12 +22,12 @@ def get_query_params_as_dict(request: Request) -> Dict[Any, Any]: return query_params -def make_func(f, name: str = ""): +def make_func(f, name: str = ''): """Based on http://stackoverflow.com/a/6528148/190597 (Glenn Maynard)""" g = types.FunctionType( f.__code__, f.__globals__, - name=f"{name}-{f.__name__}", + name=f'{name}-{f.__name__}', argdefs=f.__defaults__, closure=f.__closure__, ) @@ -40,7 +42,7 @@ def __init__(self, debug: bool = False) -> None: self._app = Flask(__name__) def serve(self, port: int = 8000) -> None: - self._app.run(host="0.0.0.0", port=port, debug=self._debug_mode) + self._app.run(host='0.0.0.0', port=port, debug=self._debug_mode) def on(self, method: str, url: str, controller: Controller) -> None: async def view() -> Any: @@ -58,13 +60,13 @@ async def view() -> Any: json.dumps(response.body), status=response.status_code, headers=response.headers, - content_type="application/json", + content_type='application/json', ) view.__name__ = controller.__class__.__name__ self._app.add_url_rule(url, None, view, methods=[method.upper()]) self._app.add_url_rule( - url + "/", + url + '/', None, view, methods=[method.upper()], diff --git a/system_notification/infra/jwt/jose_jwt_adapter.py b/system_notification/infra/jwt/jose_jwt_adapter.py index d71804e..3fddbae 100644 --- a/system_notification/infra/jwt/jose_jwt_adapter.py +++ b/system_notification/infra/jwt/jose_jwt_adapter.py @@ -8,19 +8,19 @@ class JoseJWTAdapter: def __init__( - self, *, secret: str = SETTINGS.get("SECRET", ""), jwt_handler=jwt + self, *, secret: str = SETTINGS.get('SECRET', ''), jwt_handler=jwt ) -> None: self._jwt = jwt_handler self._secret = secret def encode(self, data: Dict[JWT_TYPES, JWT_TYPES]) -> str: - return self._jwt.encode(data, self._secret, algorithm="HS256") + return self._jwt.encode(data, self._secret, algorithm='HS256') def decode( self, data: str, verify_signature: bool = True ) -> Dict[JWT_TYPES, JWT_TYPES]: return self._jwt.decode( - data, self._secret, options={"verify_signature": verify_signature} + data, self._secret, options={'verify_signature': verify_signature} ) def is_valid(self, data: str, verify_signature: bool = True) -> bool: diff --git a/system_notification/infra/notification_handlers/slack_notification_handler.py b/system_notification/infra/notification_handlers/slack_notification_handler.py index af6e13b..487857a 100644 --- a/system_notification/infra/notification_handlers/slack_notification_handler.py +++ b/system_notification/infra/notification_handlers/slack_notification_handler.py @@ -1,38 +1,44 @@ from slack_sdk import WebClient from slack_sdk.errors import SlackApiError -from system_notification.domain.exceptions.notification_error import TargetNotFound -from system_notification.domain.notifications.base_notification import BaseNotification +from system_notification.domain.exceptions.notification_error import ( + TargetNotFound, +) +from system_notification.domain.notifications.base_notification import ( + BaseNotification, +) class SlackNotificationHandler: def __init__(self, token: str, client=WebClient) -> None: self._token = token self._client = client(token=token) - self.target_type = "slack_channel" + self.target_type = 'slack_channel' async def send(self, notification: BaseNotification) -> None: # TODO: find a way to notification make a self validation - assert notification.target, "O target da notificação não pode estar vazio" + assert ( + notification.target + ), 'O target da notificação não pode estar vazio' try: channel = notification.target.target response = self._client.chat_postMessage( - channel=f"#{channel}", + channel=f'#{channel}', text=notification.get_text(), username=notification.title, icon_emoji=notification.icon, ) - assert response["ok"] + assert response['ok'] except SlackApiError as err: raise TargetNotFound( { - "is_sent": False, - "destin": { - "target_type": notification.target.type, - "target": notification.target.target, + 'is_sent': False, + 'destin': { + 'target_type': notification.target.type, + 'target': notification.target.target, }, - "detail": err.response.get("error", "unknown error"), + 'detail': err.response.get('error', 'unknown error'), } ) - notification.status = "sent" + notification.status = 'sent' notification.mark_as_sent() diff --git a/tests/application/test_send_notification_usecase.py b/tests/application/test_send_notification_usecase.py index ff0003c..b6af4d9 100644 --- a/tests/application/test_send_notification_usecase.py +++ b/tests/application/test_send_notification_usecase.py @@ -10,19 +10,25 @@ from system_notification.domain.notification_factory_caller import ( NotificationFactoryCaller, ) -from system_notification.domain.notifications.base_notification import BaseNotification +from system_notification.domain.notifications.base_notification import ( + BaseNotification, +) from system_notification.domain.notifications.notification_target import ( NotificationTarget, ) -from system_notification.domain.protocols.notification_protocol import Notification -from system_notification.domain.protocols.notification_sender import NotificationSender +from system_notification.domain.protocols.notification_protocol import ( + Notification, +) +from system_notification.domain.protocols.notification_sender import ( + NotificationSender, +) class FakeNotificationSender: def __init__(self) -> None: self.send_is_called: bool = False self.notification: Notification - self.target_type: str = "testing" + self.target_type: str = 'testing' self.send_is_called_count: int = 0 async def send(self, notification: Notification) -> None: @@ -34,23 +40,31 @@ async def send(self, notification: Notification) -> None: class FakeTestingFactory: def __init__(self, sender: NotificationSender) -> None: - self.target_type: str = "testing" + self.target_type: str = 'testing' self.sender = sender def make_notificaton( - self, title: str, content: str, target: str, priority: Literal[0, 1, 2, 3] = 0 + self, + title: str, + content: str, + target: str, + priority: Literal[0, 1, 2, 3] = 0, ) -> Notification: - return BaseNotification(title=title, content=content, priority=priority) + return BaseNotification( + title=title, content=content, priority=priority + ) def make_sender(self) -> NotificationSender: return self.sender -async def test_ensure_execute_method_can_handle_list_of_targets_to_the_same_notification() -> None: +async def test_ensure_execute_method_can_handle_list_of_targets_to_the_same_notification() -> ( + None +): input = SendNotificationInput( - title="Any Title", - content="Any Content", - target=[NotificationTarget("testing", "#123")], + title='Any Title', + content='Any Content', + target=[NotificationTarget('testing', '#123')], ) sender = FakeNotificationSender() notification_factory_caller = NotificationFactoryCaller() diff --git a/tests/domain/decorators/test_auth_decorator.py b/tests/domain/decorators/test_auth_decorator.py index 769eb1f..58bcfe1 100644 --- a/tests/domain/decorators/test_auth_decorator.py +++ b/tests/domain/decorators/test_auth_decorator.py @@ -1,9 +1,13 @@ from system_notification.domain.decorators import JWTAuthControllerDecorator -from system_notification.infra.http.server.helpers.http_request import HttpRequest -from system_notification.infra.http.server.helpers.http_response import HttpResponse +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) +from system_notification.infra.http.server.helpers.http_response import ( + HttpResponse, +) from system_notification.infra.jwt import JoseJWTAdapter -jwt_adapter = JoseJWTAdapter(secret="secret") +jwt_adapter = JoseJWTAdapter(secret='secret') jwt_auth_validation = JWTAuthControllerDecorator(jwt_adapter=jwt_adapter) @@ -16,21 +20,27 @@ async def handle(self, request: HttpRequest) -> HttpResponse: return self._response -async def test_jwt_auth_validation_decorator_returns_controller_response_when_client_is_authenticated() -> None: - jwt_token = jwt_adapter.encode({"email": "mail@mail.com", "user": "John Duo"}) +async def test_jwt_auth_validation_decorator_returns_controller_response_when_client_is_authenticated() -> ( + None +): + jwt_token = jwt_adapter.encode( + {'email': 'mail@mail.com', 'user': 'John Duo'} + ) response = HttpResponse( status_code=200, body={}, ) controller = FakeController(response) output_response = await controller.handle( - HttpRequest(headers={"authorization": f"Bearer {jwt_token}"}) + HttpRequest(headers={'authorization': f'Bearer {jwt_token}'}) ) assert jwt_auth_validation.authenticated assert response is output_response -async def test_jwt_auth_validation_decorator_returns_status_401_when_client_is_not_authenticated() -> None: +async def test_jwt_auth_validation_decorator_returns_status_401_when_client_is_not_authenticated() -> ( + None +): jwt_auth_validation = JWTAuthControllerDecorator(jwt_adapter=jwt_adapter) response = HttpResponse( status_code=200, @@ -39,12 +49,12 @@ async def test_jwt_auth_validation_decorator_returns_status_401_when_client_is_n controller = FakeController(response) output_response = await controller.handle( - HttpRequest(headers={"authorization": f"Bearer INVALID TOKEN"}) + HttpRequest(headers={'authorization': f'Bearer INVALID TOKEN'}) ) assert jwt_auth_validation.authenticated == False assert output_response.status_code == 401 assert output_response.body == { - "message": "Not Authorized", - "detail": "invalid API token", + 'message': 'Not Authorized', + 'detail': 'invalid API token', } diff --git a/tests/domain/notifications/test_base_notification.py b/tests/domain/notifications/test_base_notification.py index 6f277e5..763d1b4 100644 --- a/tests/domain/notifications/test_base_notification.py +++ b/tests/domain/notifications/test_base_notification.py @@ -3,15 +3,22 @@ from system_notification.domain import BaseNotification from system_notification.domain.notifications import NotificationTarget -title = "My Notification" -vars = {"full_name": "John Duo"} -content = "Hello, {full_name}! This is a basic notification" -final_text = "Hello, John Duo! This is a basic notification" +title = 'My Notification' +vars = {'full_name': 'John Duo'} +content = 'Hello, {full_name}! This is a basic notification' +final_text = 'Hello, John Duo! This is a basic notification' @pytest.mark.parametrize( - ["priority", "expected_priority"], - [("Any value different from int", 0), (None, 0), (0, 0), (1, 1), (2, 2), (3, 3)], + ['priority', 'expected_priority'], + [ + ('Any value different from int', 0), + (None, 0), + (0, 0), + (1, 1), + (2, 2), + (3, 3), + ], ) async def test_instatiation_params(priority, expected_priority) -> None: sut = BaseNotification(title=title, content=content, priority=priority) @@ -20,14 +27,14 @@ async def test_instatiation_params(priority, expected_priority) -> None: assert sut.is_sent == False assert sut.priority == expected_priority assert sut.vars == {} - assert sut.status == "filled" + assert sut.status == 'filled' with pytest.raises(NotImplementedError): assert sut.is_scheduled == False @pytest.mark.parametrize( - ["status", "expected_is_sent"], - [("queued", True), ("sent", True), ("filled", False)], + ['status', 'expected_is_sent'], + [('queued', True), ('sent', True), ('filled', False)], ) async def test_ensure_is_sent_property_follows_status_property( status, expected_is_sent @@ -49,7 +56,9 @@ async def test_get_text_should_replace_string_placeholders_using_vars_values() - assert sut.get_text() == final_text -async def test_get_text_should_not_replace_string_placeholders_when_apply_vars_flag_is_false() -> None: +async def test_get_text_should_not_replace_string_placeholders_when_apply_vars_flag_is_false() -> ( + None +): sut = BaseNotification(title=title, content=content) sut.set_vars(vars) assert sut.get_text(apply_vars=False) == content @@ -61,8 +70,8 @@ async def test_get_text_should_return_the_raw_text_if_no_vars_was_setted() -> No async def test_add_target() -> None: - target = NotificationTarget(_type="slack_channel", _target="tech_logs") - target_not_in_sut = NotificationTarget(_type="any", _target="any") + target = NotificationTarget(_type='slack_channel', _target='tech_logs') + target_not_in_sut = NotificationTarget(_type='any', _target='any') sut = BaseNotification(title=title, content=content) sut.add_target(target) assert target in sut diff --git a/tests/domain/notifications/test_notification_target.py b/tests/domain/notifications/test_notification_target.py index a43fc27..60fd70e 100644 --- a/tests/domain/notifications/test_notification_target.py +++ b/tests/domain/notifications/test_notification_target.py @@ -2,6 +2,6 @@ async def test_notification_target_instantiation_parmas() -> None: - sut = NotificationTarget(_type="slack_channel", _target="tech_logs") - assert sut.target == "tech_logs" - assert sut.type == "slack_channel" + sut = NotificationTarget(_type='slack_channel', _target='tech_logs') + assert sut.target == 'tech_logs' + assert sut.type == 'slack_channel' diff --git a/tests/domain/test_factory_caller.py b/tests/domain/test_factory_caller.py index fe6d05b..15791fb 100644 --- a/tests/domain/test_factory_caller.py +++ b/tests/domain/test_factory_caller.py @@ -1,10 +1,14 @@ import pytest -from system_notification.domain.exceptions.notification_error import TargetNotFound +from system_notification.domain.exceptions.notification_error import ( + TargetNotFound, +) from system_notification.domain.notification_factory_caller import ( NotificationFactoryCaller, ) -from system_notification.domain.notifications.base_notification import BaseNotification +from system_notification.domain.notifications.base_notification import ( + BaseNotification, +) from system_notification.domain.notifications.factories.slack_notification_factory import ( SlackNotificationFactory, ) @@ -23,35 +27,35 @@ async def test_ensure_has_no_factory_registered_at_instantiation_time(): async def test_registering_a_factory(): sut = NotificationFactoryCaller() - sut.add_factory(SlackNotificationFactory(slack_token="any valid token")) + sut.add_factory(SlackNotificationFactory(slack_token='any valid token')) assert len(sut) == 1 async def test_raises_TargetNotFound_at_get_an_unknown_sender(): sut = NotificationFactoryCaller() - sut.add_factory(SlackNotificationFactory(slack_token="any valid token")) + sut.add_factory(SlackNotificationFactory(slack_token='any valid token')) with pytest.raises(TargetNotFound): await sut.get_sender( - NotificationTarget(_type="any unknown target", _target="123") + NotificationTarget(_type='any unknown target', _target='123') ) async def test_can_recover_a_sender(): sut = NotificationFactoryCaller() - sut.add_factory(SlackNotificationFactory(slack_token="any valid token")) + sut.add_factory(SlackNotificationFactory(slack_token='any valid token')) sender = await sut.get_sender( - NotificationTarget(_type="slack_channel", _target="123") + NotificationTarget(_type='slack_channel', _target='123') ) assert type(sender) is SlackNotificationHandler async def test_make_a_notification(): sut = NotificationFactoryCaller() - sut.add_factory(SlackNotificationFactory(slack_token="any valid token")) + sut.add_factory(SlackNotificationFactory(slack_token='any valid token')) notification = await sut.get_notification( - title="title", - content="content", - destin=NotificationTarget(_type="slack_channel", _target="123"), + title='title', + content='content', + destin=NotificationTarget(_type='slack_channel', _target='123'), ) assert type(notification) is BaseNotification diff --git a/tests/infra/http/client/test_aiohttp_adapter.py b/tests/infra/http/client/test_aiohttp_adapter.py index 0ad9e26..077aa38 100644 --- a/tests/infra/http/client/test_aiohttp_adapter.py +++ b/tests/infra/http/client/test_aiohttp_adapter.py @@ -2,7 +2,9 @@ from typing import Any from unittest.mock import AsyncMock -from system_notification.infra.http.client.aiohttp_adapter import AioHttpAdapter +from system_notification.infra.http.client.aiohttp_adapter import ( + AioHttpAdapter, +) class ClientSessionFake: @@ -23,7 +25,7 @@ def request(self, *args, **kwargs) -> Any: async def read(self) -> Any: self.read_is_called = True - async def __aenter__(self) -> "ClientSessionFake": + async def __aenter__(self) -> 'ClientSessionFake': return self async def __aexit__(self, *args, **kwargs) -> None: @@ -39,7 +41,7 @@ async def test_make_async_returns_an_object_where_json_mehod_is_a_couroutine() - async def test_get_method_calls_make_request() -> None: session = ClientSessionFake() sut = AioHttpAdapter(client_session=session) - output = await sut.get("https://companyhero.com") + output = await sut.get('https://companyhero.com') assert session.json_is_called assert session.request_is_called assert session.read_is_called @@ -49,7 +51,7 @@ async def test_get_method_calls_make_request() -> None: async def test_post_method_calls_make_request() -> None: client_session = ClientSessionFake() sut = AioHttpAdapter(client_session=client_session) - await sut.post("https://companyhero.com", json={}) + await sut.post('https://companyhero.com', json={}) assert client_session.json_is_called assert client_session.request_is_called assert client_session.read_is_called @@ -58,7 +60,7 @@ async def test_post_method_calls_make_request() -> None: async def test_post_method_calls_make_request_with_files() -> None: client_session = ClientSessionFake() sut = AioHttpAdapter(client_session=client_session) - await sut.post("https://companyhero.com", json={}, files={"anyfile": b""}) + await sut.post('https://companyhero.com', json={}, files={'anyfile': b''}) assert client_session.json_is_called assert client_session.request_is_called assert client_session.read_is_called @@ -67,7 +69,7 @@ async def test_post_method_calls_make_request_with_files() -> None: async def test_put_method_calls_make_request() -> None: client_session = ClientSessionFake() sut = AioHttpAdapter(client_session=client_session) - await sut.put("https://companyhero.com", json={}) + await sut.put('https://companyhero.com', json={}) assert client_session.json_is_called assert client_session.request_is_called assert client_session.read_is_called @@ -76,7 +78,7 @@ async def test_put_method_calls_make_request() -> None: async def test_patch_method_calls_make_request() -> None: client_session = ClientSessionFake() sut = AioHttpAdapter(client_session=client_session) - await sut.patch("https://companyhero.com", json={}) + await sut.patch('https://companyhero.com', json={}) assert client_session.json_is_called assert client_session.request_is_called assert client_session.read_is_called @@ -85,7 +87,7 @@ async def test_patch_method_calls_make_request() -> None: async def test_delete_method_calls_make_request() -> None: client_session = ClientSessionFake() sut = AioHttpAdapter(client_session=client_session) - await sut.delete("https://companyhero.com") + await sut.delete('https://companyhero.com') assert client_session.json_is_called assert client_session.request_is_called assert client_session.read_is_called diff --git a/tests/infra/http/controller/api_notification_serializer.py b/tests/infra/http/controller/api_notification_serializer.py index bae2331..7c69418 100644 --- a/tests/infra/http/controller/api_notification_serializer.py +++ b/tests/infra/http/controller/api_notification_serializer.py @@ -15,7 +15,7 @@ class NotificationSchema(pydantic.BaseModel): destin: List[DestinSchema] priority: int = 0 placeholders: Dict[str, str] = {} - icon: str = "" + icon: str = '' class APINotificationSchema(pydantic.BaseModel): @@ -26,21 +26,27 @@ class ApiNotificationSerializer: def __init__(self) -> None: self.errors = {} - def from_raw(self, input: Union[str, bytes]) -> Optional[APINotificationSchema]: + def from_raw( + self, input: Union[str, bytes] + ) -> Optional[APINotificationSchema]: try: return APINotificationSchema.parse_raw(input) except pydantic.error_wrappers.ValidationError as err: self.errors = json.loads(err.json()) return None - def from_dict(self, input: Dict[str, Any]) -> Optional[APINotificationSchema]: + def from_dict( + self, input: Dict[str, Any] + ) -> Optional[APINotificationSchema]: try: return APINotificationSchema.parse_obj(input) except pydantic.error_wrappers.ValidationError as err: self.errors = json.loads(err.json()) return None - def to_dict(self, input: APINotificationSchema) -> Optional[APINotificationSchema]: + def to_dict( + self, input: APINotificationSchema + ) -> Optional[APINotificationSchema]: try: return input.json() except pydantic.error_wrappers.ValidationError as err: diff --git a/tests/infra/http/controller/test_send_notification_controller.py b/tests/infra/http/controller/test_send_notification_controller.py index 0587759..80a1117 100644 --- a/tests/infra/http/controller/test_send_notification_controller.py +++ b/tests/infra/http/controller/test_send_notification_controller.py @@ -3,7 +3,9 @@ from system_notification.application.send_notification_usecase.send_notification_usecase import ( SendNotificationInput, ) -from system_notification.domain.exceptions.notification_error import TargetNotFound +from system_notification.domain.exceptions.notification_error import ( + TargetNotFound, +) from system_notification.domain.notifications.notification_target import ( NotificationTarget, ) @@ -11,21 +13,23 @@ from system_notification.infra.http.controller.send_notification_controller import ( SendNotificationController, ) -from system_notification.infra.http.server.helpers.http_request import HttpRequest +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) payload = { - "data": { - "title": "What is Lorem Ipsum?", - "content": "Lorem Ipsum is simply dummy text of the printing and typesetting industry. ", - "destin": [{"type": "slack_channel", "target": "tech_logs"}], - "priority": 0, + 'data': { + 'title': 'What is Lorem Ipsum?', + 'content': 'Lorem Ipsum is simply dummy text of the printing and typesetting industry. ', + 'destin': [{'type': 'slack_channel', 'target': 'tech_logs'}], + 'priority': 0, } } def undecorate_controller(instance: Controller) -> Controller: - undecorated_method = getattr(instance.handle, "__wrapped__") - setattr(instance, "handle", undecorated_method) + undecorated_method = getattr(instance.handle, '__wrapped__') + setattr(instance, 'handle', undecorated_method) return instance @@ -68,7 +72,9 @@ async def test_ensure_return_400_when_serialize_has_errors() -> None: serializer=serializer, ) undecorate_controller(sut) - response = await sut.handle(sut, HttpRequest(headers={}, body={}, params={})) + response = await sut.handle( + sut, HttpRequest(headers={}, body={}, params={}) + ) assert response.status_code == 400 @@ -86,9 +92,11 @@ async def test_ensure_usecase_is_called_with_right_params() -> None: await sut.handle(sut, HttpRequest(headers={}, body=payload, params={})) send_notifcation_usecase.execute.assert_called_with( SendNotificationInput( - title="What is Lorem Ipsum?", - content="Lorem Ipsum is simply dummy text of the printing and typesetting industry. ", - target=[NotificationTarget(_type="slack_channel", _target="tech_logs")], + title='What is Lorem Ipsum?', + content='Lorem Ipsum is simply dummy text of the printing and typesetting industry. ', + target=[ + NotificationTarget(_type='slack_channel', _target='tech_logs') + ], priority=0, placeholders={}, icon=None, @@ -111,9 +119,11 @@ def execute(*args, **kwargs): await sut.handle(sut, HttpRequest(headers={}, body=payload, params={})) send_notifcation_usecase.execute.assert_called_with( SendNotificationInput( - title="What is Lorem Ipsum?", - content="Lorem Ipsum is simply dummy text of the printing and typesetting industry. ", - target=[NotificationTarget(_type="slack_channel", _target="tech_logs")], + title='What is Lorem Ipsum?', + content='Lorem Ipsum is simply dummy text of the printing and typesetting industry. ', + target=[ + NotificationTarget(_type='slack_channel', _target='tech_logs') + ], priority=0, placeholders={}, icon=None, diff --git a/tests/infra/jwt/test_jose_jw_adapter.py b/tests/infra/jwt/test_jose_jw_adapter.py index 2d79b25..1ca0307 100644 --- a/tests/infra/jwt/test_jose_jw_adapter.py +++ b/tests/infra/jwt/test_jose_jw_adapter.py @@ -2,30 +2,30 @@ async def test_instantiation_parms() -> None: - secret = "any valid secret" + secret = 'any valid secret' sut = JoseJWTAdapter(secret=secret) assert sut._secret is secret async def test_encode_decode_methods() -> None: - secret = "any valid secret" + secret = 'any valid secret' sut = JoseJWTAdapter(secret=secret) - decoded_data = {"type": "str", "content": "Any value"} + decoded_data = {'type': 'str', 'content': 'Any value'} encoded_data = sut.encode(decoded_data) assert sut.decode(encoded_data) == decoded_data async def test_is_valid_should_return_true_when_jwt_is_valid() -> None: - secret = "any valid secret" + secret = 'any valid secret' sut = JoseJWTAdapter(secret=secret) - decoded_data = {"type": "str", "content": "Any value"} + decoded_data = {'type': 'str', 'content': 'Any value'} encoded_data = sut.encode(decoded_data) assert sut.is_valid(encoded_data) async def test_is_valid_should_return_false_when_jwt_is_invalid() -> None: - secret = "any valid secret" + secret = 'any valid secret' sut = JoseJWTAdapter(secret=secret) - decoded_data = {"type": "str", "content": "Any value"} + decoded_data = {'type': 'str', 'content': 'Any value'} encoded_data = sut.encode(decoded_data) - assert not sut.is_valid(encoded_data + "asdf") + assert not sut.is_valid(encoded_data + 'asdf') diff --git a/tests/infra/notifications_handlers/test_slack_notification_handler.py b/tests/infra/notifications_handlers/test_slack_notification_handler.py index 5bdf151..5db5867 100644 --- a/tests/infra/notifications_handlers/test_slack_notification_handler.py +++ b/tests/infra/notifications_handlers/test_slack_notification_handler.py @@ -3,16 +3,22 @@ import pytest from slack_sdk.errors import SlackApiError -from system_notification.domain.exceptions.notification_error import TargetNotFound +from system_notification.domain.exceptions.notification_error import ( + TargetNotFound, +) from system_notification.domain.notifications import BaseNotification from system_notification.domain.notifications.notification_target import ( NotificationTarget, ) -from system_notification.infra.notification_handlers import SlackNotificationHandler +from system_notification.infra.notification_handlers import ( + SlackNotificationHandler, +) -token = "ANY VALID SLACK API TOKEN" -notification = BaseNotification(title="My Notification", content="Thats works!") -notification.add_target(NotificationTarget("slack_channel", "tech_logs")) +token = 'ANY VALID SLACK API TOKEN' +notification = BaseNotification( + title='My Notification', content='Thats works!' +) +notification.add_target(NotificationTarget('slack_channel', 'tech_logs')) class SlackFakeClient: @@ -31,7 +37,7 @@ def chat_postMessage(self, *args, **kwargs) -> Dict[str, Any]: message=self.raises_with[0], response=self.raises_with[1] ) self.chat_postMessage_is_called = True - return {"ok": self.as_success} + return {'ok': self.as_success} def __call__(self, *args: Any, **kwds: Any) -> Any: return self @@ -53,7 +59,7 @@ async def test_ensure_send_method_calls_chat_postMessage_from_client(): async def test_when_the_server_responded_with_an_error_the_handler_should_raises_RuntimeError(): - client = SlackFakeClient(as_success=True, raises_with=("token_error", {})) + client = SlackFakeClient(as_success=True, raises_with=('token_error', {})) sut = SlackNotificationHandler(token=token, client=client) with pytest.raises(TargetNotFound) as e: await sut.send(notification) diff --git a/tests/infra/server/test_fastapi_http_server.py b/tests/infra/server/test_fastapi_http_server.py index 298c67d..1ccf489 100644 --- a/tests/infra/server/test_fastapi_http_server.py +++ b/tests/infra/server/test_fastapi_http_server.py @@ -5,9 +5,15 @@ from fastapi.testclient import TestClient from system_notification.domain.protocols.controller_protocol import HttpServer -from system_notification.infra.http.server.fastapi_http_server import FastApiHttpServer -from system_notification.infra.http.server.helpers.http_request import HttpRequest -from system_notification.infra.http.server.helpers.http_response import HttpResponse +from system_notification.infra.http.server.fastapi_http_server import ( + FastApiHttpServer, +) +from system_notification.infra.http.server.helpers.http_request import ( + HttpRequest, +) +from system_notification.infra.http.server.helpers.http_response import ( + HttpResponse, +) class FakeServer: @@ -25,8 +31,8 @@ def __init__( self, http_server: HttpServer, response_dict: Dict[str, str], - endpoint: str = "/", - method: str = "GET", + endpoint: str = '/', + method: str = 'GET', status_code: int = 200, ) -> None: self.endpoint = endpoint @@ -36,7 +42,9 @@ def __init__( self.status_code = status_code async def handle(self, request: HttpRequest) -> HttpResponse: - return HttpResponse(status_code=self.status_code, body=self.response_dict) + return HttpResponse( + status_code=self.status_code, body=self.response_dict + ) async def test_serve_calls_run_with_right_params() -> None: @@ -51,24 +59,26 @@ async def test_register_controller_calls_add_api_route_with_right_params() -> No controller = AsyncMock() server = FakeServer() sut = FastApiHttpServer(server=server, app=fake_app) - sut.on("GET", "/", controller=controller) + sut.on('GET', '/', controller=controller) fake_app.add_api_route.assert_called() -async def test_ensure_fastapi_can_handle_controller_post_inside_its_inner_view() -> None: +async def test_ensure_fastapi_can_handle_controller_post_inside_its_inner_view() -> ( + None +): app = FastAPI() sut = FastApiHttpServer(app=app) controller = FakeController( http_server=sut, - response_dict={"data": "testing"}, + response_dict={'data': 'testing'}, status_code=200, - endpoint="/testing", - method="POST", + endpoint='/testing', + method='POST', ) client = TestClient(app) response = client.post( - controller.endpoint + "?name=John&name=Duo&email=mail@mail.com", - json={"post_data": {}}, + controller.endpoint + '?name=John&name=Duo&email=mail@mail.com', + json={'post_data': {}}, ) assert response.status_code == controller.status_code assert response.json() == controller.response_dict @@ -79,14 +89,14 @@ async def test_ensure_fastapi_can_handle_controller_get_inside_its_inner_view() sut = FastApiHttpServer(app=app) controller = FakeController( http_server=sut, - response_dict={"data": "testing"}, + response_dict={'data': 'testing'}, status_code=200, - endpoint="/testing", - method="GET", + endpoint='/testing', + method='GET', ) client = TestClient(app) response = client.get( - controller.endpoint + "?name=John&name=Duo&email=mail@mail.com", + controller.endpoint + '?name=John&name=Duo&email=mail@mail.com', ) assert response.status_code == controller.status_code assert response.json() == controller.response_dict