"""Push notifications tests.""" import datetime import json import urllib.parse as urlparse from urllib.parse import parse_qs import pytest from flask import url_for from apollo_main_db.push_notifications.models import PushMessage, TopicEnum, UserDeviceToken, UserNotificationSetting from core import http_status from push_notifications.tests.factories import PushMessageFactory, SettingFactory, UserDeviceFactory, UserSettingFactory @pytest.mark.parametrize( "status_code,token,is_new", ((http_status.HTTP_400_BAD_REQUEST, None, False), (http_status.HTTP_201_CREATED, "qwerty", True)), ) def test_register_device(db_session, user_id, client, status_code, token, is_new, patch_auth_user): patch_auth_user(user_id) url = url_for("push_notifications_blueprint.register_device") response = client.post(url, data=json.dumps(dict(token=token)), content_type="application/json") assert response.status_code == status_code if is_new: assert db_session.query(UserDeviceToken).filter_by(user_id=user_id, token=token, is_active=True).all() def test_unregister_device(db_session, user_id, client, patch_auth_user): user_device_1, user_device_2 = UserDeviceFactory.create_batch(size=2, user_id=user_id, is_active=True) patch_auth_user(user_id) url = url_for("push_notifications_blueprint.unregister_device") response = client.post(url, data=json.dumps(dict(token=user_device_1.token)), content_type="application/json") assert response.status_code == http_status.HTTP_200_OK inactive_device = db_session.query(UserDeviceToken).filter_by(user_id=user_id, token=user_device_1.token).first() active_device = db_session.query(UserDeviceToken).filter_by(user_id=user_id, token=user_device_2.token).first() assert not inactive_device.is_active assert active_device.is_active def test_get_push_messages_list(db_session, patch_datetime_now, client, patch_auth_user): user_devices = UserDeviceFactory.create_batch(2) first_user_id = user_devices[0] patch_auth_user(first_user_id.user_id) PushMessageFactory.create(user_id=user_devices[0].user_id, created_at=datetime.datetime(2020, 2, 9, 10, 0, 0)) PushMessageFactory.create(user_id=user_devices[0].user_id, created_at=datetime.datetime(2020, 2, 10, 0, 0, 0)) PushMessageFactory.create_batch(2, user_id=user_devices[1].user_id) assert len(db_session.query(PushMessage).all()) == 4 url = url_for("push_notifications_blueprint.push_message_all") response = client.get(url) data = response.json assert data["count"] == 2 assert len(data["items"]) == 2 assert data["items"][0]["time_delta"] == 36000 assert data["items"][1]["time_delta"] == 86400 def test_get_push_message_detail(db_session, client, patch_auth_user): user_device = UserDeviceFactory.create() patch_auth_user(user_device.user_id) push, _ = PushMessageFactory.create_batch(2, user_id=user_device.user_id, topic=TopicEnum.PLAYLIST_ADDITIONS) url = url_for("push_notifications_blueprint.push_message", push_id=push.id) response = client.get(url) assert response.json["id"] == str(push.id) assert response.json["track_id"] == push.track_id assert response.json["topic"] == TopicEnum.PLAYLIST_ADDITIONS.name.lower() assert response.json["position"] assert response.json["change"] assert response.json["track_name"] assert response.json["artist_name"] assert response.json["isrc"] assert "country_code" not in response.json assert "target" in response.json def test_filtering_push_messages_list(db_session, client, patch_auth_user): user_device = UserDeviceFactory.create() patch_auth_user(user_device.user_id) PushMessageFactory.create_batch(size=3, user_id=user_device.user_id, is_new=True) PushMessageFactory.create(user_id=user_device.user_id, is_new=False) url = url_for("push_notifications_blueprint.push_message_all", is_new=True) response = client.get(url) assert response.json["count"] == 3 for r in response.json["items"]: assert r["is_new"] is True url = url_for("push_notifications_blueprint.push_message_all", is_new=False) response = client.get(url) assert response.json["count"] == 1 assert response.json["items"][0]["is_new"] is False # query params will be ignored url = url_for("push_notifications_blueprint.push_message_all", foo="bar") response = client.get(url) assert response.json["count"] == 4 # testing pagination next and previous urls url = url_for("push_notifications_blueprint.push_message_all", is_new=True, limit=1, offset=1) response = client.get(url) assert response.json["count"] == 3 next_url = response.json["next"] parsed_next_url = urlparse.urlparse(next_url) assert parse_qs(parsed_next_url.query) == {"is_new": ["True"], "limit": ["1"], "offset": ["2"]} previous_url = response.json["previous"] parsed_previous_url = urlparse.urlparse(previous_url) assert parse_qs(parsed_previous_url.query) == {"is_new": ["True"]} @pytest.mark.parametrize( "status_code,push_id,count", ( (http_status.HTTP_200_OK, None, 1), (http_status.HTTP_200_OK, "850e69a8-4f9f-4926-8415-1dbdcffbf719", 0), (http_status.HTTP_400_BAD_REQUEST, "foo", 0), ), ) def test_mark_push_messages_as_read(db_session, client, patch_auth_user, status_code, push_id, count): user_device = UserDeviceFactory.create() patch_auth_user(user_device.user_id) push = PushMessageFactory.create(user_id=user_device.user_id, is_new=True) push_id = push_id or str(push.id) url = url_for("push_notifications_blueprint.push_messages_bulk") payload = dict(data=[{"id": push_id, "is_new": False}]) response = client.put(url, data=json.dumps(payload), content_type="application/json") assert response.status_code == status_code assert db_session.query(PushMessage).filter_by(is_new=False).count() == count def test_get_notification_settings(db_session, client, patch_auth_user, user_id): patch_auth_user(user_id) setting = SettingFactory.create() SettingFactory.create_batch(size=5) UserSettingFactory.create(user_id=user_id, setting_id=setting.id) url = url_for("push_notifications_blueprint.settings") response = client.get(url) for item in response.json["items"]: assert item["id"] assert item["code"] assert item["enabled"] if item["id"] == setting.id else item["enabled"] is False def test_post_notification_settings(db_session, client, patch_auth_user, user_id): patch_auth_user(user_id) setting_1 = SettingFactory.create(parent_id=None) setting_2 = SettingFactory.create(parent_id=None) setting_3 = SettingFactory.create(parent_id=setting_2.id) UserSettingFactory.create(user_id=user_id, setting_id=setting_1.id) url = url_for("push_notifications_blueprint.settings") # already saved setting response = client.post(url, data=json.dumps({"id": setting_1.id}), content_type="application/json") assert response.status_code == http_status.HTTP_400_BAD_REQUEST # not existed setting response = client.post(url, data=json.dumps({"id": 100500}), content_type="application/json") assert response.status_code == http_status.HTTP_400_BAD_REQUEST # not saved parent setting response = client.post(url, data=json.dumps({"id": setting_3.id}), content_type="application/json") assert response.status_code == http_status.HTTP_400_BAD_REQUEST response = client.post(url, data=json.dumps({"id": setting_2.id}), content_type="application/json") assert response.status_code == http_status.HTTP_201_CREATED response = client.post(url, data=json.dumps({"id": setting_3.id}), content_type="application/json") assert response.status_code == http_status.HTTP_201_CREATED setting_ids = [i for i, in db_session.query(UserNotificationSetting.setting_id).filter_by(user_id=user_id)] assert {setting_1.id, setting_2.id, setting_3.id} == set(setting_ids) def test_delete_notification_settings(db_session, client, patch_auth_user, user_id): patch_auth_user(user_id) setting_1, setting_2 = SettingFactory.create_batch(size=2) UserSettingFactory.create(user_id=user_id, setting_id=setting_1.id) url = url_for("push_notifications_blueprint.settings") response = client.delete(url, data=json.dumps({"id": setting_2.id}), content_type="application/json") assert response.status_code == http_status.HTTP_400_BAD_REQUEST response = client.post(url, data=json.dumps({"id": 100500}), content_type="application/json") assert response.status_code == http_status.HTTP_400_BAD_REQUEST response = client.delete(url, data=json.dumps({"id": setting_1.id}), content_type="application/json") assert response.status_code == http_status.HTTP_204_NO_CONTENT settings = db_session.query(UserNotificationSetting.setting_id).filter_by(user_id=user_id).all() assert not len(settings)