81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
import logging
|
|
from time import sleep
|
|
import pytest
|
|
|
|
import requests
|
|
from fastapi.testclient import TestClient
|
|
|
|
from .main import app
|
|
from app.authentik.api import Authentik
|
|
from app.authentik.models import AuthentikUser
|
|
from app.authentik.settings import AuthentikSettings
|
|
from app.consumer.wekan.models import User as WekanUser
|
|
from app.consumer.wekan.api import WekanApi
|
|
|
|
from app.authentik.test_authentik import settings
|
|
from app.consumer.wekan.test_wekan import settings as wekan_settings
|
|
client = TestClient(app)
|
|
|
|
|
|
@pytest.fixture()
|
|
def wekan(wekan_settings):
|
|
w = None
|
|
try:
|
|
r = requests.post("http://localhost:3000/users/register", json={
|
|
"username": "api", "password": "foobar123", "email": "foo@example.org"})
|
|
w = WekanApi(wekan_settings)
|
|
except Exception as e:
|
|
logging.error(e)
|
|
return w
|
|
|
|
|
|
@pytest.fixture()
|
|
def authentik(settings: AuthentikSettings):
|
|
a = Authentik(settings)
|
|
try:
|
|
r = a.create_web_hook(
|
|
hook_endpoint="http://172.17.0.1:8000/authentik/hook/") # docker localhost
|
|
except Exception as e:
|
|
logging.error(e)
|
|
return a
|
|
|
|
|
|
@pytest.fixture()
|
|
def authentik_user(authentik):
|
|
user = authentik.create_user(AuthentikUser(username="foobar", name="Foo Bar", email="foo@bar.com"))
|
|
yield user
|
|
authentik.delete_user(user)
|
|
|
|
|
|
|
|
def test_create_user(mocker, authentik_user: AuthentikUser, wekan: WekanApi):
|
|
# Actually authentik user creation should already trigger the hook, but in authentik it doesn't trigger when come from api
|
|
# mock = mocker.patch("app.event_controller.EventController.handle_model_created_event")
|
|
authentik_message = {"model": {"pk": authentik_user.pk, "app": "authentik_core", "name": authentik_user.name,
|
|
"model_name": "user"}, "http_request": {"args": {}, "path": "/api/v3/core/users/", "method": "POST"}}
|
|
response = client.post("/authentik/hook/", json=authentik_message)
|
|
assert response.status_code == 200
|
|
wu = wekan.get_user(authentik_user.username)
|
|
assert not wu == None
|
|
assert wu.username == authentik_user.username
|
|
assert authentik_user.email in [i.address for i in wu.emails]
|
|
wekan.delete_user(wu.id)
|
|
|
|
|
|
@pytest.mark.skip()
|
|
def test_create_user_with_same_email(wekan_api, authentik_api):
|
|
logging.error(
|
|
"authentik notifcation rule doesn't work with api?? , create two user with identical email in authentik")
|
|
assert False
|
|
|
|
def test_user_already_exists_excepts(authentik_user: AuthentikUser, wekan: WekanApi):
|
|
authentik_message = {"model": {"pk": authentik_user.pk, "app": "authentik_core", "name": authentik_user.name,
|
|
"model_name": "user"}, "http_request": {"args": {}, "path": "/api/v3/core/users/", "method": "POST"}}
|
|
response = client.post("/authentik/hook/", json=authentik_message)
|
|
response = client.post("/authentik/hook/", json=authentik_message)
|
|
assert response.status_code == 200
|
|
errors = client.get("/authentik/errors/")
|
|
assert errors.status_code == 200
|
|
assert errors.json() == ["[Wekan] User already exists"] # TODO introduce error models
|
|
wekan.delete_user(wekan.get_user(authentik_user.username).id)
|