Add group iteration and syncing support to Keystone auth

This commit is contained in:
Joseph Schorr 2017-02-23 14:41:27 -05:00
parent 47278cc559
commit d7825c6720
6 changed files with 148 additions and 15 deletions

View file

@ -7,6 +7,7 @@ from data.users.federated import FederatedUsers, UserInformation
from data.users.teamsync import sync_team, sync_teams_to_groups
from test.fixtures import app, appconfig, database_uri, init_db_path, sqlitedb_file
from test.test_ldap import mock_ldap
from test.test_keystone_auth import fake_keystone
from util.names import parse_robot_username
_FAKE_AUTH = 'fake'
@ -213,20 +214,28 @@ def test_sync_teams_to_groups(app):
assert third_sync_info.last_updated == updated_sync_info.last_updated
assert third_sync_info.transaction_id == updated_sync_info.transaction_id
# Set the stale threshold to -1 seconds, and ensure the team is resynced.
sync_teams_to_groups(fake_auth, timedelta(seconds=-1))
# Set the stale threshold to -10 seconds, and ensure the team is resynced.
sync_teams_to_groups(fake_auth, timedelta(seconds=-120))
fourth_sync_info = model.team.get_team_sync_information('buynlarge', 'synced')
assert fourth_sync_info.transaction_id != updated_sync_info.transaction_id
@pytest.mark.parametrize('auth_system_builder', [
mock_ldap,
@pytest.mark.parametrize('auth_system_builder,config', [
(mock_ldap, {'group_dn': 'cn=AwesomeFolk'}),
(fake_keystone, {'group_id': 'somegroupid'}),
])
def test_teamsync_end_to_end(auth_system_builder, app):
# Assert the team has not yet been updated.
sync_team_info = model.team.get_team_sync_information('buynlarge', 'synced')
assert sync_team_info.last_updated is None
def test_teamsync_end_to_end(auth_system_builder, config, app):
with auth_system_builder() as auth:
# Create an new team to sync.
org = model.organization.get_organization('buynlarge')
new_synced_team = model.team.create_team('synced2', org, 'member', 'Some synced team.')
sync_team_info = model.team.set_team_syncing(new_synced_team, auth.federated_service, config)
# Sync the team.
assert sync_team(auth, sync_team_info)
# Ensure we now have members.
msg = 'Auth system: %s' % auth.federated_service
sync_team_info = model.team.get_team_sync_information('buynlarge', 'synced2')
assert len(list(model.team.list_team_users(sync_team_info.team))) > 0, msg