-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add parametrized tests to check database queries
- Loading branch information
Showing
1 changed file
with
66 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,66 @@ | ||
from tests.fixtures.insert_data_in_db import * | ||
from tests.fixtures.fetch_data_from_db import * | ||
from tests.tests_data.code import code_data | ||
from tests.tests_data.user import user_data | ||
from tests.tests_data.role import role_data | ||
from tests.tests_data.permission import permission_data | ||
from tests.tests_data.session import session_data | ||
from tests.tests_data.device import device_data | ||
from tests.tests_data.file_info import file_info_data | ||
from tests.tests_data.file_format import file_format_data | ||
from tests.tests_data.column_stats import column_stats_data | ||
from tests.tests_data.task import task_data | ||
from tests.tests_data.feedback import feedback_data | ||
|
||
|
||
@pytest.mark.parametrize( | ||
"fixture_name, table_name, data", | ||
[ | ||
("insert_user_data", "user", user_data), | ||
("insert_permission_data", "permission", permission_data), | ||
("insert_device_data", "device", device_data), | ||
("insert_code_data", "code", code_data), | ||
("insert_role_data", "role", role_data), | ||
("insert_session_data", "session", session_data), | ||
("insert_feedback_data", "feedback", feedback_data), | ||
("insert_file_info_data", "file_info", file_info_data), | ||
("insert_file_format_data", "file_format", file_format_data), | ||
("insert_column_stats_data", "column_stats", column_stats_data), | ||
("insert_task_data", "task", task_data), | ||
], | ||
) | ||
def test_database_queries( | ||
fixture_name: str, | ||
table_name: str, | ||
data: list[dict], | ||
fetch_all_data_from_table, | ||
request, | ||
): | ||
insert_data = request.getfixturevalue(fixture_name) | ||
for i in range(len(data)): | ||
insert_data(**data[i]) | ||
|
||
fetched_data = fetch_all_data_from_table(table_name) | ||
assert len(data) == len(fetched_data) | ||
|
||
str_data = [{str(k): str(v) for k, v in entry.items()} for entry in data] | ||
|
||
for d in str_data: | ||
exists_in_fetched = False | ||
for fetched_obj in fetched_data: | ||
fetched_dict = {str(k): str(getattr(fetched_obj, k)) for k in d} | ||
if fetched_dict == d: | ||
exists_in_fetched = True | ||
break | ||
assert exists_in_fetched, f"Data {d} is not found in fetched data" | ||
|
||
for fetched_obj in fetched_data: | ||
exists_in_data = False | ||
for d in str_data: | ||
fetched_dict = {str(k): str(getattr(fetched_obj, k)) for k in d} | ||
if fetched_dict == d: | ||
exists_in_data = True | ||
break | ||
assert ( | ||
exists_in_data | ||
), f"Fetched data {fetched_obj} is not found in the original data" |