Source code for toxicbuild.common.client

# -*- coding: utf-8 -*-

# Copyright 2015-2017, 2019 Juca Crispim <>

# This file is part of toxicbuild.

# toxicbuild is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# toxicbuild is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with toxicbuild. If not, see <>.

from toxicbuild.core import BaseToxicClient
from toxicbuild.core.exceptions import ToxicClientException
from toxicbuild.core.utils import LoggerMixin
from .exceptions import (UserDoesNotExist, NotEnoughPerms,
                         BadResetPasswordToken, AlreadyExists)

[docs]class HoleClient(BaseToxicClient, LoggerMixin): """Client for the master's hole. """ settings = None def __init__(self, requester, *args, hole_token=None, **kwargs): """:param requester: The user who is willing to talk to the master. :param args: List arguments passed to super() constructor. :param hole_token: The token for access on the master. :param kwargs: Named arguments passed to super() constructor.""" self.hole_token = hole_token or self.settings.HOLE_TOKEN self.requester = requester super().__init__(*args, **kwargs) def __getattr__(self, name): action = name.replace('_', '-') async def _2serverandback(**kwargs): r = await self.request2server(action, body=kwargs) return r return _2serverandback
[docs] async def request2server(self, action, body): """Performs a request to a hole server. :param action: The action to perform on the server. :param body: The body of the request, with the actions parameters. """ data = {'action': action, 'body': body, 'token': self.hole_token} self.log('requesting action: ' + str(data), level='debug') if action not in ['user-authenticate']: data['user_id'] = str( await self.write(data) response = await self.get_response() return response['body'][action]
[docs] async def connect2stream(self, body): """Connects the client to the master's hole stream.""" action = 'stream' user_body = {'user_id': str(} body.update(user_body) await self.request2server(action, body)
[docs] async def get_response(self): response = await excs = {1: ToxicClientException, 2: UserDoesNotExist, 3: NotEnoughPerms, 4: BadResetPasswordToken, 5: AlreadyExists} if 'code' not in response: return response code = int(response['code']) exc = excs.get(code) if exc: raise exc(response['body'], 'error') return response
[docs]async def get_hole_client(requester, host, port, hole_token=None, **kwargs): client = HoleClient(requester, host, port, hole_token=hole_token, **kwargs) await client.connect() return client