Vašek Šraier
6 years ago
commit
4848f02484
4 changed files with 189 additions and 0 deletions
@ -0,0 +1,5 @@ |
|||||
|
click/ |
||||
|
swagger_client/ |
||||
|
tmp/ |
||||
|
.idea/ |
||||
|
.idea/ |
@ -0,0 +1,31 @@ |
|||||
|
GITEA_URL=https://git.vakabus.cz/swagger.v1.json |
||||
|
BUILDDIR=tmp |
||||
|
|
||||
|
build: swagger_client click clean_builddir |
||||
|
|
||||
|
swagger_client: |
||||
|
mkdir -p "${BUILDDIR}" |
||||
|
swagger-codegen generate -i "${GITEA_URL}" -l python -o "${BUILDDIR}/gitea_api" |
||||
|
mv "${BUILDDIR}/gitea_api/swagger_client" . |
||||
|
rm -rf "${BUILDDIR}/gitea_api" |
||||
|
|
||||
|
# Protoze ten generator psali nejaci dementi, neumi to cyklicke reference |
||||
|
# je proto potreba zakomentovat dva importy |
||||
|
sed -i 's/from swagger_client\.models\.gpg_key import GPGKey/#from swagger_client.models.gpg_key import GPGKey/' swagger_client/models/gpg_key.py |
||||
|
sed -i 's/from swagger_client\.models\.repository import Repository/#from swagger_client.models.repository import Repository/' swagger_client/models/repository.py |
||||
|
|
||||
|
click: |
||||
|
mkdir -p "${BUILDDIR}" |
||||
|
cd ${BUILDDIR}; git clone --depth=1 https://github.com/pallets/click.git |
||||
|
mv ${BUILDDIR}/click/click click |
||||
|
|
||||
|
.PHONY: clean_builddir |
||||
|
clean_builddir: |
||||
|
rm -rf ${BUILDDIR} |
||||
|
|
||||
|
.PHONY: clean |
||||
|
clean: |
||||
|
rm -rf swagger_client/ |
||||
|
rm -rf click/ |
||||
|
|
||||
|
|
@ -0,0 +1,2 @@ |
|||||
|
# Odkomentovanim radku nize je mozne nastavit vychozi server Gitei |
||||
|
# HOST = 'https://gitea.url/api/v1' |
@ -0,0 +1,151 @@ |
|||||
|
from swagger_client.api.admin_api import AdminApi |
||||
|
from swagger_client.api.user_api import UserApi |
||||
|
from swagger_client.api.organization_api import OrganizationApi |
||||
|
from swagger_client.api_client import ApiClient |
||||
|
from swagger_client.models import CreateUserOption, Team, User, Organization |
||||
|
from swagger_client.rest import ApiException |
||||
|
from sys import stderr |
||||
|
from typing import * |
||||
|
import click |
||||
|
import secrets |
||||
|
import termcolor |
||||
|
|
||||
|
|
||||
|
def text_red(text) -> str: |
||||
|
return termcolor.colored(text, 'red') |
||||
|
|
||||
|
|
||||
|
def text_green(text) -> str: |
||||
|
return termcolor.colored(text, 'green') |
||||
|
|
||||
|
|
||||
|
def text_orange(text) -> str: |
||||
|
return termcolor.colored(text, 'yellow') |
||||
|
|
||||
|
|
||||
|
def does_user_exist(user_api: UserApi, username: str) -> bool: |
||||
|
try: |
||||
|
user: User = user_api.user_get(username) |
||||
|
return user is not None |
||||
|
except ApiException as e: |
||||
|
if e.status == 404: |
||||
|
# uzivatel nebyl nalezen |
||||
|
return False |
||||
|
else: |
||||
|
raise e |
||||
|
|
||||
|
|
||||
|
def does_organization_exist(org_api: OrganizationApi, orgname: str) -> bool: |
||||
|
try: |
||||
|
org: Organization = org_api.org_get(orgname) |
||||
|
return org is not None |
||||
|
except ApiException as e: |
||||
|
if e.status == 404: |
||||
|
# organizace nebyla nalezena |
||||
|
return False |
||||
|
else: |
||||
|
raise e |
||||
|
|
||||
|
|
||||
|
def does_organization_have_org_team(org_api: OrganizationApi, orgname: str) -> bool: |
||||
|
teams: List[Team] = org_api.org_list_teams(orgname) |
||||
|
for t in teams: |
||||
|
if 'org' == t.name: |
||||
|
return True |
||||
|
return False |
||||
|
|
||||
|
|
||||
|
def get_org_team_id(org_api: OrganizationApi, orgname: str) -> int: |
||||
|
teams: List[Team] = org_api.org_list_teams(orgname) |
||||
|
for t in teams: |
||||
|
if 'org' == t.name: |
||||
|
return t.id |
||||
|
raise AssertionError('Tento kod mel byt unreachable. Zmizel tym orgu za behu skriptu?') |
||||
|
|
||||
|
|
||||
|
def validate(func_check, message: str): |
||||
|
print(message, end="") |
||||
|
try: |
||||
|
if func_check(): |
||||
|
print(text_green(' [OK]')) |
||||
|
return |
||||
|
except ApiException as e: |
||||
|
print(text_red(' [FAIL]')) |
||||
|
print(text_red(e.body)) |
||||
|
print("Request se nepodaril! Duvod:", e.reason, file=stderr) |
||||
|
exit(1) |
||||
|
|
||||
|
print(text_red(' [FAIL]')) |
||||
|
print("Neocekavany stav systemu!", file=stderr) |
||||
|
exit(1) |
||||
|
|
||||
|
|
||||
|
def checked_api_action(func_action, message: str): |
||||
|
print(message, end='') |
||||
|
try: |
||||
|
res = func_action() |
||||
|
print(text_green('[OK]')) |
||||
|
return res |
||||
|
except ApiException as e: |
||||
|
print(text_red('[FAIL]')) |
||||
|
print(text_red('Server odpovedel:')) |
||||
|
print(text_red(e.body)) |
||||
|
exit(1) |
||||
|
|
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
# default configuration |
||||
|
HOST = "https://git.vakabus.cz/api/v1" |
||||
|
try: |
||||
|
with open('config.py') as f: |
||||
|
conf = f.read() |
||||
|
exec(conf) |
||||
|
except Exception: |
||||
|
print(text_orange('Nepodarilo se nacist konfiguraci ze souboru \'config.py\'')) |
||||
|
|
||||
|
|
||||
|
@click.command() |
||||
|
@click.option('-a', '--admin-token', help=f'Admin access token', required=True) |
||||
|
@click.option('-h', '--host', help=f'Gitea URL (default: {HOST})') |
||||
|
@click.option('-e','--email', help='Emailova adresa noveho uzivatele.', required=True) |
||||
|
@click.option('-u','--username', help='Uzivatelske jmeno noveho uzivatele.', required=True) |
||||
|
@click.option('-f','--fullname', help='Plne jmeno noveho uzivatele.') |
||||
|
@click.option('-s','--seminar', help='Seminar, kde uzivatel orguje.', required=True) |
||||
|
def register(username: str, seminar: str, fullname: str, email: str, host, admin_token): |
||||
|
""" |
||||
|
Nastroj pro automatizaci zakladani uctu na serveru Gitea. |
||||
|
|
||||
|
Prvni probehne kontrola stavu systemu, zda je mozne uzivatele bezpecne zalozit. Pote se provede pokus o zalozeni |
||||
|
nasledovan pridanim uzivatele do skupiny 'org' v pozadovane organizaci (jmeno dle seminare). |
||||
|
|
||||
|
Vytvoreny uzivatel ma nastavene nahodne heslo. Zadny email o vytvoreni uctu neni zaslan. |
||||
|
""" |
||||
|
|
||||
|
api_client = ApiClient() |
||||
|
api_client.configuration.host = HOST |
||||
|
api_client.configuration.api_key = {'token': admin_token} |
||||
|
|
||||
|
admin_api = AdminApi(api_client) |
||||
|
user_api = UserApi(api_client) |
||||
|
org_api = OrganizationApi(api_client) |
||||
|
|
||||
|
# kontrola predpokladu o stavu systemu |
||||
|
validate(lambda: not does_user_exist(user_api, username), "Uzivatelske jmeno je volne...") |
||||
|
validate(lambda: does_organization_exist(org_api, seminar), "Organizace pro seminar existuje...") |
||||
|
validate(lambda: does_organization_have_org_team(org_api, seminar), "Team se jmenem 'org' v danem seminari existuje...") |
||||
|
print(text_green('Zakladni predpoklady pro uspesne zalozeni uctu splneny!\n')) |
||||
|
|
||||
|
# nasbirame vsechny potrebne informace |
||||
|
org_team_id = get_org_team_id(org_api, seminar) |
||||
|
|
||||
|
# pripravime si uzivatele na zalozeni |
||||
|
create_user_req = CreateUserOption(email=email, full_name=fullname, username=username, password=secrets.token_urlsafe(20)) |
||||
|
|
||||
|
# provedeme zmeny v systemu |
||||
|
user = checked_api_action(lambda: admin_api.admin_create_user(body=create_user_req), "Zakladam noveho uzivatele...") |
||||
|
checked_api_action(lambda: org_api.org_add_team_member(org_team_id, username), f"Pridavam do teamu 'org' v seminari {seminar}...") |
||||
|
print(text_green('Uzivatel zalozen!')) |
||||
|
|
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
register() |
Loading…
Reference in new issue