quay/data/users/test/test_teamsync.py

from datetime import timedelta

import pytest

from data import model, database
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 util.names import parse_robot_username

_FAKE_AUTH = 'fake'

class FakeUsers(FederatedUsers):
  def __init__(self, group_members):
    super(FakeUsers, self).__init__(_FAKE_AUTH, False)
    self.group_tuples = [(m, None) for m in group_members]

  def iterate_group_members(self, group_lookup_args, page_size=None, disable_pagination=False):
    return (self.group_tuples, None)


@pytest.mark.parametrize('starting_membership,group_membership,expected_membership', [
  # Empty team + single member in group => Single member in team.
  ([],
   [
     UserInformation('someuser', 'someuser', 'someuser@devtable.com'),
   ],
   ['someuser']),

  # Team with a Quay user + empty group => empty team.
  ([('someuser', None)],
   [],
   []),

  # Team with an existing external user + user is in the group => no changes.
  ([
    ('someuser', 'someuser'),
   ],
   [
     UserInformation('someuser', 'someuser', 'someuser@devtable.com'),
   ],
   ['someuser']),

  # Team with an existing external user (with a different Quay username) + user is in the group.
  # => no changes
  ([
    ('anotherquayname', 'someuser'),
   ],
   [
     UserInformation('someuser', 'someuser', 'someuser@devtable.com'),
   ],
   ['someuser']),

  # Team missing a few members that are in the group => members added.
  ([('someuser', 'someuser')],
   [
     UserInformation('anotheruser', 'anotheruser', 'anotheruser@devtable.com'),
     UserInformation('someuser', 'someuser', 'someuser@devtable.com'),
     UserInformation('thirduser', 'thirduser', 'thirduser@devtable.com'),
   ],
   ['anotheruser', 'someuser', 'thirduser']),

  # Team has a few extra members no longer in the group => members removed.
  ([
    ('anotheruser', 'anotheruser'),
    ('someuser', 'someuser'),
    ('thirduser', 'thirduser'),
    ('nontestuser', None),
   ],
   [
     UserInformation('thirduser', 'thirduser', 'thirduser@devtable.com'),
   ],
   ['thirduser']),

  # Team has different membership than the group => members added and removed.
  ([
    ('anotheruser', 'anotheruser'),
    ('someuser', 'someuser'),
    ('nontestuser', None),
   ],
   [
     UserInformation('anotheruser', 'anotheruser', 'anotheruser@devtable.com'),
     UserInformation('missinguser', 'missinguser', 'missinguser@devtable.com'),
   ],
   ['anotheruser', 'missinguser']),

  # Team has same membership but some robots => robots remain and no other changes.
  ([
    ('someuser', 'someuser'),
    ('buynlarge+anotherbot', None),
    ('buynlarge+somerobot', None),
   ],
   [
     UserInformation('someuser', 'someuser', 'someuser@devtable.com'),
   ],
   ['someuser', 'buynlarge+somerobot', 'buynlarge+anotherbot']),

  # Team has an extra member and some robots => member removed and robots remain.
  ([
    ('someuser', 'someuser'),
    ('buynlarge+anotherbot', None),
    ('buynlarge+somerobot', None),
   ],
   [
     # No members.
   ],
   ['buynlarge+somerobot', 'buynlarge+anotherbot']),

  # Team has a different member and some robots => member changed and robots remain.
  ([
    ('someuser', 'someuser'),
    ('buynlarge+anotherbot', None),
    ('buynlarge+somerobot', None),
   ],
   [
     UserInformation('anotheruser', 'anotheruser', 'anotheruser@devtable.com'),
   ],
   ['anotheruser', 'buynlarge+somerobot', 'buynlarge+anotherbot']),

  # Team with an existing external user (with a different Quay username) + user is in the group.
  # => no changes and robots remain.
  ([
    ('anotherquayname', 'someuser'),
    ('buynlarge+anotherbot', None),
   ],
   [
     UserInformation('someuser', 'someuser', 'someuser@devtable.com'),
   ],
   ['someuser', 'buynlarge+anotherbot']),
])
def test_syncing(starting_membership, group_membership, expected_membership, app):
  org = model.organization.get_organization('buynlarge')

  # Necessary for the fake auth entries to be created in FederatedLogin.
  database.LoginService.create(name=_FAKE_AUTH)

  # Assert the team is empty, so we have a clean slate.
  sync_team_info = model.team.get_team_sync_information('buynlarge', 'synced')
  assert len(list(model.team.list_team_users(sync_team_info.team))) == 0

  # Add the existing starting members to the team.
  for starting_member in starting_membership:
    (quay_username, fakeauth_username) = starting_member
    if '+' in quay_username:
      # Add a robot.
      (_, shortname) = parse_robot_username(quay_username)
      robot, _ = model.user.create_robot(shortname, org)
      model.team.add_user_to_team(robot, sync_team_info.team)
    else:
      email = quay_username + '@devtable.com'

      if fakeauth_username is None:
        quay_user = model.user.create_user_noverify(quay_username, email)
      else:
        quay_user = model.user.create_federated_user(quay_username, email, _FAKE_AUTH,
                                                     fakeauth_username, False)

      model.team.add_user_to_team(quay_user, sync_team_info.team)

  # Call syncing on the team.
  fake_auth = FakeUsers(group_membership)
  assert sync_team(fake_auth, sync_team_info)

  # Ensure the last updated time and transaction_id's have changed.
  updated_sync_info = model.team.get_team_sync_information('buynlarge', 'synced')
  assert updated_sync_info.last_updated is not None
  assert updated_sync_info.transaction_id != sync_team_info.transaction_id

  users_expected = set([name for name in expected_membership if '+' not in name])
  robots_expected = set([name for name in expected_membership if '+' in name])
  assert len(users_expected) + len(robots_expected) == len(expected_membership)

  # Check that the team's users match those expected.
  service_user_map = model.team.list_federated_team_members(sync_team_info.team, _FAKE_AUTH)
  assert set(service_user_map.keys()) == users_expected

  quay_users = model.team.list_team_users(sync_team_info.team)
  assert len(quay_users) == len(users_expected)

  for quay_user in quay_users:
    fakeauth_record = model.user.lookup_federated_login(quay_user, _FAKE_AUTH)
    assert fakeauth_record is not None
    assert fakeauth_record.service_ident in users_expected
    assert service_user_map[fakeauth_record.service_ident] == quay_user.id

  # Check that the team's robots match those expected.
  robots_found = set([r.username for r in model.team.list_team_robots(sync_team_info.team)])
  assert robots_expected == robots_found


def test_sync_teams_to_groups(app):
  # Necessary for the fake auth entries to be created in FederatedLogin.
  database.LoginService.create(name=_FAKE_AUTH)

  # 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

  # Call to sync all teams.
  fake_auth = FakeUsers([])
  sync_teams_to_groups(fake_auth, timedelta(seconds=1))

  # Ensure the team was synced.
  updated_sync_info = model.team.get_team_sync_information('buynlarge', 'synced')
  assert updated_sync_info.last_updated is not None
  assert updated_sync_info.transaction_id != sync_team_info.transaction_id

  # Set the stale threshold to a high amount and ensure the team is not resynced.
  sync_teams_to_groups(fake_auth, timedelta(seconds=120))

  third_sync_info = model.team.get_team_sync_information('buynlarge', 'synced')
  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))

  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,
])
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

  with auth_system_builder() as auth:
    assert sync_team(auth, sync_team_info)