create user authentik

This commit is contained in:
Philipp Rothmann 2022-03-06 12:31:12 +01:00
parent 00de186ad4
commit cc1b8eeb01
5 changed files with 53 additions and 24 deletions

View file

@ -7,6 +7,10 @@ init:
run: run:
./env/bin/uvicorn app.main:app --reload --host 0.0.0.0 ./env/bin/uvicorn app.main:app --reload --host 0.0.0.0
run integration:
docker-compose -f compose.authentik.yml up -d
./env/bin/uvicorn app.main:app --reload --host 0.0.0.0
test: test:
./env/bin/pytest app ./env/bin/pytest app

View file

@ -1,6 +1,6 @@
# integrate # integrate
does what nobody else want's to do. integrate all the api's
## Development ## Development
@ -16,3 +16,4 @@ make run
https://pydantic-docs.helpmanual.io/ https://pydantic-docs.helpmanual.io/
https://jsontopydantic.com/ https://jsontopydantic.com/
https://pydantic-docs.helpmanual.io/datamodel_code_generator/

View file

@ -1,8 +1,8 @@
from typing import Dict from typing import Dict, Optional
import requests import requests
from requests import Request from requests import Request
from app.authentik.models import User from .models import User
class Authentik: class Authentik:
@ -132,15 +132,15 @@ class Authentik:
raise Exception(r.status_code, r.url, r.text) raise Exception(r.status_code, r.url, r.text)
def get_user(self, user: User): def get_user(self, user: User) -> Optional[User]:
if user.pk: if user.pk:
r = self.get(f"core/users/{user.pk}").json() r = self.get(f"core/users/{user.pk}").json()
else: else:
r = self.get(f"core/users/?search={user.username}").json() r = self.get(f"core/users/?search={user.username}").json()
if len(r["results"]) == 0 :
return None
if len(r["results"]) > 1 : if len(r["results"]) > 1 :
raise Exception("More than one user with that username", r) raise Exception("More than one user with that username", r)
if len(r["results"]) == 0 :
raise Exception("No user with that username", r)
r = r["results"][0] r = r["results"][0]
if "pk" in r: if "pk" in r:
@ -155,5 +155,13 @@ class Authentik:
raise Exception(r) raise Exception(r)
def delete_user(self, user: User): def delete_user(self, user: User):
r = self.delete(f"core/users/{user.username}") if user.pk == None:
print(r.json()) raise Exception("need pk")
r = self.delete(f"core/users/{user.pk}").json()
if("detail" in r):
if r["detail"] == "Not found.":
return None
print(r)

View file

@ -20,8 +20,8 @@ class GroupsObjItem(BaseModel):
pk: str pk: str
name: str name: str
is_superuser: bool is_superuser: bool
parent: str parent: Optional[str]
parent_name: str parent_name: Optional[str]
users: List[int] users: List[int]
attributes: Dict[str, Any] attributes: Dict[str, Any]
users_obj: List[UsersObjItem] users_obj: List[UsersObjItem]
@ -33,7 +33,7 @@ class User(BaseModel):
is_active: bool = None is_active: bool = None
last_login: Optional[str] = None last_login: Optional[str] = None
is_superuser: Optional[bool] = None is_superuser: Optional[bool] = None
groups: List[str] groups: Optional[List[str]]
groups_obj: Optional[List[GroupsObjItem]] = None groups_obj: Optional[List[GroupsObjItem]] = None
email: str email: str
avatar: Optional[str] = None avatar: Optional[str] = None

View file

@ -1,12 +1,29 @@
import email import email
from app.authentik.models import User
from .authentik import Authentik from .authentik import Authentik
from .models import User
import pytest
def test_connection(): @pytest.fixture
a = Authentik(base="http://localhost:9000/", token="foobar123", ) def api() -> Authentik:
return Authentik(base="http://localhost:9000/", token="foobar123", )
def test_get_user_by_username(api: Authentik):
u = User(username="akadmin",
name="",
groups=[],
email="",
is_active=True,
attributes={}
)
u = api.get_user(u)
assert not u == None
assert u.username == "akadmin"
def test_create_user(api: Authentik):
# res = a.create_web_hook(hook_endpoint="http://172.17.0.1:8000") # res = a.create_web_hook(hook_endpoint="http://172.17.0.1:8000")
u = User(username="banane", u = User(username="banane",
name="banane", name="banane",
groups=[], groups=[],
@ -14,12 +31,11 @@ def test_connection():
is_active=True, is_active=True,
attributes={} attributes={}
) )
if(a.get_user(u)): if(api.get_user(u)):
print("delete") api.delete_user(api.get_user(u))
a.delete(u)
assert False
c = a.create_user(u)
assert api.get_user(u) == None
c = api.create_user(u)
assert u.username == c.username assert u.username == c.username
assert not c.pk == None assert not c.pk == None