From f2ab02804b366b93b0ed4aed1210a9a509f51ef0 Mon Sep 17 00:00:00 2001 From: Cayo Puigdefabregas Date: Thu, 17 Mar 2022 13:13:15 +0100 Subject: [PATCH] new userclient from flask instead of teal --- ereuse_devicehub/client.py | 329 +++++++++++++++++++++---------------- tests/conftest.py | 8 +- 2 files changed, 193 insertions(+), 144 deletions(-) diff --git a/ereuse_devicehub/client.py b/ereuse_devicehub/client.py index 3ddf94ff..36d3bd49 100644 --- a/ereuse_devicehub/client.py +++ b/ereuse_devicehub/client.py @@ -1,10 +1,11 @@ from inspect import isclass from typing import Dict, Iterable, Type, Union -from flask.testing import FlaskClient -from flask_wtf.csrf import generate_csrf from ereuse_utils.test import JSON, Res -from teal.client import Client as TealClient, Query, Status +from flask.testing import FlaskClient +from flask_wtf.csrf import generate_csrf +from teal.client import Client as TealClient +from teal.client import Query, Status from werkzeug.exceptions import HTTPException from ereuse_devicehub.resources import models, schemas @@ -15,110 +16,156 @@ ResourceLike = Union[Type[Union[models.Thing, schemas.Thing]], str] class Client(TealClient): """A client suited for Devicehub main usage.""" - def __init__(self, application, - response_wrapper=None, - use_cookies=False, - allow_subdomain_redirects=False): - super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects) + def __init__( + self, + application, + response_wrapper=None, + use_cookies=False, + allow_subdomain_redirects=False, + ): + super().__init__( + application, response_wrapper, use_cookies, allow_subdomain_redirects + ) - def open(self, - uri: str, - res: ResourceLike = None, - status: Status = 200, - query: Query = tuple(), - accept=JSON, - content_type=JSON, - item=None, - headers: dict = None, - token: str = None, - **kw) -> Res: + def open( + self, + uri: str, + res: ResourceLike = None, + status: Status = 200, + query: Query = tuple(), + accept=JSON, + content_type=JSON, + item=None, + headers: dict = None, + token: str = None, + **kw, + ) -> Res: if isclass(res) and issubclass(res, (models.Thing, schemas.Thing)): res = res.t - return super().open(uri, res, status, query, accept, content_type, item, headers, token, - **kw) + return super().open( + uri, res, status, query, accept, content_type, item, headers, token, **kw + ) - def get(self, - uri: str = '', - res: ResourceLike = None, - query: Query = tuple(), - status: Status = 200, - item: Union[int, str] = None, - accept: str = JSON, - headers: dict = None, - token: str = None, - **kw) -> Res: + def get( + self, + uri: str = '', + res: ResourceLike = None, + query: Query = tuple(), + status: Status = 200, + item: Union[int, str] = None, + accept: str = JSON, + headers: dict = None, + token: str = None, + **kw, + ) -> Res: return super().get(uri, res, query, status, item, accept, headers, token, **kw) - def post(self, - data: str or dict, - uri: str = '', - res: ResourceLike = None, - query: Query = tuple(), - status: Status = 201, - content_type: str = JSON, - accept: str = JSON, - headers: dict = None, - token: str = None, - **kw) -> Res: - return super().post(data, uri, res, query, status, content_type, accept, headers, token, - **kw) + def post( + self, + data: str or dict, + uri: str = '', + res: ResourceLike = None, + query: Query = tuple(), + status: Status = 201, + content_type: str = JSON, + accept: str = JSON, + headers: dict = None, + token: str = None, + **kw, + ) -> Res: + return super().post( + data, uri, res, query, status, content_type, accept, headers, token, **kw + ) - def patch(self, - data: str or dict, - uri: str = '', - res: ResourceLike = None, - query: Query = tuple(), - item: Union[int, str] = None, - status: Status = 200, - content_type: str = JSON, - accept: str = JSON, - headers: dict = None, - token: str = None, - **kw) -> Res: - return super().patch(data, uri, res, query, item, status, content_type, accept, token, - headers, **kw) + def patch( + self, + data: str or dict, + uri: str = '', + res: ResourceLike = None, + query: Query = tuple(), + item: Union[int, str] = None, + status: Status = 200, + content_type: str = JSON, + accept: str = JSON, + headers: dict = None, + token: str = None, + **kw, + ) -> Res: + return super().patch( + data, + uri, + res, + query, + item, + status, + content_type, + accept, + token, + headers, + **kw, + ) - def put(self, - data: str or dict, - uri: str = '', - res: ResourceLike = None, - query: Query = tuple(), - item: Union[int, str] = None, - status: Status = 201, - content_type: str = JSON, - accept: str = JSON, - headers: dict = None, - token: str = None, - **kw) -> Res: - return super().put(data, uri, res, query, item, status, content_type, accept, token, - headers, **kw) + def put( + self, + data: str or dict, + uri: str = '', + res: ResourceLike = None, + query: Query = tuple(), + item: Union[int, str] = None, + status: Status = 201, + content_type: str = JSON, + accept: str = JSON, + headers: dict = None, + token: str = None, + **kw, + ) -> Res: + return super().put( + data, + uri, + res, + query, + item, + status, + content_type, + accept, + token, + headers, + **kw, + ) - def delete(self, - uri: str = '', - res: ResourceLike = None, - query: Query = tuple(), - status: Status = 204, - item: Union[int, str] = None, - accept: str = JSON, - headers: dict = None, - token: str = None, - **kw) -> Res: - return super().delete(uri, res, query, status, item, accept, headers, token, **kw) + def delete( + self, + uri: str = '', + res: ResourceLike = None, + query: Query = tuple(), + status: Status = 204, + item: Union[int, str] = None, + accept: str = JSON, + headers: dict = None, + token: str = None, + **kw, + ) -> Res: + return super().delete( + uri, res, query, status, item, accept, headers, token, **kw + ) def login(self, email: str, password: str): assert isinstance(email, str) assert isinstance(password, str) - return self.post({'email': email, 'password': password}, '/users/login/', status=200) + return self.post( + {'email': email, 'password': password}, '/users/login/', status=200 + ) - def get_many(self, - res: ResourceLike, - resources: Iterable[Union[dict, int]], - key: str = None, - **kw) -> Iterable[Union[Dict[str, object], str]]: + def get_many( + self, + res: ResourceLike, + resources: Iterable[Union[dict, int]], + key: str = None, + **kw, + ) -> Iterable[Union[Dict[str, object], str]]: """Like :meth:`.get` but with many resources.""" return ( - self.get(res=res, item=r[key] if key else r, **kw)[0] - for r in resources + self.get(res=res, item=r[key] if key else r, **kw)[0] for r in resources ) @@ -128,30 +175,47 @@ class UserClient(Client): It will automatically perform login on the first request. """ - def __init__(self, application, - email: str, - password: str, - response_wrapper=None, - use_cookies=False, - allow_subdomain_redirects=False): - super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects) + def __init__( + self, + application, + email: str, + password: str, + response_wrapper=None, + use_cookies=False, + allow_subdomain_redirects=False, + ): + super().__init__( + application, response_wrapper, use_cookies, allow_subdomain_redirects + ) self.email = email # type: str self.password = password # type: str self.user = None # type: dict - def open(self, - uri: str, - res: ResourceLike = None, - status: int or HTTPException = 200, - query: Query = tuple(), - accept=JSON, - content_type=JSON, - item=None, - headers: dict = None, - token: str = None, - **kw) -> Res: - return super().open(uri, res, status, query, accept, content_type, item, headers, - self.user['token'] if self.user else token, **kw) + def open( + self, + uri: str, + res: ResourceLike = None, + status: int or HTTPException = 200, + query: Query = tuple(), + accept=JSON, + content_type=JSON, + item=None, + headers: dict = None, + token: str = None, + **kw, + ) -> Res: + return super().open( + uri, + res, + status, + query, + accept, + content_type, + item, + headers, + self.user['token'] if self.user else token, + **kw, + ) # noinspection PyMethodOverriding def login(self): @@ -161,13 +225,15 @@ class UserClient(Client): class UserClientFlask: - - def __init__(self, application, - email: str, - password: str, - response_wrapper=None, - use_cookies=True, - follow_redirects=True): + def __init__( + self, + application, + email: str, + password: str, + response_wrapper=None, + use_cookies=True, + follow_redirects=True, + ): self.email = email self.password = password self.follow_redirects = follow_redirects @@ -181,38 +247,25 @@ class UserClientFlask: 'password': password, 'csrf_token': generate_csrf(), } - body, status, headers = self.client.post('/login/', data=data, follow_redirects=True) + body, status, headers = self.client.post( + '/login/', data=data, follow_redirects=True + ) self.headers = headers body = next(body).decode("utf-8") assert "Unassgined" in body - def get(self, - uri='', - data=None, - follow_redirects=True, - **kw): + def get(self, uri='', data=None, follow_redirects=True, **kw): body, status, headers = self.client.get( - uri, - data=data, - follow_redirects=follow_redirects, - headers=self.headers + uri, data=data, follow_redirects=follow_redirects, headers=self.headers ) body = next(body).decode("utf-8") return (body, status) - def post(self, - uri='', - data=None, - follow_redirects=True, - **kw): + def post(self, uri='', data=None, follow_redirects=True, **kw): - import pdb; pdb.set_trace() body, status, headers = self.client.post( - uri, - data=data, - follow_redirects=follow_redirects, - headers=self.headers + uri, data=data, follow_redirects=follow_redirects, headers=self.headers ) body = next(body).decode("utf-8") return (body, status) diff --git a/tests/conftest.py b/tests/conftest.py index e571e96b..1bd1ec71 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -142,9 +142,7 @@ def user3(app: Devicehub) -> UserClientFlask: with app.app_context(): password = 'foo' user = create_user(password=password) - client = UserClientFlask( - app, user.email, password - ) + client = UserClientFlask(app, user.email, password) return client @@ -155,9 +153,7 @@ def user4(app: Devicehub) -> UserClient: password = 'foo' email = 'foo2@foo.com' user = create_user(email=email, password=password) - client = UserClientFlask( - app, user.email, password - ) + client = UserClientFlask(app, user.email, password) return client