-
Notifications
You must be signed in to change notification settings - Fork 3
/
test_daos_users.py
208 lines (157 loc) · 6.9 KB
/
test_daos_users.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
from __future__ import annotations
import json
import unittest
from collections import namedtuple
from datetime import datetime
from unittest.mock import MagicMock
from unittest.mock import patch
import pytest
from alchemy_mock.mocking import AlchemyMagicMock
from alchemy_mock.mocking import UnifiedAlchemyMagicMock
from fastapi import HTTPException
from freezegun import freeze_time
from sqlalchemy import Select
from werkzeug.security import generate_password_hash
from app.daos.users import create_user
from app.daos.users import get_user
from app.daos.users import list_users
from app.daos.users import login
from app.schemas.users.users_request import CreateUser
from app.schemas.users.users_request import Login
@pytest.fixture
def mock_create_cache():
with patch("app.wrappers.cache_wrappers.CacheUtils.create_cache") as mock_create_cache:
yield mock_create_cache
@pytest.fixture
def mock_retrieve_cache():
with patch("app.wrappers.cache_wrappers.CacheUtils.retrieve_cache") as mock_retrieve_cache:
yield mock_retrieve_cache
@patch("app.wrappers.cache_wrappers.CacheUtils.create_cache")
@pytest.mark.asyncio
@freeze_time(datetime(2024, 3, 15, 17, 20, 37, 495390).strftime("%Y-%m-%d %H:%M:%S.%f"))
async def test_get_user(self, mock_create_cache):
# Mocking cache functions
mock_create_cache.return_value = None
# Mocking the database session
mock_db_session = AlchemyMagicMock()
# Assuming User model is imported from your module
User = namedtuple("User", ["id", "name", "email", "mobile", "created_at", "updated_at"])
user = User(
id=100,
name="John",
email="john@example.com",
mobile=1234567890,
created_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),
updated_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),
)
# Mocking the query method to return the user object
mock_db_session.query.return_value.where.return_value.with_entities.return_value.first.return_value = user
# Call the function you want to test
result = await get_user(100, mock_db_session)
print(result)
print(user._asdict())
# Assert the result
assert result == json.loads(json.dumps(user._asdict(), default=str))
# Mock data
create_user_data = CreateUser(
name="Test User",
email="test@gmail.com",
mobile="1234567890",
password="Test@123",
)
# Mocking the database session
@pytest.fixture
def db_session():
return UnifiedAlchemyMagicMock()
# Test if user is created successfully
def test_create_user(db_session):
response = create_user(create_user_data, db_session)
expected_response = {
"success": True,
"message": "User registered successfully.",
"data": None,
}
assert response == expected_response
class TestListUsers(unittest.TestCase):
@patch("app.models.users.User") # Patch the User model
@patch("app.daos.users.paginate") # Patch the paginate function
def test_list_users_success(self, mock_paginate, mock_user):
# Mocking the Session
mock_session = AlchemyMagicMock()
# Creating mock users
user1 = UnifiedAlchemyMagicMock(id=1, name="User1", email="user1@example.com", mobile="1234567890")
user2 = UnifiedAlchemyMagicMock(id=2, name="User2", email="user2@example.com", mobile="9876543210")
# Mocking the query result
mock_query = Select()
mock_user.query = mock_query
# Mocking the paginate function
mock_paginate.return_value = [user1, user2] # Assuming paginate returns the same users for simplicity
# Call the function
result = list_users(mock_session)
# Assertions
self.assertEqual(result, [user1, user2])
@patch("app.daos.users.paginate")
def test_list_users_exception(self, mock_paginate):
# Mocking the Session
mock_session = UnifiedAlchemyMagicMock()
# Mocking the query result
mock_query = AlchemyMagicMock()
mock_query.all.side_effect = Exception("Test exception")
# Mocking the User model
mock_user = UnifiedAlchemyMagicMock()
mock_user.query = mock_query
# Mocking the paginate function
mock_paginate.side_effect = HTTPException(status_code=400, detail="Test exception")
# Call the function
with self.assertRaises(HTTPException) as cm:
list_users(mock_session)
# Assertions
self.assertEqual(cm.exception.status_code, 400)
class TestGetUser(unittest.IsolatedAsyncioTestCase):
@patch("app.wrappers.cache_wrappers.CacheUtils.retrieve_cache")
async def test_get_user_no_cache_no_user_found(self, mock_retrieve_cache):
# Mocking retrieve_cache to return no cached user
mock_retrieve_cache.return_value = None, None
# Mocking the database session and query result
mock_session = AlchemyMagicMock()
mock_query = MagicMock()
mock_query.where.return_value = mock_query
mock_query.with_entities.return_value = mock_query
mock_query.first.return_value = None # Simulate no user found in the database
mock_session.query.return_value = mock_query
# Call the function and expect an exception
with self.assertRaises(HTTPException) as cm:
await get_user(0, mock_session)
# Assertions
mock_retrieve_cache.assert_called_once_with("user_0")
mock_query.where.assert_called_once()
self.assertEqual(cm.exception.status_code, 400)
self.assertEqual(str(cm.exception.detail), "No User found for given ID.")
# Write similar tests for other scenarios (e.g., cache hit, database query exception, cache creation exception)
login_data = Login(email="test@gmail.com", password="Test@123")
# Test if user login is successful
@patch("app.constants.jwt_utils.create_access_token")
@patch("app.daos.users.check_password_hash")
def test_login_successful(mock_create_access_token, mock_check_password_hash):
mock_create_access_token.return_value = True
mock_check_password_hash.return_value = True
mock_db_session = AlchemyMagicMock()
User = namedtuple("User", ["id", "email", "password"])
user = User(id=1, email="test@gmail.com", password=generate_password_hash("Test@123", method="pbkdf2"))
mock_db_session.query.return_value.where.return_value.first.return_value = user
response = login(login_data, mock_db_session)
expected_response = {
"success": True,
"message": "User logged in successfully.",
"data": {"token": True},
}
assert response == expected_response
def test_login_invalid_password():
mock_db_session = AlchemyMagicMock()
User = namedtuple("User", ["id", "email", "password"])
user = User(id=1, email="test@gmail.com", password="Test@13")
mock_db_session.query.return_value.where.return_value.first.return_value = user
with pytest.raises(Exception):
login(login_data, mock_db_session)
if __name__ == "__main__":
unittest.main()