110 def _pull(self): |
110 def _pull(self): |
111 with self.session.repo.internal_session() as isession: |
111 with self.session.repo.internal_session() as isession: |
112 lfsource = isession.repo.sources_by_uri['ldapuser'] |
112 lfsource = isession.repo.sources_by_uri['ldapuser'] |
113 stats = lfsource.pull_data(isession, force=True, raise_on_error=True) |
113 stats = lfsource.pull_data(isession, force=True, raise_on_error=True) |
114 isession.commit() |
114 isession.commit() |
|
115 |
|
116 def test_filter_inactivate(self): |
|
117 """ filtered out people should be deactivated, unable to authenticate """ |
|
118 source = self.session.execute('CWSource S WHERE S type="ldapfeed"').get_entity(0,0) |
|
119 config = source.repo_source.check_config(source) |
|
120 # filter with adim's phone number |
|
121 config['user-filter'] = u'(%s=%s)' % ('telephoneNumber', '109') |
|
122 source.repo_source.update_config(source, config) |
|
123 self.commit() |
|
124 self._pull() |
|
125 self.assertRaises(AuthenticationError, self.repo.connect, 'syt', password='syt') |
|
126 self.assertEqual(self.execute('Any N WHERE U login "syt", ' |
|
127 'U in_state S, S name N').rows[0][0], |
|
128 'deactivated') |
|
129 self.assertEqual(self.execute('Any N WHERE U login "adim", ' |
|
130 'U in_state S, S name N').rows[0][0], |
|
131 'activated') |
115 |
132 |
116 def test_delete(self): |
133 def test_delete(self): |
117 """ delete syt, pull, check deactivation, repull, |
134 """ delete syt, pull, check deactivation, repull, |
118 readd syt, pull, check activation |
135 readd syt, pull, check activation |
119 """ |
136 """ |