| 18 | test_utils.temp_folder) |
| 19 | |
| 20 | def connect(username, password): |
| 21 | server = Server('mock_server', get_info=OFFLINE_AD_2012_R2) |
| 22 | connection = Connection( |
| 23 | server, |
| 24 | user=username, |
| 25 | password=password, |
| 26 | authentication=SIMPLE, |
| 27 | read_only=True, |
| 28 | client_strategy=MOCK_SYNC |
| 29 | ) |
| 30 | |
| 31 | for dn, attrs in self._entries.items(): |
| 32 | dn = safe_dn(dn).lower() |
| 33 | |
| 34 | entry_added = connection.strategy.add_entry(dn, attrs) |
| 35 | if not entry_added: |
| 36 | raise Exception('Failed to add entry ' + dn) |
| 37 | |
| 38 | lower_keys = {key.lower(): key for key in attrs.keys()} |
| 39 | |
| 40 | if 'samaccountname' in lower_keys: |
| 41 | account_name = attrs[lower_keys['samaccountname']][0] |
| 42 | domain_start = dn.find('dc=') + 3 |
| 43 | domain_end = dn.find(',', domain_start) |
| 44 | domain = dn[domain_start:domain_end] |
| 45 | connection.server.dit[domain + '\\' + account_name] = connection.server.dit[dn] |
| 46 | |
| 47 | if 'userprincipalname' in lower_keys: |
| 48 | principal_name = attrs[lower_keys['userprincipalname']][0] |
| 49 | connection.server.dit[principal_name] = connection.server.dit[dn] |
| 50 | |
| 51 | connection.bind() |
| 52 | return connection |
| 53 | |
| 54 | authenticator._connect = connect |
| 55 | |