CreateRoom.transform_response() - Code Metrics - Inspection of "pyejabberd 0.3.0?" - dirkmoors/pyejabberd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#23)

unknown

created 2016-08-29 18:41 UTC

CreateRoom.transform_response() A

↳ Parent: CreateRoom

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	2
rs	10
cc	1

# -*- coding: utf-8 -*-
from __future__ import unicode_literals

from .core.arguments import StringArgument
from .core.definitions import API
from .core.serializers import StringSerializer
from .muc import muc_room_options_serializers
from .muc.arguments import MUCRoomArgument, AffiliationArgument
from .muc.enums import MUCRoomOption

from .errors import UserAlreadyRegisteredError
from pyejabberd.muc.enums import Affiliation
from .utils import format_password_hash_sha


class Echo(API):
    method = 'echothisnew'
    arguments = [StringArgument('sentence')]

    def transform_response(self, api, arguments, response):
        return response.get('repeated')


class RegisteredUsers(API):
    method = 'registered_users'
    arguments = [StringArgument('host')]

    def transform_response(self, api, arguments, response):
        return response.get('users', [])


class Register(API):
    method = 'register'
    arguments = [StringArgument('user'), StringArgument('host'), StringArgument('password')]

    def validate_response(self, api, arguments, response):
        if response.get('res') == 1:
            username = arguments.get('user')
            raise UserAlreadyRegisteredError('User with username %s already exists' % username)

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class UnRegister(API):
    method = 'unregister'
    arguments = [StringArgument('user'), StringArgument('host')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class ChangePassword(API):
    method = 'change_password'
    arguments = [StringArgument('user'), StringArgument('host'), StringArgument('newpass')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class CheckPasswordHash(API):
    method = 'check_password_hash'
    arguments = [StringArgument('user'), StringArgument('host'), StringArgument('passwordhash'),
                 StringArgument('hashmethod')]

    def transform_arguments(self, **kwargs):
        passwordhash = format_password_hash_sha(password=kwargs.pop('password'))
        kwargs.update({
            'passwordhash': passwordhash,
            'hashmethod': 'sha'
        })
        return kwargs

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class SetNickname(API):
    method = 'set_nickname'
    arguments = [StringArgument('user'), StringArgument('host'), StringArgument('nickname')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0

class ConnectedUsers(API):
    method = 'connected_users'
    arguments = []

    def transform_response(self, api, arguments, response):
        connected_users = response.get('connected_users', [])

        return [user["sessions"] for user in connected_users]

class ConnectedUsersInfo(API):
    method = 'connected_users_info'
    arguments = []

    def transform_response(self, api, arguments, response):
        connected_users_info = response.get('connected_users_info', [])

        return [user["sessions"] for user in connected_users_info]

class ConnectedUsersNumber(API):
    method = 'connected_users_number'
    arguments = []

    def transform_response(self, api, arguments, response):
        return response.get('num_sessions')

class UserSessionsInfo(API):
    method = 'user_sessions_info'
    arguments = [StringArgument('user'), StringArgument('host')]

    def process_sessions_info(self, sessions_info):
        return [
            dict((k, v) for property_k_v in session["session"] for k, v in property_k_v.items())
            for session in sessions_info
        ]

    def transform_response(self, api, arguments, response):
        return self.process_sessions_info(response.get('sessions_info', []))

class MucOnlineRooms(API):
    method = 'muc_online_rooms'
    arguments = [StringArgument('host')]

    def transform_response(self, api, arguments, response):
        return [result_dict.get('room') for result_dict in response.get('rooms', {})]


class CreateRoom(API):
    method = 'create_room'
    arguments = [StringArgument('name'), StringArgument('service'), StringArgument('host')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class DestroyRoom(API):
    method = 'destroy_room'
    arguments = [StringArgument('name'), StringArgument('service')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class GetRoomOptions(API):
    method = 'get_room_options'
    arguments = [StringArgument('name'), StringArgument('service')]

    def transform_response(self, api, arguments, response):
        result = {}
        for option_dict in response.get('options', []):
            option = option_dict.get('option')
            if option is None:
                raise ValueError('Unexpected option in response: ' % str(option_dict))
            name_dict, value_dict = option
            result[name_dict['name']] = value_dict['value']
        return result


class ChangeRoomOption(API):
    method = 'change_room_option'
    arguments = [StringArgument('name'), StringArgument('service'), MUCRoomArgument('option'), StringArgument('value')]

    def transform_arguments(self, **kwargs):
        option = kwargs.get('option')
        assert isinstance(option, MUCRoomOption)
        serializer_class = muc_room_options_serializers.get(option, StringSerializer)
        kwargs['value'] = serializer_class().to_api(kwargs['value'])
        return kwargs

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class SetRoomAffiliation(API):
    method = 'set_room_affiliation'
    arguments = [StringArgument('name'), StringArgument('service'), StringArgument('jid'),
                 AffiliationArgument('affiliation')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class GetRoomAffiliations(API):
    method = 'get_room_affiliations'
    arguments = [StringArgument('name'), StringArgument('service')]

    def transform_response(self, api, arguments, response):
        affiliations = response.get('affiliations', [])
        return [{
            'username': subdict['affiliation'][0]['username'],
            'domain': subdict['affiliation'][1]['domain'],
            'affiliation': Affiliation.get_by_name(subdict['affiliation'][2]['affiliation']),
            'reason': subdict['affiliation'][3]['reason'],
        } for subdict in affiliations]


class AddRosterItem(API):
    method = 'add_rosteritem'
    arguments = [StringArgument('localuser'), StringArgument('localserver'),
                 StringArgument('user'), StringArgument('server'),
                 StringArgument('nick'), StringArgument('group'), StringArgument('subs')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class DeleteRosterItem(API):
    method = 'delete_rosteritem'
    arguments = [StringArgument('localuser'), StringArgument('localserver'),
                 StringArgument('user'), StringArgument('server')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class GetRoster(API):
    method = 'get_roster'
    arguments = [StringArgument('user'), StringArgument('server')]

    def transform_response(self, api, arguments, response):
        roster = []
        for contact in response.get('contacts', []):
            contact_details = {}
            for parameter in contact['contact']:
                for key, value in parameter.items():
                    contact_details[key] = value
            roster.append(contact_details)
        return roster


class CheckAccount(API):
    method = 'check_account'
    arguments = [StringArgument('user'), StringArgument('host')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


class KickUser(API):
    method = 'kick_user'
    arguments = [StringArgument('user'), StringArgument('host')]

    def transform_response(self, api, arguments, response):
        return response.get('num_resources')


class KickSession(API):
    method = 'kick_session'
    arguments = [StringArgument('user'), StringArgument('host'), StringArgument('resource'), StringArgument('reason')]

    def transform_response(self, api, arguments, response):
        return response.get('res') == 0


1			# -- coding: utf-8 --
2			from __future__ import unicode_literals
3
4			from .core.arguments import StringArgument
5			from .core.definitions import API
6			from .core.serializers import StringSerializer
7			from .muc import muc_room_options_serializers
8			from .muc.arguments import MUCRoomArgument, AffiliationArgument
9			from .muc.enums import MUCRoomOption
10
11			from .errors import UserAlreadyRegisteredError
12			from pyejabberd.muc.enums import Affiliation
13			from .utils import format_password_hash_sha
14
15
16			class Echo(API):
17			method = 'echothisnew'
18			arguments = [StringArgument('sentence')]
19
20			def transform_response(self, api, arguments, response):
21			return response.get('repeated')
22
23
24			class RegisteredUsers(API):
25			method = 'registered_users'
26			arguments = [StringArgument('host')]
27
28			def transform_response(self, api, arguments, response):
29			return response.get('users', [])
30
31
32			class Register(API):
33			method = 'register'
34			arguments = [StringArgument('user'), StringArgument('host'), StringArgument('password')]
35
36			def validate_response(self, api, arguments, response):
37			if response.get('res') == 1:
38			username = arguments.get('user')
39			raise UserAlreadyRegisteredError('User with username %s already exists' % username)
40
41			def transform_response(self, api, arguments, response):
42			return response.get('res') == 0
43
44
45			class UnRegister(API):
46			method = 'unregister'
47			arguments = [StringArgument('user'), StringArgument('host')]
48
49			def transform_response(self, api, arguments, response):
50			return response.get('res') == 0
51
52
53			class ChangePassword(API):
54			method = 'change_password'
55			arguments = [StringArgument('user'), StringArgument('host'), StringArgument('newpass')]
56
57			def transform_response(self, api, arguments, response):
58			return response.get('res') == 0
59
60
61			class CheckPasswordHash(API):
62			method = 'check_password_hash'
63			arguments = [StringArgument('user'), StringArgument('host'), StringArgument('passwordhash'),
64			StringArgument('hashmethod')]
65
66			def transform_arguments(self, **kwargs):
67			passwordhash = format_password_hash_sha(password=kwargs.pop('password'))
68			kwargs.update({
69			'passwordhash': passwordhash,
70			'hashmethod': 'sha'
71			})
72			return kwargs
73
74			def transform_response(self, api, arguments, response):
75			return response.get('res') == 0
76
77
78			class SetNickname(API):
79			method = 'set_nickname'
80			arguments = [StringArgument('user'), StringArgument('host'), StringArgument('nickname')]
81
82			def transform_response(self, api, arguments, response):
83			return response.get('res') == 0
84
85			class ConnectedUsers(API):
86			method = 'connected_users'
87			arguments = []
88
89			def transform_response(self, api, arguments, response):
90			connected_users = response.get('connected_users', [])
91
92			return [user["sessions"] for user in connected_users]
93
94			class ConnectedUsersInfo(API):
95			method = 'connected_users_info'
96			arguments = []
97
98			def transform_response(self, api, arguments, response):
99			connected_users_info = response.get('connected_users_info', [])
100
101			return [user["sessions"] for user in connected_users_info]
102
103			class ConnectedUsersNumber(API):
104			method = 'connected_users_number'
105			arguments = []
106
107			def transform_response(self, api, arguments, response):
108			return response.get('num_sessions')
109
110			class UserSessionsInfo(API):
111			method = 'user_sessions_info'
112			arguments = [StringArgument('user'), StringArgument('host')]
113
114			def process_sessions_info(self, sessions_info):
115			return [
116			dict((k, v) for property_k_v in session["session"] for k, v in property_k_v.items())
117			for session in sessions_info
118			]
119
120			def transform_response(self, api, arguments, response):
121			return self.process_sessions_info(response.get('sessions_info', []))
122
123			class MucOnlineRooms(API):
124			method = 'muc_online_rooms'
125			arguments = [StringArgument('host')]
126
127			def transform_response(self, api, arguments, response):
128			return [result_dict.get('room') for result_dict in response.get('rooms', {})]
129
130
131			class CreateRoom(API):
132			method = 'create_room'
133			arguments = [StringArgument('name'), StringArgument('service'), StringArgument('host')]
134
135			def transform_response(self, api, arguments, response):
136			return response.get('res') == 0
137
138
139			class DestroyRoom(API):
140			method = 'destroy_room'
141			arguments = [StringArgument('name'), StringArgument('service')]
142
143			def transform_response(self, api, arguments, response):
144			return response.get('res') == 0
145
146
147			class GetRoomOptions(API):
148			method = 'get_room_options'
149			arguments = [StringArgument('name'), StringArgument('service')]
150
151			def transform_response(self, api, arguments, response):
152			result = {}
153			for option_dict in response.get('options', []):
154			option = option_dict.get('option')
155			if option is None:
156			raise ValueError('Unexpected option in response: ' % str(option_dict))
157			name_dict, value_dict = option
158			result[name_dict['name']] = value_dict['value']
159			return result
160
161
162			class ChangeRoomOption(API):
163			method = 'change_room_option'
164			arguments = [StringArgument('name'), StringArgument('service'), MUCRoomArgument('option'), StringArgument('value')]
165
166			def transform_arguments(self, **kwargs):
167			option = kwargs.get('option')
168			assert isinstance(option, MUCRoomOption)
169			serializer_class = muc_room_options_serializers.get(option, StringSerializer)
170			kwargs['value'] = serializer_class().to_api(kwargs['value'])
171			return kwargs
172
173			def transform_response(self, api, arguments, response):
174			return response.get('res') == 0
175
176
177			class SetRoomAffiliation(API):
178			method = 'set_room_affiliation'
179			arguments = [StringArgument('name'), StringArgument('service'), StringArgument('jid'),
180			AffiliationArgument('affiliation')]
181
182			def transform_response(self, api, arguments, response):
183			return response.get('res') == 0
184
185
186			class GetRoomAffiliations(API):
187			method = 'get_room_affiliations'
188			arguments = [StringArgument('name'), StringArgument('service')]
189
190			def transform_response(self, api, arguments, response):
191			affiliations = response.get('affiliations', [])
192			return [{
193			'username': subdict['affiliation'][0]['username'],
194			'domain': subdict['affiliation'][1]['domain'],
195			'affiliation': Affiliation.get_by_name(subdict['affiliation'][2]['affiliation']),
196			'reason': subdict['affiliation'][3]['reason'],
197			} for subdict in affiliations]
198
199
200			class AddRosterItem(API):
201			method = 'add_rosteritem'
202			arguments = [StringArgument('localuser'), StringArgument('localserver'),
203			StringArgument('user'), StringArgument('server'),
204			StringArgument('nick'), StringArgument('group'), StringArgument('subs')]
205
206			def transform_response(self, api, arguments, response):
207			return response.get('res') == 0
208
209
210			class DeleteRosterItem(API):
211			method = 'delete_rosteritem'
212			arguments = [StringArgument('localuser'), StringArgument('localserver'),
213			StringArgument('user'), StringArgument('server')]
214
215			def transform_response(self, api, arguments, response):
216			return response.get('res') == 0
217
218
219			class GetRoster(API):
220			method = 'get_roster'
221			arguments = [StringArgument('user'), StringArgument('server')]
222
223			def transform_response(self, api, arguments, response):
224			roster = []
225			for contact in response.get('contacts', []):
226			contact_details = {}
227			for parameter in contact['contact']:
228			for key, value in parameter.items():
229			contact_details[key] = value
230			roster.append(contact_details)
231			return roster
232
233
234			class CheckAccount(API):
235			method = 'check_account'
236			arguments = [StringArgument('user'), StringArgument('host')]
237
238			def transform_response(self, api, arguments, response):
239			return response.get('res') == 0
240
241
242			class KickUser(API):
243			method = 'kick_user'
244			arguments = [StringArgument('user'), StringArgument('host')]
245
246			def transform_response(self, api, arguments, response):
247			return response.get('num_resources')
248
249
250			class KickSession(API):
251			method = 'kick_session'
252			arguments = [StringArgument('user'), StringArgument('host'), StringArgument('resource'), StringArgument('reason')]
253
254			def transform_response(self, api, arguments, response):
255			return response.get('res') == 0
256

dirkmoors / pyejabberd

GitHub Access Token became invalid

Pull Request — master (#23)

CreateRoom.transform_response() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like