import os
from typing import TYPE_CHECKING, List
import pytest
from django.core.files import File
from django.urls import reverse
from elasticsearch_dsl.connections import connections
from rest_framework import status
from rest_framework.test import APITestCase
from api.posts.exceptions import (
AlreadyAddedToFavoritesAPIException,
CategoriesDontExistAPIException,
UserNotHasThisPostInFavoritesAPIException,
)
from apps.posts.models import Favorite, Post, View
from common.tests import get_token_for_user
from core.settings import POSTS_CONFIGS
from services.posts import FavoriteService
from tests.posts.factories import (
CommentFactory,
ImageCommentFactory,
LikeFactory,
PostFactory,
PostMediaFactory,
TagFactory,
UserFactory,
fake,
)
from tests.users.factories import CategoryFactory # type: ignore
if TYPE_CHECKING:
from apps.users.models import Category, User
[docs]
class PostAPIViewSetTestCase(APITestCase):
[docs]
@classmethod
def setUpClass(cls):
super().setUpClass()
# Create a user and set its password for authentication
cls.user = UserFactory()
cls.second_user = UserFactory()
cls.categories = CategoryFactory.create_batch(10)
cls.user.set_password("testpassword")
cls.user.save()
# Create a post for this user
cls.post = PostFactory(author=cls.user)
cls.post_without_category = PostFactory(author=cls.user)
cls.post_private = PostFactory(author=cls.user, visibility=Post.VisibilityChoices.PRIVATE)
cls.post.categories.set(cls.categories)
# Create proceed and ready for publish posts
PostFactory.create_batch(3, author=cls.user, status="in_process")
PostFactory.create_batch(3, author=cls.user, status="ready_for_publish")
# def test_create_post(self):
# token = get_token_for_user(self.user)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# category = self.categories[0]
# data = {
# "author": self.user.pk,
# "title": "Test Post",
# "content": "This is a test post.",
# "category": category.pk,
# }
#
# response = self.client.post(reverse("posts:posts"), data)
# post = Post.objects.get(pk=response.data["id"])
# self.assertEqual(post.categories.first(), category)
# self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# self.assertGreaterEqual(Post.objects.count(), 4)
[docs]
def test_retrieve_post(self):
response = self.client.get(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}))
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["title"], self.post.title)
[docs]
def test_partial_update_post(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
data = {"title": "Updated Post", "content": "Updated Post"}
response = self.client.patch(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}), data)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.post.refresh_from_db()
self.assertEqual(self.post.title, data["title"])
self.assertEqual(self.post.content, data["content"])
[docs]
def test_delete_post(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = self.client.delete(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}))
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.post.refresh_from_db()
self.assertTrue(self.post.is_deleted)
[docs]
def test_only_author_can_update_post(self):
# Create another user and set its password for authentication
another_user = UserFactory.create()
token = get_token_for_user(another_user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
data = {"title": "Unauthorized Update Attempt"}
response = self.client.patch(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}), data)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.post.refresh_from_db()
self.assertNotEqual(self.post.title, "Unauthorized Update Attempt")
[docs]
def test_only_author_can_delete_post(self):
# Create another user and set its password for authentication
another_user = UserFactory.create()
token = get_token_for_user(another_user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = self.client.delete(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}))
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.post.refresh_from_db()
self.assertFalse(self.post.is_deleted)
[docs]
def test_post_stays_with_deleted_mark_after_deletion(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = self.client.delete(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}))
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.post.refresh_from_db()
self.assertTrue(self.post.is_deleted)
[docs]
def test_post_view(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
url = reverse("posts:detailed-post", kwargs={"pk": self.post.pk})
response = self.client.get(url)
self.post.refresh_from_db()
views = View.objects.filter(post=self.post)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(views.exists())
self.assertGreaterEqual(views.filter(author=self.user).count(), 1)
self.assertGreaterEqual(views.count(), 1)
self.assertEqual(views.count(), response.data["views_quantity"])
self.assertEqual(response.data["views_quantity"], self.post.views_quantity)
[docs]
def test_double_post_view(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
url = reverse("posts:detailed-post", kwargs={"pk": self.post.pk})
response_before = self.client.get(url)
self.post.refresh_from_db()
quantity_before = self.post.views_quantity
url = reverse("posts:detailed-post", kwargs={"pk": self.post.pk})
response_after = self.client.get(url)
self.post.refresh_from_db()
quantity_after = self.post.views_quantity
views = View.objects.filter(post=self.post)
self.assertEqual(response_after.status_code, status.HTTP_200_OK)
self.assertTrue(views.exists())
self.assertEqual(quantity_before, quantity_after)
self.assertEqual(response_before.data["views_quantity"], response_after.data["views_quantity"])
[docs]
def test_post_multiple_users_view(self):
# First user
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
url = reverse("posts:detailed-post", kwargs={"pk": self.post.pk})
response_first = self.client.get(url)
self.post.refresh_from_db()
quantity_first = self.post.views_quantity
# Second user
token = get_token_for_user(self.second_user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
url = reverse("posts:detailed-post", kwargs={"pk": self.post.pk})
response_second = self.client.get(url)
self.post.refresh_from_db()
quantity_second = self.post.views_quantity
views = View.objects.filter(post=self.post)
self.assertEqual(response_second.status_code, status.HTTP_200_OK)
self.assertTrue(views.exists())
self.assertGreater(quantity_second, quantity_first)
self.assertGreater(response_second.data["views_quantity"], response_first.data["views_quantity"])
[docs]
def test_partial_update_post_without_category(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
data = {
"title": "Updated Post",
"content": "Updated Post",
}
response = self.client.patch(reverse("posts:detailed-post", kwargs={"pk": self.post_without_category.pk}), data)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["detail"], CategoriesDontExistAPIException.default_detail)
[docs]
def test_partial_update_post_new_category(self):
token = get_token_for_user(self.user)
categories = list(self.post.categories.values_list("pk", flat=True))
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
category = self.categories[-1]
data = {"title": "Updated Post", "content": "Updated Post", "category": category.pk}
response = self.client.patch(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}), data)
self.post.refresh_from_db()
new_categories = list(self.post.categories.values_list("pk", flat=True))
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(new_categories, [category.pk])
self.assertNotEqual(new_categories, categories)
[docs]
def test_posts_lists_with_private(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = self.client.get(reverse("posts:posts"))
self.assertEqual(response.data.get("count"), 3)
[docs]
def test_posts_lists_without_private(self):
token = get_token_for_user(self.second_user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = self.client.get(reverse("posts:posts"))
self.assertEqual(response.data.get("count"), 2)
[docs]
@pytest.mark.usefixtures("jpg_file")
class UploadPostApiViewTestCase(APITestCase):
[docs]
def setUp(self) -> None:
self.user = UserFactory()
[docs]
def test_posts_permission_error(self):
self.client.force_authenticate(None)
data = {"content[0]original": self.jpg_file}
response = self.client.post(reverse("posts:upload"), data, format="multipart")
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
[docs]
def test_posts_empty_data(self):
self.client.force_authenticate(self.user)
data = {}
response = self.client.post(reverse("posts:upload"), data)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs]
def test_posts_upload(self):
self.client.force_authenticate(self.user)
data = {"content[0]original": self.jpg_file}
response = self.client.post(reverse("posts:upload"), data, format="multipart")
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
[docs]
class PostLikeAPIViewTestCase(APITestCase):
[docs]
@classmethod
def setUpTestData(cls) -> None:
cls.user = UserFactory() # type: ignore
cls.post = PostFactory() # type: ignore
cls.url = reverse("posts:like-post", kwargs={"pk": cls.post.pk}) # type: ignore
[docs]
def test_resolve_url(self):
self.assertEqual(self.url, f"/api/posts/{self.post.pk}/like/")
response = self.client.post(self.url)
self.assertNotIn(response.status_code, [status.HTTP_404_NOT_FOUND, status.HTTP_405_METHOD_NOT_ALLOWED])
[docs]
def test_unathorize(self):
response = self.client.post(self.url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
self.assertEqual(response.data["detail"], "Authentication credentials were not provided.")
[docs]
def test_already_liked(self):
self.client.force_authenticate(self.user)
LikeFactory(author=self.user, post=self.post)
response = self.client.post(self.url)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["detail"], "You already liked this post")
[docs]
def test_post_not_exists(self):
self.client.force_authenticate(self.user)
url = reverse("posts:like-post", kwargs={"pk": 999999})
response = self.client.post(url)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertEqual(response.data["detail"].code, "not_found")
[docs]
def test_success(self):
self.client.force_authenticate(self.user)
response = self.client.post(self.url)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
[docs]
class PostTagsAPIViewSetTestCase(APITestCase):
[docs]
def setUp(self):
# Create a user and set its password for authentication
self.user = UserFactory()
self.user.set_password("testpassword")
self.user.save()
self.token = get_token_for_user(self.user)
self.category = CategoryFactory()
# Create a post for this user
self.post = PostFactory(author=self.user)
self.tags = TagFactory.create_batch(5)
self.tags_raw = [fake.word() for _ in range(5)]
self.title = fake.catch_phrase()
# def test_create_post_with_tags(self):
# initial_tags_list = [tag.name for tag in self.tags]
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.token}")
#
# tags = [f"#{tag}" for tag in self.tags_raw]
# full_title = f"{self.title} {' '.join(tags)}"
# data = {
# "author": self.user.pk,
# "title": full_title,
# "content": "This is a test post.",
# "category": self.category.pk,
# }
# all_tags = sorted(initial_tags_list + self.tags_raw)
# response = self.client.post(reverse("posts:posts"), data)
# existing_tags = sorted(Tag.objects.values_list("name", flat=True))
# self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# self.assertEqual(Post.objects.count(), 2)
# self.assertListEqual(existing_tags, all_tags)
# self.assertListEqual(sorted(response.data["tags"]), sorted(self.tags_raw))
#
# def test_create_post_with_existing_tags(self):
# initial_tags_list = [tag.name for tag in self.tags]
# quantity_before = len(initial_tags_list)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.token}")
# tags = [f"#{tag}" for tag in initial_tags_list]
# full_title = f"{self.title} {' '.join(tags)}"
# data = {
# "author": self.user.pk,
# "title": full_title,
# "content": "This is a test post.",
# "category": self.category.pk,
# }
# all_tags = sorted(initial_tags_list)
# response = self.client.post(reverse("posts:posts"), data)
# existing_tags = sorted(Tag.objects.values_list("name", flat=True))
# self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# self.assertListEqual(existing_tags, all_tags)
# self.assertEqual(Tag.objects.count(), quantity_before)
# self.assertListEqual(sorted(response.data["tags"]), sorted(existing_tags))
#
# def test_update_post_with_tags(self):
# initial_tags_list = [tag.name for tag in self.tags]
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.token}")
# tags = [f"#{tag}" for tag in self.tags_raw]
# fake.catch_phrase()
# full_title = f"{self.title} {' '.join(tags)}"
# data = {"title": full_title, "content": "This is a test post.", "category": self.category.pk}
# all_tags = sorted(initial_tags_list + self.tags_raw)
# response = self.client.patch(reverse("posts:detailed-post", kwargs={"pk": self.post.pk}), data)
# existing_tags = sorted(Tag.objects.values_list("name", flat=True))
# self.assertEqual(response.status_code, status.HTTP_200_OK)
# self.assertListEqual(existing_tags, all_tags)
# self.assertListEqual(sorted(response.data["tags"]), sorted(self.tags_raw))
[docs]
class PostMediaTestCase(APITestCase):
[docs]
@classmethod
def setUpClass(cls):
super().setUpClass()
# Create a user and set its password for authentication
cls.user = UserFactory()
cls.second_user = UserFactory()
cls.second_user = UserFactory()
cls.user.set_password("testpassword")
cls.user.save()
# Create a post for this user
cls.post = PostFactory(author=cls.user)
cls.category = CategoryFactory()
num_media_items = POSTS_CONFIGS.get("MAX_MEDIA_ITEMS", 10) - 2
for _ in range(num_media_items):
PostMediaFactory(post=cls.post)
#
# def test_successful_creation_post_with_images(self):
# token = get_token_for_user(self.user)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# image_path = os.path.join(os.path.dirname(__file__), "../_data/1.jpg")
# with open(image_path, "rb") as image:
# data = {
# "author": self.user.pk,
# "title": "Test Post",
# "content": "This is a test post.",
# "media": [File(image)],
# "category": self.category.pk,
# }
# response = self.client.post(reverse("posts:posts"), data, format="multipart")
# post = Post.objects.get(pk=response.data["id"])
# media = post.linked_media.first()
# path = media.media.url.replace("/media/", "").split("/")
# self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# self.assertEqual(len(data["media"]), len(response.data["media_list"]))
# self.assertEqual(len(data["media"]), post.linked_media.count())
# self.assertEqual(path[1], str(post.pk))
#
# def test_wrong_creation_post_with_images__more_then_limits(self):
# token = get_token_for_user(self.user)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# image_path = os.path.join(os.path.dirname(__file__), "../_data/1.jpg")
#
# # Create multiple File objects by re-opening the image for each one
# images = [image_path for _ in range(POSTS_CONFIGS.get("MAX_MEDIA_ITEMS", 10) + 1)]
# with ExitStack() as stack:
# data = {
# "author": self.user.pk,
# "title": "Test Post",
# "content": "This is a test post.",
# "media": [File(stack.enter_context(open(file_path, "rb"))) for file_path in images], # noqa SIM115
# "category": self.category.pk,
# }
# response = self.client.post(reverse("posts:posts"), data, format="multipart")
# self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
# self.assertEqual(response.data["detail"], TooManyLinkedMediaAPIException.default_detail)
#
# def test_wrong_update_post_with_images__more_then_limits(self):
# token = get_token_for_user(self.user)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# image_path = os.path.join(os.path.dirname(__file__), "../_data/1.jpg")
#
# # Create multiple File objects by re-opening the image for each one
# images = [image_path for _ in range(POSTS_CONFIGS.get("MAX_MEDIA_ITEMS", 10) + 1)]
# with ExitStack() as stack:
# data = {"media": [File(stack.enter_context(open(file_path, "rb"))) for file_path in images]} # noqa SIM115
# response = self.client.post(
# reverse("posts:add-media", kwargs={"pk": self.post.pk}), data, format="multipart"
# )
# self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
# self.assertEqual(response.data["detail"], TooManyLinkedMediaAPIException.default_detail)
#
# def test_update_post_with_images(self):
# token = get_token_for_user(self.user)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# image_path = os.path.join(os.path.dirname(__file__), "../_data/1.jpg")
#
# # Create multiple File objects by re-opening the image for each one
# images = [image_path for _ in range(2)]
# with ExitStack() as stack:
# data = {"media": [File(stack.enter_context(open(file_path, "rb"))) for file_path in images]} # noqa SIM115
# response = self.client.post(
# reverse("posts:add-media", kwargs={"pk": self.post.pk}), data, format="multipart"
# )
# self.assertEqual(response.status_code, status.HTTP_201_CREATED)
#
# def test_post_media_permissions(self):
# token = get_token_for_user(self.second_user)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# image_path = os.path.join(os.path.dirname(__file__), "../_data/1.jpg")
# with open(image_path, "rb") as image:
# data = {"media": [File(image)]}
# media = self.post.linked_media.first()
# response = self.client.patch(
# reverse("posts:detailed-media", kwargs={"pk": media.pk}), data, format="multipart"
# )
# self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
# def test_post_media_success_permissions(self):
# token = get_token_for_user(self.user)
# self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# image_path = os.path.join(os.path.dirname(__file__), "../_data/1.jpg")
# with open(image_path, "rb") as image:
# data = {"media": [File(image)]}
# media = self.post.linked_media.first()
# response = self.client.patch(
# reverse("posts:detailed-media", kwargs={"pk": media.pk}), data, format="multipart"
# )
# self.assertEqual(response.status_code, status.HTTP_200_OK)
[docs]
class FavoriteViewSetTestCase(APITestCase):
user: "User"
second_user: "User"
categories: List["Category"]
post: Post
post_second: Post
[docs]
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.user = UserFactory()
cls.second_user = UserFactory()
cls.categories = CategoryFactory.create_batch(10)
cls.user.set_password("testpassword")
cls.user.save()
cls.post = PostFactory(author=cls.user)
cls.post_second = PostFactory(author=cls.user)
[docs]
def setUp(self) -> None:
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
self.favorite_url = reverse("posts:post-favorite", kwargs={"pk": self.post.pk})
self.favorites_list_url = reverse("posts:favorites-list")
self.post_url = reverse("posts:detailed-post", kwargs={"pk": self.post.pk})
[docs]
def test_add_to_favorites(self):
response = self.client.post(self.favorite_url)
favorites = Favorite.objects.filter(post=self.post)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertTrue(favorites.exists())
self.assertEqual(favorites.count(), 1)
[docs]
def test_add_already_added_post_to_favorites(self):
FavoriteService.create(self.post, self.user)
response = self.client.post(self.favorite_url)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["detail"], AlreadyAddedToFavoritesAPIException.default_detail)
[docs]
def test_add_to_favorites_not_authenticated(self):
self.client.logout()
response = self.client.post(self.favorite_url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
[docs]
def test_remove_from_favorites(self):
FavoriteService.create(self.post, self.user)
response = self.client.delete(self.favorite_url)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
[docs]
def test_remove_not_added_post_from_favorites(self):
response = self.client.delete(self.favorite_url)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["detail"], UserNotHasThisPostInFavoritesAPIException.default_detail)
[docs]
def test_remove_from_favorites_not_authenticated(self):
self.client.logout()
response = self.client.delete(self.favorite_url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
[docs]
def test_list_favorites(self):
FavoriteService.create(self.post, self.user)
FavoriteService.create(self.post_second, self.user)
response = self.client.get(self.favorites_list_url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data["results"]), 2)
[docs]
def test_add_to_favorites_with_field_check(self):
self.client.post(self.favorite_url)
Favorite.objects.filter(post=self.post)
post_response = self.client.get(self.post_url)
self.assertTrue(post_response.data["is_favorite"])
self.assertEqual(post_response.data["favorites_quantity"], 1)
[docs]
def test_remove_from_favorites_with_field_check(self):
FavoriteService.create(self.post, self.user)
self.client.delete(self.favorite_url)
post_response = self.client.get(self.post_url)
self.assertFalse(post_response.data["is_favorite"])
self.assertEqual(post_response.data["favorites_quantity"], 0)
[docs]
class PostsSearchSortingAPITestCase(APITestCase):
[docs]
def setUp(self):
self.tag = TagFactory(name="testtag")
self.category = CategoryFactory(name="TestCategory")
self.post = PostFactory(
title="Test title", content="Content for testing", views_quantity=1000, likes_quantity=200
)
self.post.tags.add(self.tag)
self.post.categories.add(self.category)
self.user = UserFactory()
self.user.interests.add(self.category)
self.post2 = PostFactory(views_quantity=0, likes_quantity=500)
self.post3 = PostFactory(views_quantity=1000, likes_quantity=1000)
self.post2.categories.add(self.category)
self.post3.categories.add(self.category)
self.es = connections.get_connection()
[docs]
def tearDown(self):
self.es.delete_by_query(index="posts", body={"query": {"match_all": {}}})
self.es.indices.refresh(index="posts")
[docs]
def test_search_functionality_full_name(self):
self.es.indices.refresh(index="posts")
response = self.client.get(reverse("posts:posts"), {"search": "Test title TestCategory testtag"}, format="json")
self.assertEqual(response.status_code, 200)
self.assertGreaterEqual(response.data["count"], 1)
self.assertEqual(response.data["results"][0]["title"], self.post.title)
[docs]
def test_search_functionality_title_category(self):
self.es.indices.refresh(index="posts")
response = self.client.get(reverse("posts:posts"), {"search": "Test title TestCategory"}, format="json")
self.assertEqual(response.status_code, 200)
self.assertGreaterEqual(response.data["count"], 1)
self.assertEqual(response.data["results"][0]["title"], self.post.title)
[docs]
def test_search_functionality_title_tag(self):
self.es.indices.refresh(index="posts")
response = self.client.get(reverse("posts:posts"), {"search": "Test title"}, format="json")
self.assertEqual(response.status_code, 200)
self.assertGreaterEqual(response.data["count"], 1)
self.assertEqual(response.data["results"][0]["title"], self.post.title)
[docs]
def test_search_functionality_title(self):
self.es.indices.refresh(index="posts")
response = self.client.get(reverse("posts:posts"), {"search": "Test title"}, format="json")
self.assertEqual(response.status_code, 200)
self.assertGreaterEqual(response.data["count"], 1)
self.assertEqual(response.data["results"][0]["title"], self.post.title)
[docs]
def test_sorting(self):
# Create additional posts with different numbers of likes and views
response = self.client.get(reverse("posts:posts"))
# Check if the posts are sorted according to their coefficient
self.assertEqual(response.data["results"][0]["id"], self.post3.id)
self.assertEqual(response.data["results"][1]["id"], self.post2.id)
self.assertEqual(response.data["results"][2]["id"], self.post.id)
[docs]
def test_get_personal_queryset(self):
token = get_token_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = self.client.get(reverse("posts:posts"), format="json")
self.assertEqual(response.status_code, status.HTTP_200_OK)
results = response.data["results"]
self.assertEqual(results[0]["id"], self.post3.id)
self.assertEqual(results[1]["id"], self.post2.id)
self.assertEqual(results[2]["id"], self.post.id)
[docs]
class PostFilterTestCase(APITestCase):
[docs]
@classmethod
def setUpTestData(cls):
# Creating some tags and categories
cls.tag1 = TagFactory(name="Tag1")
cls.tag2 = TagFactory(name="Tag2")
cls.category1 = CategoryFactory(name="Category1")
cls.category2 = CategoryFactory(name="Category2")
# Creating two users
cls.user1 = UserFactory()
cls.user2 = UserFactory()
# Creating posts with different tags, categories, and authors
cls.post1 = PostFactory(author=cls.user1)
cls.post2 = PostFactory(author=cls.user2)
cls.post1.categories.set([cls.category1])
cls.post1.tags.set([cls.tag1])
cls.post2.categories.set([cls.category2])
cls.post2.tags.set([cls.tag2])
[docs]
def test_filter_by_tag(self):
url = reverse("posts:posts")
response = self.client.get(url, {"tags": self.tag1.name})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data["results"]), 1) # Only one post should be returned
self.assertEqual(
response.data["results"][0]["id"], self.post1.id
) # The returned post should have the id of post1
[docs]
def test_filter_by_category(self):
url = reverse("posts:posts")
response = self.client.get(url, {"categories": self.category1.pk})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data["results"]), 1)
self.assertEqual(response.data["results"][0]["id"], self.post1.id)
[docs]
class TestWebhookPostMediaAPIView(APITestCase):
"""Test WebhookPostMediaAPIView"""
[docs]
def setUp(self) -> None:
self.post = PostFactory()
self.post_media = PostMediaFactory(post=self.post)
self.data = {
"post": self.post.id,
"post_media": self.post_media.id,
"upload_video_path": "media/post-content/%s/video/formated/stream.m3u8",
"upload_preview_path": "media/post-content/%s/video/preview.jpg",
}
self.url = reverse("posts:webhook_post")
[docs]
def test_url(self):
url = "/api/posts/webhook/"
self.assertEqual(self.url, url)
response = self.client.post(self.url)
self.assertNotIn(response.status_code, [status.HTTP_404_NOT_FOUND, status.HTTP_405_METHOD_NOT_ALLOWED])
[docs]
def test_empty_request(self):
response = self.client.post(self.url)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs]
def test_success(self):
response = self.client.post(self.url, data=self.data)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)