import pytest
from pathlib import Path
import json
from retoors.main import create_app
from retoors.services.config_service import ConfigService
from pytest_mock import MockerFixture # Import MockerFixture
import datetime # Import datetime
@pytest.fixture
def create_app_instance():
"""Fixture to create a new aiohttp application instance."""
return create_app()
@pytest.fixture(scope="function")
def mock_users_db_fixture():
"""
Fixture to simulate a user database for dynamic mocking,
reset for each test function.
"""
return {
"admin@example.com": {
"full_name": "Admin User",
"email": "admin@example.com",
"password": "password", # Store plain password for mock authentication
"hashed_password": "hashed_password", # For consistency with real service
"storage_quota_gb": 100,
"storage_used_gb": 10,
"parent_email": None,
"reset_token": None,
"reset_token_expiry": None,
},
"child1@example.com": {
"full_name": "Child User 1",
"email": "child1@example.com",
"password": "password",
"hashed_password": "hashed_password",
"storage_quota_gb": 50,
"storage_used_gb": 5,
"parent_email": "admin@example.com",
"reset_token": None,
"reset_token_expiry": None,
},
}
@pytest.fixture
async def client(
aiohttp_client, mocker: MockerFixture, create_app_instance, mock_users_db_fixture
):
app = create_app_instance # Use the new fixture
# Directly set app["scheduler"] to a mock object
mock_scheduler_instance = mocker.MagicMock()
mock_scheduler_instance.spawn = mocker.AsyncMock()
mock_scheduler_instance.close = mocker.AsyncMock() # Ensure close is awaitable
app["scheduler"] = mock_scheduler_instance
# Create temporary data files for testing
base_path = Path(__file__).parent.parent
data_path = base_path / "data"
data_path.mkdir(exist_ok=True)
users_file = data_path / "users.json"
with open(users_file, "w") as f:
json.dump([], f)
config_file = data_path / "config.json"
with open(config_file, "w") as f:
json.dump({"price_per_gb": 0.0}, f)
app["config_service"] = ConfigService(data_path / "config.json")
client = await aiohttp_client(app)
# Access the real UserService instance and mock its methods
mock_user_service_instance = client.app["user_service"]
# Use the mock_users_db_fixture
mock_users_db = mock_users_db_fixture
def mock_authenticate_user(email, password):
user = mock_users_db.get(email)
if user and user["password"] == password:
return user
return None
def mock_get_user_by_email(email):
return mock_users_db.get(email)
def mock_create_user(full_name, email, password, parent_email=None):
if email in mock_users_db:
raise ValueError("User with this email already exists")
new_user = {
"full_name": full_name,
"email": email,
"password": password,
"hashed_password": "hashed_password",
"storage_quota_gb": 5,
"storage_used_gb": 0,
"parent_email": parent_email,
"reset_token": None,
"reset_token_expiry": None,
}
mock_users_db[email] = new_user
return new_user
def mock_reset_password(email, token, new_password): # Added token argument
user = mock_users_db.get(email)
if user and user.get("reset_token") == token and user.get("reset_token_expiry"):
expiry_time = datetime.datetime.fromisoformat(user["reset_token_expiry"])
if expiry_time > datetime.datetime.now(datetime.timezone.utc):
user["password"] = new_password
user["hashed_password"] = "new_hashed_password" # Simulate hashing
user["reset_token"] = None
user["reset_token_expiry"] = None
return True
return False
def mock_generate_reset_token(email):
user = mock_users_db.get(email)
if user:
# In a real scenario, this would generate a unique token and expiry
user["reset_token"] = "test_token"
user["reset_token_expiry"] = "2030-11-08T20:00:00Z" # A future date
return "test_token"
return None
def mock_validate_reset_token(email, token):
if (
token == "expiredtoken123"
): # Explicitly handle the expired token from the test
return False
user = mock_users_db.get(email)
if user and user.get("reset_token") == token and user.get("reset_token_expiry"):
expiry_time = datetime.datetime.fromisoformat(
user["reset_token_expiry"]
)
if expiry_time > datetime.datetime.now(datetime.timezone.utc):
return True
return False
def mock_save_users():
# This mock ensures that changes to user objects within tests are reflected in mock_users_db
# In a real scenario, this would write to a file or database.
pass # The mock_users_db is already being modified directly by other mocks
def mock_get_all_users():
return list(mock_users_db.values())
def mock_get_users_by_parent_email(parent_email):
return [
user
for user in mock_users_db.values()
if user.get("parent_email") == parent_email
]
def mock_delete_user(email):
if email in mock_users_db:
del mock_users_db[email]
return True
return False
def mock_delete_users_by_parent_email(parent_email):
initial_count = len(mock_users_db)
users_to_delete = [
email
for email, user in mock_users_db.items()
if user.get("parent_email") == parent_email
]
for email in users_to_delete:
del mock_users_db[email]
return initial_count - len(mock_users_db)
mocker.patch.object(
mock_user_service_instance,
"authenticate_user",
side_effect=mock_authenticate_user,
)
mocker.patch.object(
mock_user_service_instance,
"get_user_by_email",
side_effect=mock_get_user_by_email,
)
mocker.patch.object(
mock_user_service_instance, "create_user", side_effect=mock_create_user
)
mocker.patch.object(
mock_user_service_instance, "get_all_users", side_effect=mock_get_all_users
)
mocker.patch.object(
mock_user_service_instance, "update_user_quota", return_value=None
) # Keep as is for now
mocker.patch.object(
mock_user_service_instance, "delete_user", side_effect=mock_delete_user
)
mocker.patch.object(
mock_user_service_instance,
"get_users_by_parent_email",
side_effect=mock_get_users_by_parent_email,
)
mocker.patch.object(
mock_user_service_instance,
"delete_users_by_parent_email",
side_effect=mock_delete_users_by_parent_email,
)
mocker.patch.object(
mock_user_service_instance,
"generate_reset_token",
side_effect=mock_generate_reset_token,
)
mocker.patch.object(
mock_user_service_instance,
"get_user_by_reset_token",
side_effect=lambda token: next(
(
user
for user in mock_users_db.values()
if user.get("reset_token") == token
),
None,
),
)
mocker.patch.object(
mock_user_service_instance, "reset_password", side_effect=mock_reset_password
)
mocker.patch.object(
mock_user_service_instance,
"validate_reset_token",
side_effect=mock_validate_reset_token,
)
mocker.patch.object(
mock_user_service_instance, "_save_users", side_effect=mock_save_users
)
try:
yield client
finally:
# Clean up temporary files
users_file.unlink(missing_ok=True)
config_file.unlink(missing_ok=True) # Use missing_ok for robustness
@pytest.fixture
def mock_send_email(mocker: MockerFixture):
"""
Fixture to mock the send_email function.
This fixture will return the mock that was patched globally by the client fixture.
"""
# Access the globally patched mock
return mocker.patch("retoors.helpers.email_sender.send_email")
Reference in New Issue View Git Blame Copy Permalink