import pytest from http import HTTPStatus as http_status from server.db.models import Event from server.db.session import db_session as session_scope from tests.api.test_favorites import get_dsp from tests.api.test_messages import check_bulk_create_response from tests.factories import AccountFactory, EventFactory from tests.utils import check_object, create_applications, get_country_code, create_accounts def create_event_data(i, **kwargs) -> dict: data = { "ttl": i * 10, "code": f"{i}-code", "event_id": None, "external_id": None, "meta": { "dsp": get_dsp(i), "country_code": get_country_code(i), }, "data": { "i": i, "track": { "id": f"id_{i}", "isrc": f"isrc_{i}", "name": f"track_name_{i}", "artists": [{"id": f"artist_id_{i}", "name": f"artist_name_{i}"}], "image_url": f"https://image-{i}" }, } } for k, v in kwargs.items(): data[k] = v return data @pytest.mark.parametrize( "data,headers,post_status,get_status", ( ({}, {}, http_status.BAD_REQUEST, http_status.BAD_REQUEST), ( { "public": False, "code": "test_event", "ttl": 3600, }, {"X-App-Slug": "app7"}, http_status.NOT_FOUND, http_status.NOT_FOUND ), ( { "public": False, "code": "test_event", "ttl": 3600, }, {"X-App-Slug": "app1"}, http_status.CREATED, http_status.OK ), ( { "public": False, "code": "test_event", "ttl": 3600, "meta": { "domain": "event", "function": "test" }, "data": { "domain": "event", "types": ["test", "test"], "data": { "data": {"data": "test"} } } }, {"X-App-Slug": "app1"}, http_status.CREATED, http_status.OK ), ) ) async def test_one_event_cases(data, headers, post_status, get_status, db_session, client, auth): async with session_scope() as session: await create_applications(session, 2) headers.update(auth) get_id, unset, app_slug = 1, {}, headers.get("X-App-Slug") post_response = await client.post("/api/service/events/", headers=headers, json=data) assert post_response.status == post_status if post_status == http_status.CREATED: result = await post_response.json() assert "id" in result assert result["app"] == headers["X-App-Slug"] get_id = result["id"] default = {"public": False, "data": None, "meta": None, "event_id": None, "external_id": None} unset = {k: default[k] for k in (default.keys() - data.keys())} async with session_scope(): event = await Event.get(id=get_id) check_object(event, data, app_slug=app_slug, **unset) get_response = await client.get( "/api/service/events/", headers=headers, params=dict(id=get_id, include="all") ) assert get_response.status == get_status if get_status == http_status.OK: result = await get_response.json() data.update(unset) data.update({"id": get_id, "context": None}) assert result.pop("created_at", None) is not None assert result.pop("updated_at", None) is not None assert result == data @pytest.mark.parametrize( "data,ok_indexes,failed_indexes,headers,post_status,get_status", ( ({}, None, None, {}, http_status.BAD_REQUEST, http_status.BAD_REQUEST), ( { "public": False, "data": [ create_event_data(1) ], }, None, None, {"X-App-Slug": "app7"}, http_status.NOT_FOUND, http_status.OK ), ( { "public": False, "data": [create_event_data(i, ttl=-100) for i in range(1, 3)] }, None, list(range(2)), {"X-App-Slug": "app1"}, http_status.OK, http_status.OK ), ( { "public": False, "data": [create_event_data(i) for i in range(1, 4)] }, list(range(3)), None, {"X-App-Slug": "app1"}, http_status.OK, http_status.OK ), ( { "public": False, "data": [ create_event_data(1, data=None, external_id="1_ext_id"), create_event_data(2, code=None) ] }, (0,), (1,), {"X-App-Slug": "app1"}, http_status.OK, http_status.OK ), ) ) async def test_multiple_events_cases( data, ok_indexes, failed_indexes, headers, post_status, get_status, db_session, client, auth): async with session_scope() as session: await create_applications(session, 2) headers.update(auth) app_slug = headers.get("X-App-Slug") ok_input_items = [item for i, item in enumerate(data.get("data", [])) if ok_indexes and i in ok_indexes] default = { "public": data.get("public", True), "data": None, "meta": None, "event_id": None, "external_id": None, } id_list = [] post_response = await client.post( "/api/v2/service/events/", headers=headers, json=data) assert post_response.status == post_status if post_status == http_status.OK: result = await post_response.json() check_bulk_create_response(result, ok_indexes=ok_indexes, failed_indexes=failed_indexes) async with session_scope(): events, _ = await Event.list(order_by=Event.id.asc()) assert len(events) == len(ok_indexes or []) for i, event in enumerate(events): input_item = ok_input_items[i] unset = {k: default[k] for k in (default.keys() - input_item.keys())} unset["app_slug"] = app_slug check_object(event, input_item, **unset) id_list.append(event.id) list_data = { "id": id_list or [1, 2], "include": "all" } get_response = await client.post( "/api/service/events/list/", headers=headers, json=list_data) assert get_response.status == get_status if get_status == http_status.OK: result = await get_response.json() list_response_items = result.get("data", []) assert len(list_response_items) == len(id_list) for i, response_item in enumerate(list_response_items): input_data = ok_input_items[i] input_item = {k: default[k] for k in (default.keys() - input_data.keys())} input_item.update(input_data) input_item.update({"id": id_list[i], "context": None}) assert response_item.pop("created_at") assert response_item.pop("updated_at") assert response_item == input_item @pytest.mark.parametrize( "data,ok_indexes,failed_indexes,headers,status", ( ( { "public": False, "unique_mode": "exclude", "data": [ create_event_data(1) ], }, None, None, {"X-App-Slug": "app1"}, http_status.BAD_REQUEST, ), ( { "public": False, "unique_key": "event_id", "data": [ create_event_data(1) ], }, None, None, {"X-App-Slug": "app1"}, http_status.BAD_REQUEST, ), ( { "public": False, "unique_mode": "raise", "unique_key": "external_id", "data": [create_event_data(i, external_id="ext1") for i in range(1, 3)] }, None, None, {"X-App-Slug": "app1"}, http_status.BAD_REQUEST, ), ( { "public": False, "unique_mode": "exclude", "unique_key": "external_id", "data": [create_event_data(i, external_id=f"ext{i}") for i in range(1, 4)] }, (2,), (0, 1), {"X-App-Slug": "app1"}, http_status.OK, ), ( { "public": False, "unique_mode": "exclude", "unique_key": "code", "data": [ create_event_data(1, code="new_code_1"), create_event_data(2, code="new_code_1"), create_event_data(3, code="new_code_2"), ] }, (0, 2), (1,), {"X-App-Slug": "app1"}, http_status.OK, ), ) ) async def test_uniqueness_check( data, ok_indexes, failed_indexes, headers, status, db_session, client, auth ): existing_events_data = [ create_event_data(1, code="event_1_code", external_id="ext1", app_slug="app1"), create_event_data(2, code="event_1_code", external_id="ext2", app_slug="app1"), create_event_data(3, code="event_2_code", external_id="ext2", app_slug="app1") ] async with session_scope() as session: await create_applications(session, 2) for item in existing_events_data: event = EventFactory.create(**item) session.add(event) await session.commit() headers.update(auth) post_response = await client.post( "/api/v2/service/events/", headers=headers, json=data) assert post_response.status == status if status == http_status.OK: result = await post_response.json() check_bulk_create_response(result, ok_indexes=ok_indexes, failed_indexes=failed_indexes) async with session_scope(): events_count = await Event.count() assert events_count == (len(existing_events_data) + len(ok_indexes or []))