Skip to content
This repository has been archived by the owner on Apr 19, 2023. It is now read-only.

Commit

Permalink
Modernize all API itests
Browse files Browse the repository at this point in the history
Convert all remaining API itests to the modern test framework and
retire the old fixtures.  Required adding support for setting an
email address for a group during group creation.
  • Loading branch information
Russ Allbery committed Nov 11, 2019
1 parent 51820d4 commit 9d18ab9
Show file tree
Hide file tree
Showing 8 changed files with 193 additions and 128 deletions.
8 changes: 5 additions & 3 deletions grouper/repositories/group.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@ def __init__(self, session):
# type: (Session) -> None
self.session = session

def create_group(self, name, description, join_policy):
# type: (str, str, GroupJoinPolicy) -> None
group = SQLGroup(groupname=name, description=description, canjoin=join_policy.value)
def create_group(self, name, description, join_policy, email):
# type: (str, str, GroupJoinPolicy, Optional[str]) -> None
group = SQLGroup(
groupname=name, description=description, canjoin=join_policy.value, email_address=email
)
group.add(self.session)

def get_group(self, name):
Expand Down
8 changes: 4 additions & 4 deletions grouper/services/group.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from grouper.entities.permission_grant import GroupPermissionGrant
from grouper.repositories.group import GroupRepository
from grouper.repositories.interfaces import PermissionGrantRepository
from typing import List
from typing import List, Optional


class GroupService(GroupInterface):
Expand All @@ -21,11 +21,11 @@ def __init__(self, group_repository, permission_grant_repository):
self.group_repository = group_repository
self.permission_grant_repository = permission_grant_repository

def create_group(self, name, description, join_policy):
# type: (str, str, GroupJoinPolicy) -> None
def create_group(self, name, description, join_policy, email=None):
# type: (str, str, GroupJoinPolicy, Optional[str]) -> None
if not self.is_valid_group_name(name):
raise InvalidGroupNameException(name)
self.group_repository.create_group(name, description, join_policy)
self.group_repository.create_group(name, description, join_policy, email)

def group_has_matching_permission_grant(self, group, permission, argument):
# type: (str, str, str) -> bool
Expand Down
4 changes: 2 additions & 2 deletions grouper/usecases/interfaces.py
Original file line number Diff line number Diff line change
Expand Up @@ -131,8 +131,8 @@ class GroupInterface(metaclass=ABCMeta):
"""Abstract base class for group operations and queries."""

@abstractmethod
def create_group(self, name, description, join_policy):
# type: (str, str, GroupJoinPolicy) -> None
def create_group(self, name, description, join_policy, email=None):
# type: (str, str, GroupJoinPolicy, Optional[str]) -> None
pass

@abstractmethod
Expand Down
93 changes: 65 additions & 28 deletions itests/api/groups_test.py
Original file line number Diff line number Diff line change
@@ -1,28 +1,65 @@
from itests.fixtures import api_client, async_api_server # noqa: F401
from tests.fixtures import ( # noqa: F401
graph,
groups,
permissions,
service_accounts,
session,
standard_graph,
users,
)


def test_get_groups(api_client, groups): # noqa: F811
api_groups = list(api_client.groups)
assert sorted(api_groups) == sorted(groups)


def test_get_group(api_client): # noqa: F811
group = api_client.groups.get("team-sre")
assert sorted(group.groups) == ["all-teams", "serving-team", "team-infra"]
assert sorted(group.users) == ["[email protected]", "[email protected]", "[email protected]"]
assert group.subgroups == {}

perms = [(p.permission, p.argument) for p in group.permissions]
assert sorted(perms) == [("audited", ""), ("ssh", "*"), ("sudo", "shell"), ("team-sre", "*")]

assert group.audited
assert group.contacts == {"email": "[email protected]"}
from __future__ import annotations

from typing import TYPE_CHECKING

from groupy.client import Groupy

from itests.setup import api_server

if TYPE_CHECKING:
from py.path import LocalPath
from tests.setup import SetupTest


def create_graph(setup: SetupTest) -> None:
"""Create a simple graph structure with some nesting and permissions."""
setup.create_group("team-sre", email="[email protected]")
setup.add_user_to_group("[email protected]", "team-sre", role="owner")
setup.add_user_to_group("[email protected]", "team-sre")
setup.add_user_to_group("[email protected]", "team-sre")
setup.grant_permission_to_group("ssh", "*", "team-sre")
setup.grant_permission_to_group("team-sre", "*", "team-sre")
setup.add_group_to_group("team-sre", "serving-team")
setup.add_user_to_group("[email protected]", "serving-team", role="owner")
setup.create_permission("audited", audited=True)
setup.grant_permission_to_group("audited", "", "serving-team")
setup.add_group_to_group("serving-team", "team-infra")
setup.add_user_to_group("[email protected]", "team-infra", role="owner")
setup.grant_permission_to_group("sudo", "shell", "team-infra")
setup.add_group_to_group("serving-team", "all-teams")
setup.add_user_to_group("[email protected]", "all-teams", role="owner")


def test_get_groups(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
create_graph(setup)

with api_server(tmpdir) as api_url:
api_client = Groupy(api_url)
assert sorted(api_client.groups) == ["all-teams", "serving-team", "team-infra", "team-sre"]


def test_get_group(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
create_graph(setup)

with api_server(tmpdir) as api_url:
api_client = Groupy(api_url)

group = api_client.groups.get("team-sre")
assert sorted(group.groups) == ["all-teams", "serving-team", "team-infra"]
assert sorted(group.users) == ["[email protected]", "[email protected]", "[email protected]"]
assert group.subgroups == {}
assert group.audited
assert group.contacts == {"email": "[email protected]"}

permissions = [(p.permission, p.argument) for p in group.permissions]
assert sorted(permissions) == [
("audited", ""),
("ssh", "*"),
("sudo", "shell"),
("team-sre", "*"),
]

group = api_client.groups.get("serving-team")
assert sorted(group.subgroups) == ["team-sre"]
90 changes: 48 additions & 42 deletions itests/api/service_accounts_test.py
Original file line number Diff line number Diff line change
@@ -1,69 +1,75 @@
from __future__ import annotations

from typing import TYPE_CHECKING

from groupy.client import Groupy

from itests.fixtures import api_client, async_api_server # noqa: F401
from itests.setup import api_server
from tests.fixtures import ( # noqa: F401
graph,
groups,
permissions,
service_accounts,
session,
standard_graph,
users,
)

if TYPE_CHECKING:
from py.local import LocalPath
from tests.setup import SetupTest


def test_get_service_accounts(api_client, users, service_accounts): # noqa: F811
role_users = [username for username, u in users.items() if u.role_user]
assert len(role_users) > 0
def test_get_service_accounts(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
setup.create_user("[email protected]")
setup.create_role_user("[email protected]")
setup.create_service_account("[email protected]", "team-sre")

expected = role_users + list(service_accounts.keys())
with api_server(tmpdir) as api_url:
api_client = Groupy(api_url)
assert sorted(api_client.service_accounts) == ["[email protected]", "[email protected]"]

api_service_accounts = list(api_client.service_accounts)
assert sorted(api_service_accounts) == sorted(expected)

def test_get_service_account(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
setup.create_service_account(
"[email protected]",
owner="team-sre",
machine_set="some machines",
description="some service account",
)

def test_get_service_account(api_client): # noqa: F811
service_account = api_client.service_accounts.get("[email protected]")
assert service_account.groups == {}
assert service_account.passwords == []
assert service_account.public_keys == []
assert service_account.enabled
assert service_account.service_account == {
"description": "some service account",
"machine_set": "some machines",
"owner": "team-sre",
}
assert service_account.permissions == []
assert service_account.metadata == {}
with api_server(tmpdir) as api_url:
api_client = Groupy(api_url)
service_account = api_client.service_accounts.get("[email protected]")
assert service_account.groups == {}
assert service_account.passwords == []
assert service_account.public_keys == []
assert service_account.enabled
assert service_account.service_account == {
"description": "some service account",
"machine_set": "some machines",
"owner": "team-sre",
}
assert service_account.permissions == []
assert service_account.metadata == {}


def test_get_role_user(api_client): # noqa: F811
role_user = api_client.service_accounts.get("[email protected]")
assert role_user.groups == {}
assert role_user.passwords == []
assert role_user.public_keys == []
assert role_user.enabled
assert role_user.service_account is None
assert role_user.permissions == []
assert role_user.metadata == {}
def test_get_role_user(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
setup.create_role_user("[email protected]")

with api_server(tmpdir) as api_url:
api_client = Groupy(api_url)
role_user = api_client.service_accounts.get("[email protected]")
assert sorted(role_user.groups) == ["[email protected]"]
assert role_user.passwords == []
assert role_user.public_keys == []
assert role_user.enabled
assert role_user.service_account is None
assert role_user.permissions == []
assert role_user.metadata == {}


def test_includes_disabled_service_accounts(tmpdir, setup):
# type: (LocalPath, SetupTest) -> None
def test_includes_disabled_service_accounts(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
setup.create_service_account("[email protected]", "some-group", "some machines", "an account")
with setup.transaction():
setup.disable_service_account("[email protected]")

with api_server(tmpdir) as api_url:
api_client = Groupy(api_url) # noqa: F811
api_client = Groupy(api_url)
assert list(api_client.service_accounts) == ["[email protected]"]

service_account = api_client.service_accounts.get("[email protected]")
Expand Down
91 changes: 58 additions & 33 deletions itests/api/users_test.py
Original file line number Diff line number Diff line change
@@ -1,33 +1,58 @@
from grouper.constants import GROUP_ADMIN, PERMISSION_ADMIN, USER_ADMIN
from itests.fixtures import api_client, async_api_server # noqa: F401
from tests.fixtures import ( # noqa: F401
graph,
groups,
permissions,
service_accounts,
session,
standard_graph,
users,
)


def test_get_users(api_client, users): # noqa: F811
real_users = [username for username, u in users.items() if not u.role_user]
assert len(real_users) > 0

api_users = list(api_client.users)
assert sorted(api_users) == sorted(real_users)


def test_get_user(api_client): # noqa: F811
user = api_client.users.get("[email protected]")
assert sorted(user.groups) == ["group-admins", "permission-admins", "user-admins"]
assert user.passwords == []
assert user.public_keys == []
assert user.enabled
assert user.service_account is None

perms = [(p.permission, p.argument) for p in user.permissions]
assert sorted(perms) == [(GROUP_ADMIN, ""), (PERMISSION_ADMIN, ""), (USER_ADMIN, "")]

assert user.metadata == {}
from __future__ import annotations

from typing import TYPE_CHECKING

from groupy.client import Groupy

from grouper.constants import GROUP_ADMIN, USER_ADMIN
from itests.setup import api_server

if TYPE_CHECKING:
from py.local import LocalPath
from tests.setup import SetupTest


def test_get_users(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
setup.add_user_to_group("[email protected]", "some-group")
setup.create_user("[email protected]")
setup.create_user("[email protected]")
setup.disable_user("[email protected]")
setup.create_service_account("[email protected]", "some-group")
setup.create_role_user("[email protected]")

with api_server(tmpdir) as api_url:
api_client = Groupy(api_url)
assert sorted(api_client.users) == ["[email protected]", "[email protected]", "[email protected]"]


def test_get_user(tmpdir: LocalPath, setup: SetupTest) -> None:
with setup.transaction():
setup.add_user_to_group("[email protected]", "admins")
setup.grant_permission_to_group(GROUP_ADMIN, "", "admins")
setup.grant_permission_to_group(USER_ADMIN, "", "admins")
setup.add_user_to_group("[email protected]", "some-group")
setup.grant_permission_to_group("some-permission", "one", "some-group")
setup.add_group_to_group("some-group", "parent-group")
setup.grant_permission_to_group("some-permission", "two", "some-group")

with api_server(tmpdir) as api_url:
api_client = Groupy(api_url)
user = api_client.users.get("[email protected]")

assert sorted(user.groups) == ["admins", "parent-group", "some-group"]
assert user.passwords == []
assert user.public_keys == []
assert user.enabled
assert user.service_account is None
assert user.metadata == {}

permissions = [(p.permission, p.argument) for p in user.permissions]
assert sorted(permissions) == sorted(
[
(GROUP_ADMIN, ""),
(USER_ADMIN, ""),
("some-permission", "one"),
("some-permission", "two"),
]
)
14 changes: 1 addition & 13 deletions itests/fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,8 @@
from typing import TYPE_CHECKING

import pytest
from groupy.client import Groupy

from itests.setup import api_server, frontend_server
from itests.setup import frontend_server

if TYPE_CHECKING:
from grouper.graph import GroupGraph
Expand All @@ -23,14 +22,3 @@ def async_server(standard_graph, tmpdir):
# type: (GroupGraph, LocalPath) -> Iterator[str]
with frontend_server(tmpdir, "[email protected]") as frontend_url:
yield frontend_url


@pytest.fixture
def async_api_server(standard_graph, tmpdir):
with api_server(tmpdir) as api_url:
yield api_url


@pytest.fixture
def api_client(async_api_server):
return Groupy(async_api_server)
Loading

0 comments on commit 9d18ab9

Please sign in to comment.