trustchain-oc1-orchestral
/
devicehub-teal
Archived
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

new userclient from flask instead of teal

This commit is contained in:
Cayo Puigdefabregas 2022-03-17 13:13:15 +01:00
parent ec99ad22b5
commit f2ab02804b
2 changed files with 193 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

329
ereuse_devicehub/client.py
View File

@ -1,10 +1,11 @@
from inspect import isclass from inspect import isclass
from typing import Dict, Iterable, Type, Union 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 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 werkzeug.exceptions import HTTPException
from ereuse_devicehub.resources import models, schemas from ereuse_devicehub.resources import models, schemas
@ -15,110 +16,156 @@ ResourceLike = Union[Type[Union[models.Thing, schemas.Thing]], str]
class Client(TealClient): class Client(TealClient):
"""A client suited for Devicehub main usage.""" """A client suited for Devicehub main usage."""
def __init__(self, application, def __init__(
response_wrapper=None, self,
use_cookies=False, application,
allow_subdomain_redirects=False): response_wrapper=None,
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects) use_cookies=False,
allow_subdomain_redirects=False,
):
super().__init__(
application, response_wrapper, use_cookies, allow_subdomain_redirects
)
def open(self, def open(
uri: str, self,
res: ResourceLike = None, uri: str,
status: Status = 200, res: ResourceLike = None,
query: Query = tuple(), status: Status = 200,
accept=JSON, query: Query = tuple(),
content_type=JSON, accept=JSON,
item=None, content_type=JSON,
headers: dict = None, item=None,
token: str = None, headers: dict = None,
**kw) -> Res: token: str = None,
**kw,
) -> Res:
if isclass(res) and issubclass(res, (models.Thing, schemas.Thing)): if isclass(res) and issubclass(res, (models.Thing, schemas.Thing)):
res = res.t res = res.t
return super().open(uri, res, status, query, accept, content_type, item, headers, token, return super().open(
**kw) uri, res, status, query, accept, content_type, item, headers, token, **kw
)
def get(self, def get(
uri: str = '', self,
res: ResourceLike = None, uri: str = '',
query: Query = tuple(), res: ResourceLike = None,
status: Status = 200, query: Query = tuple(),
item: Union[int, str] = None, status: Status = 200,
accept: str = JSON, item: Union[int, str] = None,
headers: dict = None, accept: str = JSON,
token: str = None, headers: dict = None,
**kw) -> Res: token: str = None,
**kw,
) -> Res:
return super().get(uri, res, query, status, item, accept, headers, token, **kw) return super().get(uri, res, query, status, item, accept, headers, token, **kw)
def post(self, def post(
data: str or dict, self,
uri: str = '', data: str or dict,
res: ResourceLike = None, uri: str = '',
query: Query = tuple(), res: ResourceLike = None,
status: Status = 201, query: Query = tuple(),
content_type: str = JSON, status: Status = 201,
accept: str = JSON, content_type: str = JSON,
headers: dict = None, accept: str = JSON,
token: str = None, headers: dict = None,
**kw) -> Res: token: str = None,
return super().post(data, uri, res, query, status, content_type, accept, headers, token, **kw,
**kw) ) -> Res:
return super().post(
data, uri, res, query, status, content_type, accept, headers, token, **kw
)
def patch(self, def patch(
data: str or dict, self,
uri: str = '', data: str or dict,
res: ResourceLike = None, uri: str = '',
query: Query = tuple(), res: ResourceLike = None,
item: Union[int, str] = None, query: Query = tuple(),
status: Status = 200, item: Union[int, str] = None,
content_type: str = JSON, status: Status = 200,
accept: str = JSON, content_type: str = JSON,
headers: dict = None, accept: str = JSON,
token: str = None, headers: dict = None,
**kw) -> Res: token: str = None,
return super().patch(data, uri, res, query, item, status, content_type, accept, token, **kw,
headers, **kw) ) -> Res:
return super().patch(
data,
uri,
res,
query,
item,
status,
content_type,
accept,
token,
headers,
**kw,
)
def put(self, def put(
data: str or dict, self,
uri: str = '', data: str or dict,
res: ResourceLike = None, uri: str = '',
query: Query = tuple(), res: ResourceLike = None,
item: Union[int, str] = None, query: Query = tuple(),
status: Status = 201, item: Union[int, str] = None,
content_type: str = JSON, status: Status = 201,
accept: str = JSON, content_type: str = JSON,
headers: dict = None, accept: str = JSON,
token: str = None, headers: dict = None,
**kw) -> Res: token: str = None,
return super().put(data, uri, res, query, item, status, content_type, accept, token, **kw,
headers, **kw) ) -> Res:
return super().put(
data,
uri,
res,
query,
item,
status,
content_type,
accept,
token,
headers,
**kw,
)
def delete(self, def delete(
uri: str = '', self,
res: ResourceLike = None, uri: str = '',
query: Query = tuple(), res: ResourceLike = None,
status: Status = 204, query: Query = tuple(),
item: Union[int, str] = None, status: Status = 204,
accept: str = JSON, item: Union[int, str] = None,
headers: dict = None, accept: str = JSON,
token: str = None, headers: dict = None,
**kw) -> Res: token: str = None,
return super().delete(uri, res, query, status, item, accept, headers, token, **kw) **kw,
) -> Res:
return super().delete(
uri, res, query, status, item, accept, headers, token, **kw
)
def login(self, email: str, password: str): def login(self, email: str, password: str):
assert isinstance(email, str) assert isinstance(email, str)
assert isinstance(password, 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, def get_many(
res: ResourceLike, self,
resources: Iterable[Union[dict, int]], res: ResourceLike,
key: str = None, resources: Iterable[Union[dict, int]],
**kw) -> Iterable[Union[Dict[str, object], str]]: key: str = None,
**kw,
) -> Iterable[Union[Dict[str, object], str]]:
"""Like :meth:`.get` but with many resources.""" """Like :meth:`.get` but with many resources."""
return ( return (
self.get(res=res, item=r[key] if key else r, **kw)[0] self.get(res=res, item=r[key] if key else r, **kw)[0] for r in resources
for r in resources
) )
@ -128,30 +175,47 @@ class UserClient(Client):
It will automatically perform login on the first request. It will automatically perform login on the first request.
""" """
def __init__(self, application, def __init__(
email: str, self,
password: str, application,
response_wrapper=None, email: str,
use_cookies=False, password: str,
allow_subdomain_redirects=False): response_wrapper=None,
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects) use_cookies=False,
allow_subdomain_redirects=False,
):
super().__init__(
application, response_wrapper, use_cookies, allow_subdomain_redirects
)
self.email = email # type: str self.email = email # type: str
self.password = password # type: str self.password = password # type: str
self.user = None # type: dict self.user = None # type: dict
def open(self, def open(
uri: str, self,
res: ResourceLike = None, uri: str,
status: int or HTTPException = 200, res: ResourceLike = None,
query: Query = tuple(), status: int or HTTPException = 200,
accept=JSON, query: Query = tuple(),
content_type=JSON, accept=JSON,
item=None, content_type=JSON,
headers: dict = None, item=None,
token: str = None, headers: dict = None,
**kw) -> Res: token: str = None,
return super().open(uri, res, status, query, accept, content_type, item, headers, **kw,
self.user['token'] if self.user else token, **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 # noinspection PyMethodOverriding
def login(self): def login(self):
@ -161,13 +225,15 @@ class UserClient(Client):
class UserClientFlask: class UserClientFlask:
def __init__(
def __init__(self, application, self,
email: str, application,
password: str, email: str,
response_wrapper=None, password: str,
use_cookies=True, response_wrapper=None,
follow_redirects=True): use_cookies=True,
follow_redirects=True,
):
self.email = email self.email = email
self.password = password self.password = password
self.follow_redirects = follow_redirects self.follow_redirects = follow_redirects
@ -181,38 +247,25 @@ class UserClientFlask:
'password': password, 'password': password,
'csrf_token': generate_csrf(), '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 self.headers = headers
body = next(body).decode("utf-8") body = next(body).decode("utf-8")
assert "Unassgined" in body assert "Unassgined" in body
def get(self, def get(self, uri='', data=None, follow_redirects=True, **kw):
uri='',
data=None,
follow_redirects=True,
**kw):
body, status, headers = self.client.get( body, status, headers = self.client.get(
uri, uri, data=data, follow_redirects=follow_redirects, headers=self.headers
data=data,
follow_redirects=follow_redirects,
headers=self.headers
) )
body = next(body).decode("utf-8") body = next(body).decode("utf-8")
return (body, status) return (body, status)
def post(self, def post(self, uri='', data=None, follow_redirects=True, **kw):
uri='',
data=None,
follow_redirects=True,
**kw):
import pdb; pdb.set_trace()
body, status, headers = self.client.post( body, status, headers = self.client.post(
uri, uri, data=data, follow_redirects=follow_redirects, headers=self.headers
data=data,
follow_redirects=follow_redirects,
headers=self.headers
) )
body = next(body).decode("utf-8") body = next(body).decode("utf-8")
return (body, status) return (body, status)

8
tests/conftest.py
View File

@ -142,9 +142,7 @@ def user3(app: Devicehub) -> UserClientFlask:
with app.app_context(): with app.app_context():
password = 'foo' password = 'foo'
user = create_user(password=password) user = create_user(password=password)
client = UserClientFlask( client = UserClientFlask(app, user.email, password)
app, user.email, password
)
return client return client
@ -155,9 +153,7 @@ def user4(app: Devicehub) -> UserClient:
password = 'foo' password = 'foo'
email = 'foo2@foo.com' email = 'foo2@foo.com'
user = create_user(email=email, password=password) user = create_user(email=email, password=password)
client = UserClientFlask( client = UserClientFlask(app, user.email, password)
app, user.email, password
)
return client return client