import pytest from http import HTTPStatus as http_status from server.db.models import ApplicationSettingsHistory from server.db.models.users import Settings from server.db.session import db_session as session_scope from tests.factories import AccountFactory, ApplicationFactory, SettingsFactory, SettingsSchemaFactory, \ ApplicationSettingsHistoryFactory from tests import utils async def set_db(session): session.add_all([ ApplicationFactory.create(slug="app_1", settings=None, types=None), ApplicationFactory.create(slug="app_2", settings={"t_1": {"set_1": "v_1", "set_2": False}}, types=["t_1"]), ApplicationFactory.create( slug="app_3", settings={"t_2": {"s_1": "v_1"}, "t_3": {"s_2": True}}, types=["t_2", "t_3"]) ]) await session.commit() session.add_all([ AccountFactory.create(app_slug="app_1", user_id="u_1"), AccountFactory.create(app_slug="app_1", user_id="u_2"), AccountFactory.create(app_slug="app_1", user_id="u_3"), AccountFactory.create(app_slug="app_3", user_id="u_4") ]) await session.commit() session.add_all([ SettingsFactory.create(account_id=1, type=None, version=3, data={"f_1": "v_1"}), SettingsFactory.create(account_id=2, type=None, version=1, data={"f_2": "v_2"}), SettingsFactory.create(account_id=3, type=None, version=1, data={"f_3": "v_3"}), SettingsFactory.create(account_id=4, type="t_3", version=1, data={"s_2": False}) ]) await session.commit() session.add_all([ SettingsSchemaFactory.create( app_slug="app_3", type="t_2", version=1, data={"type": "object", "properties": {"s_1": {"type": "string"}}, "required": ["s_1"]}, ), SettingsSchemaFactory.create( app_slug="app_3", type="t_3", version=1, data={"type": "object", "properties": {"s_2": {"type": "boolean"}}, "required": ["s_2"]}, ), SettingsSchemaFactory.create( app_slug="app_3", type="t_3", version=2, data={"type": "object", "properties": {"s_2": {"type": "number"}}, "required": ["s_2"]}, ) ]) await session.commit() session.add_all([ ApplicationSettingsHistoryFactory.create( slug="app_3", version=1, settings={"t_2": {"s_1": "v_1"}, "t_3": {"s_2": True}}, types=["t_1", "t_2", "t_3"] ), ApplicationSettingsHistoryFactory.create( slug="app_3", version=2, settings={"t_3": {"s_2": True}}, types=["t_3"] ), ApplicationSettingsHistoryFactory.create( slug="app_2", version=1, settings={"t_1": {"set_1": "v_1", "set_2": False}}, types=["t_1"] ), ApplicationSettingsHistoryFactory.create( slug="app_1", version=1, settings=None, types=None ), ApplicationSettingsHistoryFactory.create( slug="app_1", version=3, settings=None, types=None ), ]) await session.commit() @pytest.mark.parametrize( "headers,status,expected_result,expected_type,is_inherited", ( ({"X-App-Slug": "app_1"}, http_status.BAD_REQUEST, None, None, "False"), ({"X-User-Id": "u_1"}, http_status.BAD_REQUEST, None, None, "False"), ({"X-App-Slug": "no_app", "X-User-Id": "u_1"}, http_status.NOT_FOUND, None, None, "False"), ( {"X-App-Slug": "app_1", "X-User-Id": "u_10", "X-Type": "t_1"}, http_status.BAD_REQUEST, None, None, "False" ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_10", "X-Type": "t_10"}, http_status.BAD_REQUEST, None, None, "False" ), ({"X-App-Slug": "app_3", "X-User-Id": "u_10"}, http_status.BAD_REQUEST, None, None, "False"), ( {"X-App-Slug": "app_1", "X-User-Id": "u_1"}, http_status.OK, {"data": None, "id": 5, "account_id": 1, "default": None}, None, "False" ), ( {"X-App-Slug": "app_1", "X-User-Id": "u_1", "X-Version": "3"}, http_status.OK, {"data": {"f_1": "v_1"}, "id": 1, "account_id": 1, "default": None}, None, "False" ), ( {"X-App-Slug": "app_2", "X-User-Id": "u_1"}, http_status.OK, {"data": {"set_1": "v_1", "set_2": False}, "id": 5, "account_id": 5, "default": {"set_1": "v_1", "set_2": False}}, "t_1", "False" ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_4", "X-Type": "t_3"}, http_status.OK, {"data": {"s_2": False}, "id": 4, "account_id": 4, "default": {"s_2": True}}, "t_2", "False" ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_5", "X-Type": "t_2"}, http_status.OK, {"data": {"s_1": "v_1"}, "id": 5, "account_id": 5, "default": {"s_1": "v_1"}}, "t_2", "False" ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_4", "X-Version": "2", "X-Type": "t_3"}, http_status.OK, { "data": { "categories": { "charts": {"apple": False, "spotify": False}, "starred_playlists": {"apple": False, "spotify": False}, "starred_tracks": {"apple": False, "spotify": False}, }, "s_2": False }, "id": 4, "account_id": 4, "default": {"s_2": True}, }, "t_3", "True", ), ), ) async def test_get(headers, status, expected_result, expected_type, client, auth, db_session, is_inherited): async with session_scope() as session: await set_db(session) headers.update(auth) params = {"is_inherited": is_inherited} response = await client.get("/api/v1/settings/", headers=headers, params=params) assert response.status == status if status == http_status.OK: result = await response.json() assert result == expected_result if expected_result.get("id") == 5: async with session_scope(): assert await Settings.get(id=5, type=expected_type, version=headers.get("X-Version", 1)) @pytest.mark.parametrize( "headers,params,status,expected_result", ( ({}, {}, http_status.BAD_REQUEST, None), ( {"X-App-Slug": "app_1"}, {}, http_status.OK, { "count": 3, "items": [ {"data": {"f_1": "v_1"}, "id": 1, "user_id": "u_1", "account_id": 1}, {"data": {"f_2": "v_2"}, "id": 2, "user_id": "u_2", "account_id": 2}, {"data": {"f_3": "v_3"}, "id": 3, "user_id": "u_3", "account_id": 3} ], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_1"}, {"offset": 1, "limit": 1}, http_status.OK, { "count": 3, "items": [{"data": {"f_2": "v_2"}, "id": 2, "user_id": "u_2", "account_id": 2}], "next": "/user-data-api/v1/settings/bulk/?limit=1&offset=2", "previous": "/user-data-api/v1/settings/bulk/?limit=1&offset=0", } ), ( {"X-App-Slug": "app_1"}, {"user": ["u_1", "u_3"]}, http_status.OK, { "count": 2, "items": [ {"data": {"f_1": "v_1"}, "id": 1, "user_id": "u_1", "account_id": 1}, {"data": {"f_3": "v_3"}, "id": 3, "user_id": "u_3", "account_id": 3} ], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_3"}, {"type": "t_3", "version": 1}, http_status.OK, { "count": 1, "items": [{"data": {"s_2": False}, "id": 4, "user_id": "u_4", "account_id": 4}], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_3"}, {"type": "t_2"}, http_status.OK, { "count": 0, "items": [], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_3"}, {"version": 5}, http_status.OK, { "count": 0, "items": [], "next": None, "previous": None, } ), ({}, {}, http_status.BAD_REQUEST, None), ( {"X-App-Slug": "app_1"}, {}, http_status.OK, { "count": 3, "items": [ {"data": {"f_1": "v_1"}, "id": 1, "user_id": "u_1", "account_id": 1}, {"data": {"f_2": "v_2"}, "id": 2, "user_id": "u_2", "account_id": 2}, {"data": {"f_3": "v_3"}, "id": 3, "user_id": "u_3", "account_id": 3} ], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_1"}, {"offset": 1, "limit": 1}, http_status.OK, { "count": 3, "items": [{"data": {"f_2": "v_2"}, "id": 2, "user_id": "u_2", "account_id": 2}], "next": "/user-data-api/v1/settings/bulk/?limit=1&offset=2", "previous": "/user-data-api/v1/settings/bulk/?limit=1&offset=0", } ), ( {"X-App-Slug": "app_1"}, {"user": ["u_1", "u_3"]}, http_status.OK, { "count": 2, "items": [ {"data": {"f_1": "v_1"}, "id": 1, "user_id": "u_1", "account_id": 1}, {"data": {"f_3": "v_3"}, "id": 3, "user_id": "u_3", "account_id": 3} ], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_3"}, {"type": "t_3", "version": "1"}, http_status.OK, { "count": 1, "items": [{"data": {"s_2": False}, "id": 4, "user_id": "u_4", "account_id": 4}], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_3"}, {"type": "t_2"}, http_status.OK, { "count": 0, "items": [], "next": None, "previous": None, } ), ( {"X-App-Slug": "app_3"}, {"version": "5"}, http_status.OK, { "count": 0, "items": [], "next": None, "previous": None, } ), ), ) async def test_get_all(headers, params, status, expected_result, client, auth, db_session): async with session_scope() as session: await set_db(session) headers.update(auth) response = await client.get("/api/v1/settings/bulk/", headers=headers, params=params) assert response.status == status if status == http_status.OK: result = await response.json() assert result == expected_result @pytest.mark.parametrize( "headers,status,data,expected_type", ( ({"X-App-Slug": "app_1"}, http_status.BAD_REQUEST, None, None), ({"X-User-Id": "u_1"}, http_status.BAD_REQUEST, None, None), ({"X-App-Slug": "no_app", "X-User-Id": "u_1"}, http_status.NOT_FOUND, {"f": "v"}, None), ( {"X-App-Slug": "app_1", "X-User-Id": "u_10", "X-Type": "t_1"}, http_status.BAD_REQUEST, None, None, ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_10", "X-Type": "t_10"}, http_status.BAD_REQUEST, None, None, ), ({"X-App-Slug": "app_3", "X-User-Id": "u_10"}, http_status.BAD_REQUEST, None, None), ({"X-App-Slug": "app_1", "X-User-Id": "u_1"}, http_status.BAD_REQUEST, None, None), ( {"X-App-Slug": "app_3", "X-User-Id": "u_5", "X-Type": "t_2"}, http_status.BAD_REQUEST, {"f": "v"}, None, ), ({"X-App-Slug": "app_1", "X-User-Id": "u_5"}, http_status.CREATED, {"f": "v"}, None), ( {"X-App-Slug": "app_3", "X-User-Id": "u_5", "X-Type": "t_2"}, http_status.CREATED, {"s_1": "v"}, "t_2", ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_5", "X-Type": "t_3", "X-Version": "1"}, http_status.BAD_REQUEST, {"s_2": 123}, "t_3", ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_5", "X-Type": "t_3", "X-Version": "2"}, http_status.CREATED, {"s_2": 123}, "t_3", ), ( {"X-App-Slug": "app_2", "X-User-Id": "u_5"}, http_status.CREATED, {"set_1": "v_1", "set_2": False}, "t_1", ), ), ) async def test_create(headers, status, data, expected_type, client, auth, db_session): async with session_scope() as session: await set_db(session) headers.update(auth) response = await client.post("/api/v1/settings/", headers=headers, json=({"data": data} if data else None)) assert response.status == status if status == http_status.CREATED: result = await response.json() assert result == {"data": data, "id": 5, "account_id": 5} async with session_scope(): assert await Settings.get(id=5, type=expected_type, version=int(headers.get("X-Version", 1))) @pytest.mark.parametrize( "headers,status,data,extra_data,expected_type", ( ({"X-App-Slug": "app_1"}, http_status.BAD_REQUEST, None, None, None), ({"X-User-Id": "u_1"}, http_status.BAD_REQUEST, None, None, None), ({"X-App-Slug": "no_app", "X-User-Id": "u_1"}, http_status.NOT_FOUND, {"f": "v"}, None, None), ( {"X-App-Slug": "app_1", "X-User-Id": "u_10"}, http_status.BAD_REQUEST, None, None, None ), ( {"X-App-Slug": "app_1", "X-User-Id": "u_1", "X-Type": "t_1"}, http_status.BAD_REQUEST, None, None, None ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_1", "X-Type": "t_10"}, http_status.BAD_REQUEST, None, None, None ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_5", "X-Type": "t_2"}, http_status.NOT_FOUND, {"f": "v"}, {"timestamp": 1628065511616, "default": {"markets": ["global"]}}, None, ), ({"X-App-Slug": "app_1", "X-User-Id": "u_1"}, http_status.NOT_FOUND, {"f": "v"}, None, None), ({"X-App-Slug": "app_1", "X-User-Id": "u_2"}, http_status.OK, {"f": "v"}, None, None), ( {"X-App-Slug": "app_3", "X-User-Id": "u_4", "X-Type": "t_2"}, http_status.NOT_FOUND, {"s_1": "v"}, {"timestamp": 1628065511616, "default": {"markets": ["global"]}}, None, ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_4", "X-Type": "t_3", "X-Version": "1"}, http_status.BAD_REQUEST, {"s_2": "a"}, None, None, ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_4", "X-Type": "t_3", "X-Version": "2"}, http_status.NOT_FOUND, {"s_2": 123}, None, None, ), ( {"X-App-Slug": "app_3", "X-User-Id": "u_4", "X-Type": "t_3"}, http_status.OK, {"s_2": True}, None, "t_3", ), ), ) async def test_update(headers, status, data, extra_data, expected_type, client, auth, db_session): async with session_scope() as session: await set_db(session) headers.update(auth) request_data = ({"data": data, **(extra_data or {})} if data else None) response = await client.put("/api/v1/settings/", headers=headers, json=request_data) assert response.status == status if status == http_status.CREATED: result = await response.json() assert result == {"data": data, "id": 5, "account_id": int(data["X-User-Id"].split("_")[1])} async with session_scope(): assert await Settings.get(id=5, type=expected_type, version=headers.get("X-Version", "1")) @pytest.mark.parametrize( "headers,params,status,expected_result", ( ({}, {}, http_status.BAD_REQUEST, None), ({"X-App-Slug": "app_1"}, {"user": {"id": [2, 3]}, "account": {"id": [2, 5]}}, http_status.BAD_REQUEST, None), ({"X-App-Slug": "app_1"}, {"device": {"os": "fridge OS"}}, http_status.BAD_REQUEST, None), ({"X-App-Slug": "app_1"}, {"include": "something"}, http_status.BAD_REQUEST, None), ( {"X-App-Slug": "app_1"}, {}, http_status.OK, { "data": [ { "account_id": 1, "settings": {"data": {"f_1": "v_1"}, "id": 1, "type": None, "version": "3"}, "user_id": "u_1", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"user": {"id": ["u_2", "u_3", "u_4"]}}, http_status.OK, { "data": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, { "account_id": 3, "settings": {"data": {"f_3": "v_3"}, "id": 3, "type": None, "version": "1"}, "user_id": "u_3", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"user": {"id": ["u_2", "u_3", "u_4"]}, "settings": {"type": "portal"}}, http_status.OK, { "data": [ { "account_id": 4, "settings": {"data": {"s_2": "True"}, "id": 6, "type": "portal", "version": "2"}, "user_id": "u_4", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"account": {"id": [2, 5]}}, http_status.OK, { "data": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"device": {"is_active": False}}, http_status.OK, { "data": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"device": {"is_active": False}, "settings": {"type": "portal"}}, http_status.OK, { "data": [ { "account_id": 4, "settings": {"data": {"s_2": "True"}, "id": 6, "type": "portal", "version": "2"}, "user_id": "u_4", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"device": {"os": ["android"]}}, http_status.OK, { "data": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, { "account_id": 3, "settings": {"data": {"f_3": "v_3"}, "id": 3, "type": None, "version": "1"}, "user_id": "u_3", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"settings": {"type": "mobile", "version": "1"}, "include": "devices"}, http_status.OK, { "data": [ { "account_id": 4, "devices": [{"expo_token": "token_9", "id": 9, "is_active": False, "os": "android"}], "settings": {"data": {"s_1": False}, "id": 4, "type": "mobile", "version": "1"}, "user_id": "u_4", }, ], }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "type": "mobile", "version": "1", }, "device": { "app_version": { "min": "1.1.1", "max": "3.3.3", "include_unset": False }, }, "include": "devices", }, http_status.OK, { "data": [] }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "type": "mobile", "version": "1", }, "device": { "app_version": { "min": "1.1.1", "max": "3.3.3", "include_unset": True }, }, "include": "devices", }, http_status.OK, { "data": [ { "account_id": 4, "devices": [{"expo_token": "token_9", "id": 9, "is_active": False, "os": "android"}], "settings": {"data": {"s_1": False}, "id": 4, "type": "mobile", "version": "1"}, "user_id": "u_4", }, ], }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "type": "mobile", "version": "1", }, "device": { "app_version": { "min": "1.1.1", "max": "3.3.3", "eq": "2.2.2", "include_unset": False }, }, "include": "devices", }, http_status.BAD_REQUEST, { "data": [] }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "type": "mobile", "version": "1", "versions": { "min": "1", "max": "2", "eq": "1" }, }, }, http_status.BAD_REQUEST, { "data": [] }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "type": "mobile", "version": "1", "versions": { "min": "1", "max": "2", }, }, }, http_status.OK, { "data": [ { "account_id": 4, "settings": {"data": {"s_1": False}, "id": 4, "type": "mobile", "version": "1"}, "user_id": "u_4", }, ] }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "versions": { "min": "2", }, }, }, http_status.OK, { "data": [] }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "type": "mobile", "version": "1", }, "device": { "app_version": { "min": "1.1.1", "include_unset": True }, }, "include": "devices", }, http_status.OK, { "data": [ { "account_id": 4, "devices": [{"expo_token": "token_9", "id": 9, "is_active": False, "os": "android"}], "settings": {"data": {"s_1": False}, "id": 4, "type": "mobile", "version": "1"}, "user_id": "u_4", }, ], }, ), ( {"X-App-Slug": "app_2"}, { "settings": { "type": "mobile", "versions": { "max": "2", }, }, }, http_status.OK, { "data": [ { "account_id": 4, "settings": {"data": {"s_1": False}, "id": 4, "type": "mobile", "version": "1"}, "user_id": "u_4", }, ] }, ), ), ) async def test_post_service_users_settings_list(headers, params, status, expected_result, client, auth, db_session): async with session_scope() as session: await utils.set_test_db(session, with_favorites=False) headers.update(auth) response = await client.post("/api/service/users/settings/list/", headers=headers, json=params) assert response.status == status if status == http_status.OK: result = await response.json() assert result == expected_result @pytest.mark.parametrize( "slug,version,expected_result", ( ("app_1", 1, {"t_1": {}, "t_2": {"v_1": False}}), ("app_1", 2, {"t_1": {}, "t_2": {"v_1": False, "v_2": False}}), ) ) async def test_application_settings_history_versions(slug, version, expected_result, db_session): async with session_scope() as session: session.add_all([ ApplicationSettingsHistoryFactory.create( slug="app_1", version=1, settings={"t_1": {}, "t_2": {"v_1": False}}, types=["t_1", "t_2"] ), ApplicationSettingsHistoryFactory.create( slug="app_1", version=2, settings={"t_1": {}, "t_2": {"v_1": False, "v_2": False}}, types=["t_1", "t_2"] ) ]) await session.commit() db_data = await ApplicationSettingsHistory.get(slug=slug, version=version) assert expected_result == db_data.settings @pytest.mark.parametrize( "headers,params,status,expected_result", ( ( {"X-App-Slug": "app_2"}, { "device": { "app_version": { "min": "3.3.3", "include_unset": True }, }, }, http_status.OK, { "data": [ {"account_id": 3, "settings": {"data": {"f_3": "v_3"}, "id": 3, "type": None, "version": "1"}, "user_id": "u_3"}] }, ), ( {"X-App-Slug": "app_2"}, { "device": { "app_version": { "min": "2.2.2", "include_unset": True }, }, }, http_status.OK, { "data": [ {"account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2"}, {"account_id": 3, "settings": {"data": {"f_3": "v_3"}, "id": 3, "type": None, "version": "1"}, "user_id": "u_3"}] }, ), ), ) async def test_post_service_users_settings_list_app_version_filtering(headers, params, status, expected_result, client, auth, db_session): async with session_scope() as session: await utils.set_test_db(session, with_favorites=False, with_app_versions=True) headers.update(auth) response = await client.post("/api/service/users/settings/list/", headers=headers, json=params) assert response.status == status if status == http_status.OK: result = await response.json() assert result == expected_result