From 8411922a07d8376c704062e03ed93a04f006d860 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Fri, 25 Aug 2023 10:17:46 -0300 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(test=5Fuser.py):=20add=20tests?= =?UTF-8?q?=20for=20user-related=20functionalities=20=F0=9F=90=9B=20fix(te?= =?UTF-8?q?st=5Fuser.py):=20fix=20typo=20in=20test=5Fuser=5Fwaiting=5Ffor?= =?UTF-8?q?=5Fapproval=20function=20name=20=F0=9F=90=9B=20fix(test=5Fuser.?= =?UTF-8?q?py):=20fix=20typo=20in=20test=5Fdeactivated=5Fuser=5Fcannot=5Fa?= =?UTF-8?q?ccess=20function=20name=20=F0=9F=90=9B=20fix(test=5Fuser.py):?= =?UTF-8?q?=20fix=20typo=20in=20test=5Fdata=5Fconsistency=5Fafter=5Fupdate?= =?UTF-8?q?=20function=20name=20=F0=9F=90=9B=20fix(test=5Fuser.py):=20fix?= =?UTF-8?q?=20typo=20in=20test=5Fdata=5Fconsistency=5Fafter=5Fdelete=20fun?= =?UTF-8?q?ction=20name=20=F0=9F=90=9B=20fix(test=5Fuser.py):=20fix=20typo?= =?UTF-8?q?=20in=20test=5Finactive=5Fuser=20function=20name=20=F0=9F=90=9B?= =?UTF-8?q?=20fix(test=5Fuser.py):=20fix=20typo=20in=20test=5Fnormal=5Fuse?= =?UTF-8?q?r=5Fcant=5Fread=5Fall=5Fusers=20function=20name=20=F0=9F=90=9B?= =?UTF-8?q?=20fix(test=5Fuser.py):=20fix=20typo=20in=20test=5Fpatch=5Fuser?= =?UTF-8?q?=5Fwrong=5Fid=20function=20name=20=F0=9F=90=9B=20fix(test=5Fuse?= =?UTF-8?q?r.py):=20fix=20typo=20in=20test=5Fdelete=5Fuser=5Fwrong=5Fid=20?= =?UTF-8?q?function=20name=20=F0=9F=90=9B=20fix(test=5Fuser.py):=20fix=20t?= =?UTF-8?q?ypo=20in=20test=5Fnormal=5Fuser=5Fcant=5Fdelete=5Fuser=20functi?= =?UTF-8?q?on=20name=20=E2=9C=A8=20feat(test=5Fuser.py):=20add=20test=5Fad?= =?UTF-8?q?d=5Fsuper=5Fuser=5Ffor=5Ftesting=5Fpurposes=5Fdelete=5Fme=5Fbef?= =?UTF-8?q?ore=5Fmerge=5Finto=5Fdev=20function?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_user.py | 248 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 248 insertions(+) create mode 100644 tests/test_user.py diff --git a/tests/test_user.py b/tests/test_user.py new file mode 100644 index 000000000..f8d4ff788 --- /dev/null +++ b/tests/test_user.py @@ -0,0 +1,248 @@ +from datetime import datetime +from langflow.services.auth.utils import create_super_user, get_password_hash + +from langflow.services.database.models.user.user import User +from langflow.services.utils import get_settings_manager +import pytest +from langflow.services.database.models.user import UserCreate, UserUpdate + + +@pytest.fixture +def test_user(client): + user_data = UserCreate( + username="testuser", + password="testpassword", + ) + response = client.post("/api/v1/user", json=user_data.dict()) + return response.json() + + +@pytest.fixture(scope="function") +def active_user(session): + user = User( + username="activeuser", + password=get_password_hash( + "testpassword" + ), # Assuming password needs to be hashed + is_active=True, + is_superuser=False, + ) + session.add(user) + session.commit() + return user + + +@pytest.fixture +def logged_in_headers(client, active_user): + login_data = {"username": active_user.username, "password": "testpassword"} + response = client.post("/api/v1/login", data=login_data) + assert response.status_code == 200 + tokens = response.json() + a_token = tokens["access_token"] + return {"Authorization": f"Bearer {a_token}"} + + +@pytest.fixture +def super_user(client, session): + return create_super_user(session) + + +@pytest.fixture +def super_user_headers(client, super_user): + settings_manager = get_settings_manager() + auth_settings = settings_manager.auth_settings + login_data = { + "username": auth_settings.FIRST_SUPERUSER, + "password": auth_settings.FIRST_SUPERUSER_PASSWORD, + } + response = client.post("/api/v1/login", data=login_data) + assert response.status_code == 200 + tokens = response.json() + a_token = tokens["access_token"] + return {"Authorization": f"Bearer {a_token}"} + + +@pytest.fixture +def deactivated_user(session): + user = User( + username="deactivateduser", + password=get_password_hash("testpassword"), + is_active=False, + is_superuser=False, + last_login_at=datetime.now(), + ) + session.add(user) + session.commit() + return user + + +def test_user_waiting_for_approval(client, session): + # Create a user that is not active and has never logged in + user = User( + username="waitingforapproval", + password=get_password_hash("testpassword"), + is_active=False, + last_login_at=None, + ) + session.add(user) + session.commit() + + login_data = {"username": "waitingforapproval", "password": "testpassword"} + response = client.post("/api/v1/login", data=login_data) + assert response.status_code == 400 + assert response.json()["detail"] == "Waiting for approval" + + +def test_deactivated_user_cannot_login(client, deactivated_user): + login_data = {"username": deactivated_user.username, "password": "testpassword"} + response = client.post("/api/v1/login", data=login_data) + assert response.status_code == 400, response.json() + assert response.json()["detail"] == "Inactive user" + + +def test_deactivated_user_cannot_access(client, deactivated_user, logged_in_headers): + # Assuming the headers for deactivated_user + response = client.get("/api/v1/users", headers=logged_in_headers) + assert response.status_code == 400, response.json() + assert response.json()["detail"] == "The user doesn't have enough privileges" + + +def test_data_consistency_after_update(client, active_user, logged_in_headers): + user_id = active_user.id + update_data = UserUpdate(username="newname") + + response = client.patch( + f"/api/v1/user/{user_id}", json=update_data.dict(), headers=logged_in_headers + ) + assert response.status_code == 200 + + # Fetch the updated user from the database + response = client.get("/api/v1/user", headers=logged_in_headers) + assert response.json()["username"] == "newname", response.json() + + +def test_data_consistency_after_delete(client, test_user, super_user_headers): + user_id = test_user["id"] + response = client.delete(f"/api/v1/user/{user_id}", headers=super_user_headers) + assert response.status_code == 200 + + # Attempt to fetch the deleted user from the database + response = client.get("/api/v1/users", headers=super_user_headers) + assert response.status_code == 200 + assert all(user["id"] != user_id for user in response.json()["users"]) + + +def test_inactive_user(client, session): + # Create a user that is not active and has a last_login_at value + user = User( + username="inactiveuser", + password=get_password_hash("testpassword"), + is_active=False, + last_login_at="2023-01-01T00:00:00", # Set to a valid datetime string + ) + session.add(user) + session.commit() + + login_data = {"username": "inactiveuser", "password": "testpassword"} + response = client.post("/api/v1/login", data=login_data) + assert response.status_code == 400 + assert response.json()["detail"] == "Inactive user" + + +def test_add_user(client, test_user): + assert test_user["username"] == "testuser" + + +# This is not used in the Frontend at the moment +# def test_read_current_user(client: TestClient, active_user): +# # First we need to login to get the access token +# login_data = {"username": "testuser", "password": "testpassword"} +# response = client.post("/api/v1/login", data=login_data) +# assert response.status_code == 200 + +# headers = {"Authorization": f"Bearer {response.json()['access_token']}"} + +# response = client.get("/api/v1/user", headers=headers) +# assert response.status_code == 200, response.json() +# assert response.json()["username"] == "testuser" + + +def test_read_all_users(client, super_user_headers): + response = client.get("/api/v1/users", headers=super_user_headers) + assert response.status_code == 200, response.json() + assert isinstance(response.json()["users"], list) + + +def test_normal_user_cant_read_all_users(client, logged_in_headers): + response = client.get("/api/v1/users", headers=logged_in_headers) + assert response.status_code == 400, response.json() + assert response.json() == {"detail": "The user doesn't have enough privileges"} + + +def test_patch_user(client, active_user, logged_in_headers): + user_id = active_user.id + update_data = UserUpdate( + username="newname", + ) + + response = client.patch( + f"/api/v1/user/{user_id}", json=update_data.dict(), headers=logged_in_headers + ) + assert response.status_code == 200, response.json() + + +def test_patch_user_wrong_id(client, active_user, logged_in_headers): + user_id = "wrong_id" + update_data = UserUpdate( + username="newname", + ) + + response = client.patch( + f"/api/v1/user/{user_id}", json=update_data.dict(), headers=logged_in_headers + ) + assert response.status_code == 422, response.json() + assert response.json() == { + "detail": [ + { + "loc": ["path", "user_id"], + "msg": "value is not a valid uuid", + "type": "type_error.uuid", + } + ] + } + + +def test_delete_user(client, test_user, super_user_headers): + user_id = test_user["id"] + response = client.delete(f"/api/v1/user/{user_id}", headers=super_user_headers) + assert response.status_code == 200 + assert response.json() == {"detail": "User deleted"} + + +def test_delete_user_wrong_id(client, test_user, super_user_headers): + user_id = "wrong_id" + response = client.delete(f"/api/v1/user/{user_id}", headers=super_user_headers) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "loc": ["path", "user_id"], + "msg": "value is not a valid uuid", + "type": "type_error.uuid", + } + ] + } + + +def test_normal_user_cant_delete_user(client, test_user, logged_in_headers): + user_id = test_user["id"] + response = client.delete(f"/api/v1/user/{user_id}", headers=logged_in_headers) + assert response.status_code == 400 + assert response.json() == {"detail": "The user doesn't have enough privileges"} + + +# If you still want to test the superuser endpoint +def test_add_super_user_for_testing_purposes_delete_me_before_merge_into_dev(client): + response = client.post("/api/v1/super_user") + assert response.status_code == 200 + assert response.json()["username"] == "superuser"