feedc0de/platformio-core
1
0
Fork 0
forked from platformio/platformio-core
Code Pull Requests Activity

CLI to manage teams. Resolve #3533 (#3547)

Browse Source 
* CLI to manage teams.Minor fixes. Resolve #3533

* fix teams tests

* disable org and team tests

* minor fixes. fix error texts

* fix split compatibility
This commit is contained in:
ShahRustam
2020-06-04 19:31:30 +03:00
committed by GitHub
parent 42df3c9c3f
commit 94cb808285
5 changed files with 453 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
platformio/clients/account.py
View File

@ -115,12 +115,11 @@ class AccountClient(RESTClient): # pylint:disable=too-many-public-methods
return True
def change_password(self, old_password, new_password):
self.send_auth_request(
return self.send_auth_request(
"post",
"/v1/password",
data={"old_password": old_password, "new_password": new_password},
)
return True
def registration(
self, username, email, password, firstname, lastname
@ -147,12 +146,11 @@ class AccountClient(RESTClient): # pylint:disable=too-many-public-methods
)
def auth_token(self, password, regenerate):
result = self.send_auth_request(
return self.send_auth_request(
"post",
"/v1/token",
data={"password": password, "regenerate": 1 if regenerate else 0},
)
return result.get("auth_token")
).get("auth_token")
def forgot_password(self, username):
return self.send_request("post", "/v1/forgot", data={"username": username},)
@ -192,38 +190,76 @@ class AccountClient(RESTClient): # pylint:disable=too-many-public-methods
return result
def create_org(self, orgname, email, display_name):
response = self.send_auth_request(
return self.send_auth_request(
"post",
"/v1/orgs",
data={"orgname": orgname, "email": email, "displayname": display_name},
)
return response
def get_org(self, orgname):
return self.send_auth_request("get", "/v1/orgs/%s" % orgname)
def list_orgs(self):
response = self.send_auth_request("get", "/v1/orgs",)
return response
return self.send_auth_request("get", "/v1/orgs",)
def update_org(self, orgname, data):
response = self.send_auth_request(
return self.send_auth_request(
"put", "/v1/orgs/%s" % orgname, data={k: v for k, v in data.items() if v}
)
return response
def add_org_owner(self, orgname, username):
response = self.send_auth_request(
return self.send_auth_request(
"post", "/v1/orgs/%s/owners" % orgname, data={"username": username},
)
return response
def list_org_owners(self, orgname):
response = self.send_auth_request("get", "/v1/orgs/%s/owners" % orgname,)
return response
return self.send_auth_request("get", "/v1/orgs/%s/owners" % orgname,)
def remove_org_owner(self, orgname, username):
response = self.send_auth_request(
return self.send_auth_request(
"delete", "/v1/orgs/%s/owners" % orgname, data={"username": username},
)
return response
def create_team(self, orgname, teamname, description):
return self.send_auth_request(
"post",
"/v1/orgs/%s/teams" % orgname,
data={"name": teamname, "description": description},
)
def destroy_team(self, orgname, teamname):
return self.send_auth_request(
"delete", "/v1/orgs/%s/teams/%s" % (orgname, teamname),
)
def get_team(self, orgname, teamname):
return self.send_auth_request(
"get", "/v1/orgs/%s/teams/%s" % (orgname, teamname),
)
def list_teams(self, orgname):
return self.send_auth_request("get", "/v1/orgs/%s/teams" % orgname,)
def update_team(self, orgname, teamname, data):
return self.send_auth_request(
"put",
"/v1/orgs/%s/teams/%s" % (orgname, teamname),
data={k: v for k, v in data.items() if v},
)
def add_team_member(self, orgname, teamname, username):
return self.send_auth_request(
"post",
"/v1/orgs/%s/teams/%s/members" % (orgname, teamname),
data={"username": username},
)
def remove_team_member(self, orgname, teamname, username):
return self.send_auth_request(
"delete",
"/v1/orgs/%s/teams/%s/members" % (orgname, teamname),
data={"username": username},
)
def fetch_authentication_token(self):
if os.environ.get("PLATFORMIO_AUTH_TOKEN"):

12
platformio/commands/org.py
View File

@ -55,6 +55,8 @@ def org_list(json_output):
orgs = client.list_orgs()
if json_output:
return click.echo(json.dumps(orgs))
if not orgs:
return click.echo("You do not have any organizations")
for org in orgs:
click.echo()
click.secho(org.get("orgname"), fg="cyan")
@ -76,16 +78,14 @@ def org_list(json_output):
@cli.command("update", short_help="Update organization")
@click.argument("orgname")
@click.option("--new-orgname")
@click.option(
"--new-orgname", callback=lambda _, __, value: validate_orgname(value),
)
@click.option("--email")
@click.option("--display-name",)
def org_update(orgname, **kwargs):
client = AccountClient()
org = next(
(org for org in client.list_orgs() if org.get("orgname") == orgname), None
)
if not org:
return click.ClickException("Organization '%s' not found" % orgname)
org = client.get_org(orgname)
del org["owners"]
new_org = org.copy()
if not any(kwargs.values()):

201
platformio/commands/team.py Normal file
View File

@ -0,0 +1,201 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# pylint: disable=unused-argument
import json
import re
import click
from tabulate import tabulate
from platformio.clients.account import AccountClient
def validate_orgname_teamname(value, teamname_validate=False):
if ":" not in value:
raise click.BadParameter(
"Please specify organization and team name in the next"
" format - orgname:teamname. For example, mycompany:DreamTeam"
)
teamname = str(value.strip().split(":", 1)[1])
if teamname_validate:
validate_teamname(teamname)
return value
def validate_teamname(value):
if not value:
return value
value = str(value).strip()
if not re.match(r"^[a-z\d](?:[a-z\d]|[\-_ ](?=[a-z\d])){0,19}$", value, flags=re.I):
raise click.BadParameter(
"Invalid team name format. "
"Team name must only contain alphanumeric characters, "
"single hyphens, underscores, spaces. It can not "
"begin or end with a hyphen or a underscore and must"
" not be longer than 20 characters."
)
return value
@click.group("team", short_help="Manage Teams")
def cli():
pass
@cli.command("create", short_help="Create a new team")
@click.argument(
"orgname_teamname",
metavar="ORGNAME:TEAMNAME",
callback=lambda _, __, value: validate_orgname_teamname(
value, teamname_validate=True
),
)
@click.option("--description",)
def team_create(orgname_teamname, description):
orgname, teamname = orgname_teamname.split(":", 1)
client = AccountClient()
client.create_team(orgname, teamname, description)
return click.secho(
"The team %s has been successfully created." % teamname, fg="green",
)
@cli.command("list", short_help="List teams")
@click.argument("orgname", required=False)
@click.option("--json-output", is_flag=True)
def team_list(orgname, json_output):
client = AccountClient()
data = {}
if not orgname:
for item in client.list_orgs():
teams = client.list_teams(item.get("orgname"))
data[item.get("orgname")] = teams
else:
teams = client.list_teams(orgname)
data[orgname] = teams
if json_output:
return click.echo(json.dumps(data[orgname] if orgname else data))
if not any(data.values()):
return click.secho("You do not have any teams.", fg="yellow")
for org_name in data:
for team in data[org_name]:
click.echo()
click.secho("%s:%s" % (org_name, team.get("name")), fg="cyan")
click.echo("-" * len("%s:%s" % (org_name, team.get("name"))))
table_data = []
if team.get("description"):
table_data.append(("Description:", team.get("description")))
table_data.append(
(
"Members:",
", ".join(
(member.get("username") for member in team.get("members"))
)
if team.get("members")
else "-",
)
)
click.echo(tabulate(table_data, tablefmt="plain"))
return click.echo()
@cli.command("update", short_help="Update team")
@click.argument(
"orgname_teamname",
metavar="ORGNAME:TEAMNAME",
callback=lambda _, __, value: validate_orgname_teamname(value),
)
@click.option(
"--name", callback=lambda _, __, value: validate_teamname(value),
)
@click.option("--description",)
def team_update(orgname_teamname, **kwargs):
orgname, teamname = orgname_teamname.split(":", 1)
client = AccountClient()
team = client.get_team(orgname, teamname)
del team["id"]
del team["members"]
new_team = team.copy()
if not any(kwargs.values()):
for field in team:
new_team[field] = click.prompt(
field.replace("_", " ").capitalize(), default=team[field]
)
if field == "name":
validate_teamname(new_team[field])
else:
new_team.update({key: value for key, value in kwargs.items() if value})
client.update_team(orgname, teamname, new_team)
return click.secho(
"The team %s has been successfully updated." % teamname, fg="green",
)
@cli.command("destroy", short_help="Destroy a team")
@click.argument(
"orgname_teamname",
metavar="ORGNAME:TEAMNAME",
callback=lambda _, __, value: validate_orgname_teamname(value),
)
def team_destroy(orgname_teamname):
orgname, teamname = orgname_teamname.split(":", 1)
click.confirm(
click.style(
"Are you sure you want to destroy the %s team?" % teamname, fg="yellow"
),
abort=True,
)
client = AccountClient()
client.destroy_team(orgname, teamname)
return click.secho(
"The team %s has been successfully destroyed." % teamname, fg="green",
)
@cli.command("add", short_help="Add a new member to team")
@click.argument(
"orgname_teamname",
metavar="ORGNAME:TEAMNAME",
callback=lambda _, __, value: validate_orgname_teamname(value),
)
@click.argument("username",)
def team_add_member(orgname_teamname, username):
orgname, teamname = orgname_teamname.split(":", 1)
client = AccountClient()
client.add_team_member(orgname, teamname, username)
return click.secho(
"The new member %s has been successfully added to the %s team."
% (username, teamname),
fg="green",
)
@cli.command("remove", short_help="Remove a member from team")
@click.argument(
"orgname_teamname",
metavar="ORGNAME:TEAMNAME",
callback=lambda _, __, value: validate_orgname_teamname(value),
)
@click.argument("username",)
def org_remove_owner(orgname_teamname, username):
orgname, teamname = orgname_teamname.split(":", 1)
client = AccountClient()
client.remove_team_member(orgname, teamname, username)
return click.secho(
"The %s member has been successfully removed from the %s team."
% (username, teamname),
fg="green",
)

106
tests/commands/test_orgs.py
View File

@ -37,7 +37,7 @@ def credentials():
}
def test_org_add(clirunner, credentials, validate_cliresult, isolated_pio_home):
def test_orgs(clirunner, credentials, validate_cliresult, isolated_pio_home):
try:
result = clirunner.invoke(
cmd_account,
@ -66,26 +66,11 @@ def test_org_add(clirunner, credentials, validate_cliresult, isolated_pio_home):
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) == 3
finally:
clirunner.invoke(cmd_account, ["logout"])
def test_org_list(clirunner, credentials, validate_cliresult, isolated_pio_home):
try:
result = clirunner.invoke(
cmd_account,
["login", "-u", credentials["login"], "-p", credentials["password"]],
)
validate_cliresult(result)
assert "Successfully logged in!" in result.output
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) == 3
assert len(json_result) >= 3
check = False
for org in json_result:
assert "orgname" in org
orgname = org["orgname"]
assert "displayname" in org
assert "email" in org
assert "owners" in org
@ -95,10 +80,41 @@ def test_org_list(clirunner, credentials, validate_cliresult, isolated_pio_home)
assert "firstname" in owner
assert "lastname" in owner
assert check
result = clirunner.invoke(cmd_org, ["add", orgname, "ivankravets"],)
validate_cliresult(result)
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) >= 3
check = False
for item in json_result:
if item["orgname"] != orgname:
continue
for owner in item.get("owners"):
check = owner["username"] == "ivankravets" if not check else True
assert check
result = clirunner.invoke(cmd_org, ["remove", orgname, "ivankravets"],)
validate_cliresult(result)
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) >= 3
check = False
for item in json_result:
if item["orgname"] != orgname:
continue
for owner in item.get("owners"):
check = owner["username"] == "ivankravets" if not check else True
assert not check
finally:
clirunner.invoke(cmd_account, ["logout"])
@pytest.mark.skip
def test_org_update(clirunner, credentials, validate_cliresult, isolated_pio_home):
try:
result = clirunner.invoke(
@ -111,7 +127,7 @@ def test_org_update(clirunner, credentials, validate_cliresult, isolated_pio_hom
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) == 3
assert len(json_result) >= 3
org = json_result[0]
assert "orgname" in org
assert "displayname" in org
@ -137,55 +153,3 @@ def test_org_update(clirunner, credentials, validate_cliresult, isolated_pio_hom
assert json.loads(result.output.strip()) == json_result
finally:
clirunner.invoke(cmd_account, ["logout"])
def test_org_owner(clirunner, credentials, validate_cliresult, isolated_pio_home):
try:
result = clirunner.invoke(
cmd_account,
["login", "-u", credentials["login"], "-p", credentials["password"]],
)
validate_cliresult(result)
assert "Successfully logged in!" in result.output
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) == 3
org = json_result[0]
assert "orgname" in org
assert "displayname" in org
assert "email" in org
assert "owners" in org
result = clirunner.invoke(cmd_org, ["add", org["orgname"], "ivankravets"],)
validate_cliresult(result)
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) == 3
check = False
for item in json_result:
if item["orgname"] != org["orgname"]:
continue
for owner in item.get("owners"):
check = owner["username"] == "ivankravets" if not check else True
assert check
result = clirunner.invoke(cmd_org, ["remove", org["orgname"], "ivankravets"],)
validate_cliresult(result)
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) == 3
check = False
for item in json_result:
if item["orgname"] != org["orgname"]:
continue
for owner in item.get("owners"):
check = owner["username"] == "ivankravets" if not check else True
assert not check
finally:
clirunner.invoke(cmd_account, ["logout"])

158
tests/commands/test_teams.py Normal file
View File

@ -0,0 +1,158 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import os
import time
import pytest
from platformio.commands.account import cli as cmd_account
from platformio.commands.org import cli as cmd_org
from platformio.commands.team import cli as cmd_team
pytestmark = pytest.mark.skipif(
not (
os.environ.get("PLATFORMIO_TEST_ACCOUNT_LOGIN")
and os.environ.get("PLATFORMIO_TEST_ACCOUNT_PASSWORD")
),
reason="requires PLATFORMIO_TEST_ACCOUNT_LOGIN, PLATFORMIO_TEST_ACCOUNT_PASSWORD environ variables",
)
@pytest.fixture(scope="session")
def credentials():
return {
"login": os.environ["PLATFORMIO_TEST_ACCOUNT_LOGIN"],
"password": os.environ["PLATFORMIO_TEST_ACCOUNT_PASSWORD"],
}
def test_teams(clirunner, credentials, validate_cliresult, isolated_pio_home):
orgname = ""
teamname = "test-" + str(int(time.time() * 1000))
try:
result = clirunner.invoke(
cmd_account,
["login", "-u", credentials["login"], "-p", credentials["password"]],
)
validate_cliresult(result)
assert "Successfully logged in!" in result.output
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
if len(json_result) < 3:
for i in range(3 - len(json_result)):
result = clirunner.invoke(
cmd_org,
[
"create",
"%s-%s" % (i, credentials["login"]),
"--email",
"test@test.com",
"--display-name",
"TEST ORG %s" % i,
],
)
validate_cliresult(result)
result = clirunner.invoke(cmd_org, ["list", "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) >= 3
orgname = json_result[0].get("orgname")
result = clirunner.invoke(
cmd_team,
[
"create",
"%s:%s" % (orgname, teamname),
"--description",
"team for CI test",
],
)
validate_cliresult(result)
result = clirunner.invoke(cmd_team, ["list", "%s" % orgname, "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) >= 1
check = False
for team in json_result:
assert team["id"]
assert team["name"]
if team["name"] == teamname:
check = True
assert "description" in team
assert "members" in team
assert check
result = clirunner.invoke(
cmd_team, ["add", "%s:%s" % (orgname, teamname), credentials["login"]],
)
validate_cliresult(result)
result = clirunner.invoke(cmd_team, ["list", "%s" % orgname, "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
check = False
for team in json_result:
assert team["id"]
assert team["name"]
assert "description" in team
assert "members" in team
if (
len(team["members"]) > 0
and team["members"][0]["username"] == credentials["login"]
):
check = True
assert check
result = clirunner.invoke(
cmd_team, ["remove", "%s:%s" % (orgname, teamname), credentials["login"]],
)
validate_cliresult(result)
result = clirunner.invoke(cmd_team, ["list", "%s" % orgname, "--json-output"],)
validate_cliresult(result)
result = clirunner.invoke(
cmd_team,
[
"update",
"%s:%s" % (orgname, teamname),
"--description",
"Updated Description",
],
)
validate_cliresult(result)
result = clirunner.invoke(cmd_team, ["list", "%s" % orgname, "--json-output"],)
validate_cliresult(result)
json_result = json.loads(result.output.strip())
assert len(json_result) >= 1
check = False
for team in json_result:
assert team["id"]
assert team["name"]
assert "description" in team
if team.get("description") == "Updated Description":
check = True
assert "members" in team
assert check
finally:
clirunner.invoke(
cmd_team, ["destroy", "%s:%s" % (orgname, teamname),],
)
clirunner.invoke(cmd_account, ["logout"])