import pytest from copy import deepcopy import random from unittest.mock import patch from server.core.helpers import authorization from tests.constants import ( EXPECTED_USER_SEARCH_NAME_CONSTRAINT, EMPTY_PARAMS, MISSING_DATA_FORE_REQUIRED_FIELD, PROFILE_INFO, PROFILE_INFO_2, PROFILE_INFO_FROM_ATLAS, PROFILE_INFO_FROM_ATLAS_2, PROFILE_LABELS_45, PROFILE_LABELS_45_EMPTY, PROFILE_LABELS_45_NAME, PROFILE_LABELS_45_USERS, PROFILE_LABELS_WITH_USERS_LABEL_NAMES, PROFILE_LABELS_WITH_USERS_RESPONSE, PROFILE_LABELS_WITH_USERS, PROFILE_RESPONSE_FROM_ATLAS, PROFILE_RESPONSE_FROM_ATLAS_2, REQUIRED_REQUEST_FIELDS, REQUIRED_REQUEST_FIELDS_2, SUCCESS, TEST_HEADERS_FOR_ATLAS, TEST_X_USER_ID_2, TEST_EMPTY_X_USER_ID, UNAUTHORIZED, USER_SEARCH_ID, UNKNOWN_FIELD, ) class TestUserSearch: @pytest.mark.parametrize( "url, expected_return_data, http_method, request_body", [ ("/api/dna/user_searches", UNAUTHORIZED, "get", {}), ("/api/dna/user_searches/123", UNAUTHORIZED, "get", {}), ("/api/dna/user_searches", UNAUTHORIZED, "post", deepcopy(REQUIRED_REQUEST_FIELDS)), ("/api/dna/user_searches/123", UNAUTHORIZED, "put", deepcopy(REQUIRED_REQUEST_FIELDS)), ("/api/dna/user_searches/123/archived", UNAUTHORIZED, "patch", {"archived": True}), ("/api/dna/user_searches/123", UNAUTHORIZED, "delete", {}), ], ) async def test_user_search_endpoints_unauthorized( self, url, expected_return_data, http_method, request_body, postgres_client, app_client, auth_header ): request = getattr(app_client, http_method) response = await request(url, headers=auth_header, json=request_body) assert expected_return_data == await response.json() @pytest.mark.parametrize( "request_body", [ (deepcopy(REQUIRED_REQUEST_FIELDS)), ], ) async def test_get_user_search_item(self, request_body, postgres_client, app_client, auth_header_authorized): new_user_search = await app_client.post( "/api/dna/user_searches", headers=auth_header_authorized, json=request_body ) new_user_search = await new_user_search.json() response = await app_client.get( f"/api/dna/user_searches/{new_user_search['id']}", headers=auth_header_authorized ) res = await response.json() issubset = request_body.items() <= res.items() assert issubset is True @pytest.mark.parametrize( "expected_return_data", [ ({}), ], ) async def test_get_user_search_item_empty_dict( self, expected_return_data, postgres_client, app_client, auth_header_authorized ): response = await app_client.get(f"/api/dna/user_searches/{USER_SEARCH_ID}", headers=auth_header_authorized) res = await response.json() assert expected_return_data == res @pytest.mark.parametrize( "expected_qty,request_body", [ (2, deepcopy(REQUIRED_REQUEST_FIELDS)), ], ) async def test_get_user_search_list( self, expected_qty, request_body, postgres_client, app_client, auth_header_authorized ): await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) request_body = request_body.copy() request_body["name"] = "testname" await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await app_client.get("/api/dna/user_searches", headers=auth_header_authorized) res = await response.json() assert expected_qty == len(res) @pytest.mark.parametrize( "request_body,expected_status_code", [ (deepcopy(REQUIRED_REQUEST_FIELDS), 201), ], ) async def test_save_user_search_return_201( self, request_body, expected_status_code, postgres_client, app_client, auth_header_authorized ): response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) assert expected_status_code == response.status @pytest.mark.parametrize( "request_body,expected_status_code, expectected_result", [ (deepcopy(REQUIRED_REQUEST_FIELDS), 400, EXPECTED_USER_SEARCH_NAME_CONSTRAINT), ], ) async def test_save_user_search_check_unique_constraint_name( self, request_body, expected_status_code, expectected_result, postgres_client, app_client, auth_header_authorized, ): await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) assert expected_status_code == response.status assert expectected_result == await response.json() @pytest.mark.parametrize( "request_body,expected_status_code, expectected_result", [ (deepcopy(REQUIRED_REQUEST_FIELDS), 422, EMPTY_PARAMS), ], ) async def test_save_user_search_empty_dcit( self, request_body, expected_status_code, expectected_result, postgres_client, app_client, auth_header_authorized, ): request_body = request_body.copy() request_body["params"] = {} response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) assert expected_status_code == response.status assert expectected_result == await response.json() @pytest.mark.parametrize( "request_body, request_body_2, expected_status_code, expectected_result", [ ( deepcopy(REQUIRED_REQUEST_FIELDS), deepcopy(REQUIRED_REQUEST_FIELDS_2), 400, EXPECTED_USER_SEARCH_NAME_CONSTRAINT, ), ], ) async def test_save_user_search_check_unique_constraint_name_case_sensitive( self, request_body, request_body_2, expected_status_code, expectected_result, postgres_client, app_client, auth_header_authorized, ): await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) assert expected_status_code == response.status assert expectected_result == await response.json() @pytest.mark.parametrize( "request_body, bad_request_status_code, expectected_result, created_status_code", [ (deepcopy(REQUIRED_REQUEST_FIELDS), 400, EXPECTED_USER_SEARCH_NAME_CONSTRAINT, 201), ], ) async def test_save_user_search_check_unique_constraint_for_different_users( self, request_body, bad_request_status_code, expectected_result, created_status_code, postgres_client, app_client, auth_header_authorized, ): await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) assert bad_request_status_code == response.status assert expectected_result == await response.json() auth_header_authorized = auth_header_authorized.copy() auth_header_authorized["X-User-Id"] = "test_user_id" response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) assert created_status_code == response.status response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) assert bad_request_status_code == response.status assert expectected_result == await response.json() @pytest.mark.parametrize( "request_body, expected_status", [ (deepcopy(REQUIRED_REQUEST_FIELDS), 200), ], ) async def test_update_user_search( self, request_body, expected_status, postgres_client, app_client, auth_header_authorized ): response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await response.json() user_search_id = response["id"] request_body["name"] = "testtest" response = await app_client.put( f"/api/dna/user_searches/{user_search_id}", headers=auth_header_authorized, json=request_body ) res = await response.json() assert user_search_id == res["id"] assert expected_status == response.status response = await app_client.get(f"/api/dna/user_searches/{user_search_id}", headers=auth_header_authorized) res = await response.json() issubset = request_body.items() <= res.items() assert issubset is True @pytest.mark.parametrize( "request_body, request_body_2, expected_status, error", [ ( deepcopy(REQUIRED_REQUEST_FIELDS), deepcopy(REQUIRED_REQUEST_FIELDS_2), 400, EXPECTED_USER_SEARCH_NAME_CONSTRAINT, ), ], ) async def test_update_user_search_same_name( self, request_body, request_body_2, expected_status, error, postgres_client, app_client, auth_header_authorized ): await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) request_body_2["name"] = "test_1" response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body_2) response = await response.json() user_search_id_2 = response["id"] request_body_2["name"] = "TEST" response = await app_client.put( f"/api/dna/user_searches/{user_search_id_2}", headers=auth_header_authorized, json=request_body ) assert expected_status == response.status res = await response.json() assert res == error @pytest.mark.parametrize( "request_body, url, http_method", [ (deepcopy(REQUIRED_REQUEST_FIELDS), "/api/dna/user_searches/1", "put"), (deepcopy(REQUIRED_REQUEST_FIELDS), "/api/dna/user_searches", "post"), ], ) async def test_user_search_missing_body_params( self, request_body, url, http_method, postgres_client, app_client, auth_header_authorized ): request_body = request_body.copy() expected_return_data = {} for _ in range(random.randint(1, random.randint(1, len(request_body.keys())))): item = random.choice(list(request_body.keys())) request_body.pop(item) expected_return_data[item] = MISSING_DATA_FORE_REQUIRED_FIELD request = getattr(app_client, http_method) response = await request(url, headers=auth_header_authorized, json=request_body) assert expected_return_data == await response.json() @pytest.mark.parametrize( "request_body, archived_request_body, expected_status_code", [ (deepcopy(REQUIRED_REQUEST_FIELDS), {"archived": True}, 200), ], ) async def test_update_user_search_archived( self, request_body, archived_request_body, expected_status_code, postgres_client, app_client, auth_header_authorized, ): response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await response.json() user_search_id = response["id"] response = await app_client.patch( f"/api/dna/user_searches/{user_search_id}/archived", headers=auth_header_authorized, json=archived_request_body, ) assert expected_status_code == response.status response = await app_client.get(f"/api/dna/user_searches/{user_search_id}", headers=auth_header_authorized) assert response.status == 200 res = await response.json() assert res["archived"] is not None @pytest.mark.parametrize( "request_body, expected_response", [ (deepcopy(REQUIRED_REQUEST_FIELDS), {"archived": MISSING_DATA_FORE_REQUIRED_FIELD}), ], ) async def test_update_user_search_archived_unknown_fields( self, request_body, expected_response, postgres_client, app_client, auth_header_authorized ): for key in request_body: expected_response[key] = UNKNOWN_FIELD response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await response.json() user_search_id = response["id"] response = await app_client.patch( f"/api/dna/user_searches/{user_search_id}/archived", headers=auth_header_authorized, json=request_body ) res = await response.json() assert expected_response == res @pytest.mark.parametrize( "request_body, expected_delete_response, expected_empty_item", [ (deepcopy(REQUIRED_REQUEST_FIELDS), {"message": SUCCESS}, {}), ], ) async def test_delete_user_search( self, request_body, expected_delete_response, expected_empty_item, postgres_client, app_client, auth_header_authorized, ): response = await app_client.post("/api/dna/user_searches", headers=auth_header_authorized, json=request_body) response = await response.json() user_search_id = response["id"] response = await app_client.delete(f"/api/dna/user_searches/{user_search_id}", headers=auth_header_authorized) res = await response.json() assert expected_delete_response == res response = await app_client.get(f"/api/dna/user_searches/{user_search_id}", headers=auth_header_authorized) res = await response.json() assert expected_empty_item == res @pytest.mark.parametrize( "url, expected_return_data", [ ("/api/dna/profile/info", UNAUTHORIZED), ("/api/dna/profile/labels", UNAUTHORIZED), ("/api/dna/profile/labels/45", UNAUTHORIZED), ], ) async def test_get_profile_endpoints_unauthorized(self, url, expected_return_data, app_client, auth_header): response = await app_client.get(url, headers=auth_header) assert expected_return_data == await response.json() @pytest.mark.parametrize( "expected_return_data", [ (PROFILE_INFO), ], ) @patch.object( authorization, "make_request", side_effect=[TEST_HEADERS_FOR_ATLAS, PROFILE_INFO_FROM_ATLAS, PROFILE_RESPONSE_FROM_ATLAS], ) async def test_get_profile_info(self, make_request_mock, expected_return_data, app_client, auth_header_authorized): response = await app_client.get("/api/dna/profile/info", headers=auth_header_authorized) assert expected_return_data == await response.json() @pytest.mark.parametrize( "expected_return_data_for_first_user, expected_return_data_for_second_user", [ (PROFILE_INFO, PROFILE_INFO_2), ], ) @patch.object( authorization, "make_request", side_effect=[ TEST_HEADERS_FOR_ATLAS, PROFILE_INFO_FROM_ATLAS, PROFILE_RESPONSE_FROM_ATLAS, PROFILE_INFO_FROM_ATLAS_2, PROFILE_RESPONSE_FROM_ATLAS_2, ], ) async def test_get_profile_for_specific_user_id( self, make_request_mock, expected_return_data_for_first_user, expected_return_data_for_second_user, app_client, auth_header_authorized, ): response = await app_client.get("/api/dna/profile/info", headers=auth_header_authorized) assert expected_return_data_for_first_user == await response.json() auth_header_authorized["X-User-Id"] = TEST_X_USER_ID_2 response = await app_client.get("/api/dna/profile/info", headers=auth_header_authorized) assert expected_return_data_for_second_user == await response.json() @pytest.mark.parametrize( "expected_return_data", [ (PROFILE_LABELS_45), ], ) @patch.object( authorization, "make_request", side_effect=[TEST_HEADERS_FOR_ATLAS, PROFILE_LABELS_45_NAME, PROFILE_LABELS_45_USERS], ) async def test_get_label_with_profiles( self, make_request_mock, expected_return_data, app_client, auth_header_authorized ): response = await app_client.get("/api/dna/profile/labels/45", headers=auth_header_authorized) assert expected_return_data == await response.json() @pytest.mark.parametrize( "expected_return_data", [ ({}), ], ) @patch.object(authorization, "make_request", side_effect=[TEST_HEADERS_FOR_ATLAS, PROFILE_LABELS_45_EMPTY]) async def test_get_label_with_profiles_no_label( self, make_request_mock, expected_return_data, app_client, auth_header_authorized ): auth_header_authorized["X-User-Id"] = TEST_EMPTY_X_USER_ID response = await app_client.get("/api/dna/profile/labels/45", headers=auth_header_authorized) assert expected_return_data == await response.json() @pytest.mark.parametrize( "expected_return_data", [ (PROFILE_LABELS_WITH_USERS_RESPONSE), ], ) @patch.object( authorization, "make_request", side_effect=[ TEST_HEADERS_FOR_ATLAS, PROFILE_LABELS_WITH_USERS_LABEL_NAMES, PROFILE_LABELS_WITH_USERS, PROFILE_LABELS_WITH_USERS, ], ) async def test_get_list_labels_with_profiles( self, make_request_mock, expected_return_data, app_client, auth_header_authorized ): response = await app_client.get("/api/dna/profile/labels", headers=auth_header_authorized) assert expected_return_data == await response.json() @pytest.mark.parametrize( "expected_return_data", [ ([]), ], ) @patch.object(authorization, "make_request", side_effect=[TEST_HEADERS_FOR_ATLAS, PROFILE_LABELS_45_EMPTY]) async def test_get_list_labels_with_profiles_empty_list( self, make_request_mock, expected_return_data, app_client, auth_header_authorized ): response = await app_client.get("/api/dna/profile/labels", headers=auth_header_authorized) assert expected_return_data == await response.json()