jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

🔧 fix(test_user.py): update imports to match changes in project structure

Browse source 
🔧 fix(test_user.py): update fixtures to use session_getter and get_db_manager functions for session management
✨ feat(test_user.py): add support for creating and managing database sessions using session_getter and get_db_manager functions
🔧 fix(test_user.py): update test_user_waiting_for_approval function to use session_getter and get_db_manager functions for session management
🔧 fix(test_user.py): update test_data_consistency_after_delete function to use session_getter and get_db_manager functions for session management
🔧 fix(test_user.py): update test_inactive_user function to use session_getter and get_db_manager functions for session management
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-09-20 18:35:38 -03:00
commit b4f48f1ad9
1 changed files with 43 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

78
tests/test_user.py
View file

@ -2,20 +2,22 @@ from datetime import datetime
from langflow.services.auth.utils import create_super_user, get_password_hash from langflow.services.auth.utils import create_super_user, get_password_hash
from langflow.services.database.models.user.user import User from langflow.services.database.models.user.user import User
from langflow.services.getters import get_settings_manager from langflow.services.database.utils import session_getter
from langflow.services.getters import get_db_manager, get_settings_manager
import pytest import pytest
from langflow.services.database.models.user import UserUpdate from langflow.services.database.models.user import UserUpdate
@pytest.fixture @pytest.fixture
def super_user(client, session): def super_user(client):
settings_manager = get_settings_manager() settings_manager = get_settings_manager()
auth_settings = settings_manager.auth_settings auth_settings = settings_manager.auth_settings
return create_super_user( with session_getter(get_db_manager()) as session:
db=session, return create_super_user(
username=auth_settings.SUPERUSER, db=session,
password=auth_settings.SUPERUSER_PASSWORD, username=auth_settings.SUPERUSER,
) password=auth_settings.SUPERUSER_PASSWORD,
)
@pytest.fixture @pytest.fixture
@ -34,29 +36,34 @@ def super_user_headers(client, super_user):
@pytest.fixture @pytest.fixture
def deactivated_user(session): def deactivated_user():
user = User( with session_getter(get_db_manager()) as session:
username="deactivateduser", user = User(
password=get_password_hash("testpassword"), username="deactivateduser",
is_active=False, password=get_password_hash("testpassword"),
is_superuser=False, is_active=False,
last_login_at=datetime.now(), is_superuser=False,
) last_login_at=datetime.now(),
session.add(user) )
session.commit() session.add(user)
session.commit()
session.refresh(user)
return user return user
def test_user_waiting_for_approval(client, session): def test_user_waiting_for_approval(
client,
):
# Create a user that is not active and has never logged in # Create a user that is not active and has never logged in
user = User( with session_getter(get_db_manager()) as session:
username="waitingforapproval", user = User(
password=get_password_hash("testpassword"), username="waitingforapproval",
is_active=False, password=get_password_hash("testpassword"),
last_login_at=None, is_active=False,
) last_login_at=None,
session.add(user) )
session.commit() session.add(user)
session.commit()
login_data = {"username": "waitingforapproval", "password": "testpassword"} login_data = {"username": "waitingforapproval", "password": "testpassword"}
response = client.post("/api/v1/login", data=login_data) response = client.post("/api/v1/login", data=login_data)
@ -106,16 +113,17 @@ def test_data_consistency_after_delete(client, test_user, super_user_headers):
assert all(user["id"] != user_id for user in response.json()["users"]) assert all(user["id"] != user_id for user in response.json()["users"])
def test_inactive_user(client, session): def test_inactive_user(client):
# Create a user that is not active and has a last_login_at value # Create a user that is not active and has a last_login_at value
user = User( with session_getter(get_db_manager()) as session:
username="inactiveuser", user = User(
password=get_password_hash("testpassword"), username="inactiveuser",
is_active=False, password=get_password_hash("testpassword"),
last_login_at="2023-01-01T00:00:00", # Set to a valid datetime string is_active=False,
) last_login_at="2023-01-01T00:00:00", # Set to a valid datetime string
session.add(user) )
session.commit() session.add(user)
session.commit()
login_data = {"username": "inactiveuser", "password": "testpassword"} login_data = {"username": "inactiveuser", "password": "testpassword"}
response = client.post("/api/v1/login", data=login_data) response = client.post("/api/v1/login", data=login_data)