codekingpro/portable-devtools

1"""
2"""
3
4# Created on 2014.11.17
5#
6# Author: Giovanni Cannata
7#
8# Copyright 2014 - 2020 Giovanni Cannata
9#
10# This file is part of ldap3.
11#
12# ldap3 is free software: you can redistribute it and/or modify
13# it under the terms of the GNU Lesser General Public License as published
14# by the Free Software Foundation, either version 3 of the License, or
15# (at your option) any later version.
16#
17# ldap3 is distributed in the hope that it will be useful,
18# but WITHOUT ANY WARRANTY; without even the implied warranty of
19# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20# GNU Lesser General Public License for more details.
21#
22# You should have received a copy of the GNU Lesser General Public License
23# along with ldap3 in the COPYING and COPYING.LESSER files.
24# If not, see <http://www.gnu.org/licenses/>.
25
26from ..core.results import DO_NOT_RAISE_EXCEPTIONS
27from .mockBase import MockBaseStrategy
28from .. import ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, NO_ATTRIBUTES
29from .sync import SyncStrategy
30from ..operation.bind import bind_response_to_dict
31from ..operation.delete import delete_response_to_dict
32from ..operation.add import add_response_to_dict
33from ..operation.compare import compare_response_to_dict
34from ..operation.modifyDn import modify_dn_response_to_dict
35from ..operation.modify import modify_response_to_dict
36from ..operation.search import search_result_done_response_to_dict, search_result_entry_response_to_dict
37from ..operation.extended import extended_response_to_dict
38from ..core.exceptions import LDAPSocketOpenError, LDAPOperationResult
39from ..utils.log import log, log_enabled, ERROR, PROTOCOL
40
41
42class MockSyncStrategy(MockBaseStrategy, SyncStrategy):  # class inheritance sequence is important, MockBaseStrategy must be the first one
43    """
44    This strategy create a mock LDAP server, with synchronous access
45    It can be useful to test LDAP without accessing a real Server
46    """
47    def __init__(self, ldap_connection):
48        SyncStrategy.__init__(self, ldap_connection)
49        MockBaseStrategy.__init__(self)
50
51    def post_send_search(self, payload):
52        message_id, message_type, request, controls = payload
53        self.connection.response = []
54        self.connection.result = dict()
55        if message_type == 'searchRequest':
56            responses, result = self.mock_search(request, controls)
57            for entry in responses:
58                response = search_result_entry_response_to_dict(entry, self.connection.server.schema, self.connection.server.custom_formatter, self.connection.check_names)
59                response['type'] = 'searchResEntry'
60                ###
61                if self.connection.empty_attributes:
62                    for attribute_type in request['attributes']:
63                        attribute_name = str(attribute_type)
64                        if attribute_name not in response['raw_attributes'] and attribute_name not in (ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, NO_ATTRIBUTES):
65                            response['raw_attributes'][attribute_name] = list()
66                            response['attributes'][attribute_name] = list()
67                            if log_enabled(PROTOCOL):
68                                log(PROTOCOL, 'attribute set to empty list for missing attribute <%s> in <%s>',
69                                    attribute_type, self)
70                    if not self.connection.auto_range:
71                        attrs_to_remove = []
72                        # removes original empty attribute in case a range tag is returned
73                        for attribute_type in response['attributes']:
74                            attribute_name = str(attribute_type)
75                            if ';range' in attribute_name.lower():
76                                orig_attr, _, _ = attribute_name.partition(';')
77                                attrs_to_remove.append(orig_attr)
78                        for attribute_type in attrs_to_remove:
79                            if log_enabled(PROTOCOL):
80                                log(PROTOCOL,
81                                    'attribute type <%s> removed in response because of same attribute returned as range by the server in <%s>',
82                                    attribute_type, self)
83                            del response['raw_attributes'][attribute_type]
84                            del response['attributes'][attribute_type]
85                ###
86                self.connection.response.append(response)
87            result = search_result_done_response_to_dict(result)
88            result['type'] = 'searchResDone'
89            self.connection.result = result
90            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
91                if log_enabled(PROTOCOL):
92                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
93                raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
94
95        return self.connection.response
96
97    def post_send_single_response(self, payload):  # payload is a tuple sent by self.send() made of message_type, request, controls
98        message_id, message_type, request, controls = payload
99        responses = []
100        result = None
101        if message_type == 'bindRequest':
102            result = bind_response_to_dict(self.mock_bind(request, controls))
103            result['type'] = 'bindResponse'
104        elif message_type == 'unbindRequest':
105            self.bound = None
106        elif message_type == 'abandonRequest':
107            pass
108        elif message_type == 'delRequest':
109            result = delete_response_to_dict(self.mock_delete(request, controls))
110            result['type'] = 'delResponse'
111        elif message_type == 'addRequest':
112            result = add_response_to_dict(self.mock_add(request, controls))
113            result['type'] = 'addResponse'
114        elif message_type == 'compareRequest':
115            result = compare_response_to_dict(self.mock_compare(request, controls))
116            result['type'] = 'compareResponse'
117        elif message_type == 'modDNRequest':
118            result = modify_dn_response_to_dict(self.mock_modify_dn(request, controls))
119            result['type'] = 'modDNResponse'
120        elif message_type == 'modifyRequest':
121            result = modify_response_to_dict(self.mock_modify(request, controls))
122            result['type'] = 'modifyResponse'
123        elif message_type == 'extendedReq':
124            result = extended_response_to_dict(self.mock_extended(request, controls))
125            result['type'] = 'extendedResp'
126        self.connection.result = result
127        responses.append(result)
128        if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
129            if log_enabled(PROTOCOL):
130                log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
131            raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
132        return responses
133
134