mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-19 17:47:40 +00:00
347 lines
16 KiB
Python
347 lines
16 KiB
Python
from datetime import timedelta
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
from rest_framework import status
|
|
from rest_framework.test import APIClient
|
|
|
|
from apps.epg.models import EPGData, EPGSource, ProgramData
|
|
from apps.channels.models import Channel
|
|
|
|
User = get_user_model()
|
|
|
|
SEARCH_URL = "/api/epg/programs/search/"
|
|
|
|
|
|
class ProgramSearchAPIViewTests(TestCase):
|
|
"""Tests for the /api/epg/programs/search/ endpoint."""
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.epg_source = EPGSource.objects.create(name="Test Source", source_type="xmltv")
|
|
cls.epg = EPGData.objects.create(
|
|
tvg_id="test-tvg", name="Test EPG", epg_source=cls.epg_source
|
|
)
|
|
|
|
now = timezone.now().replace(microsecond=0)
|
|
|
|
# Premier League Football — airing now
|
|
cls.prog_football = ProgramData.objects.create(
|
|
epg=cls.epg,
|
|
title="Premier League Football",
|
|
description="Live coverage of the Premier League match.",
|
|
start_time=now - timedelta(minutes=30),
|
|
end_time=now + timedelta(hours=1),
|
|
)
|
|
|
|
# Newcastle vs Villa — also airing now
|
|
cls.prog_newcastle = ProgramData.objects.create(
|
|
epg=cls.epg,
|
|
title="Newcastle vs Villa",
|
|
description="Match highlights.",
|
|
start_time=now - timedelta(minutes=15),
|
|
end_time=now + timedelta(hours=2),
|
|
)
|
|
|
|
# BBC News — starts in 3 hours
|
|
cls.prog_news = ProgramData.objects.create(
|
|
epg=cls.epg,
|
|
title="BBC News at Ten",
|
|
description="The latest news from around the world.",
|
|
start_time=now + timedelta(hours=3),
|
|
end_time=now + timedelta(hours=4),
|
|
)
|
|
|
|
# Nature Documentary — starts in 5 hours
|
|
cls.prog_doc = ProgramData.objects.create(
|
|
epg=cls.epg,
|
|
title="Nature Documentary",
|
|
description="Exploring wildlife in the Amazon.",
|
|
start_time=now + timedelta(hours=5),
|
|
end_time=now + timedelta(hours=6),
|
|
)
|
|
|
|
cls.now = now
|
|
cls.user = User.objects.create_user(username="testuser", password="pass", user_level=1)
|
|
# Standard users only see programs for EPG entries linked to accessible channels.
|
|
Channel.objects.create(
|
|
name="Test Channel",
|
|
channel_number=1,
|
|
epg_data=cls.epg,
|
|
user_level=1,
|
|
)
|
|
|
|
def setUp(self):
|
|
self.client = APIClient(REMOTE_ADDR="127.0.0.1")
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Response structure
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_response_structure(self):
|
|
"""Response includes pagination envelope and all expected program fields."""
|
|
response = self.client.get(SEARCH_URL, {"page_size": 1})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
data = response.json()
|
|
self.assertIn("count", data)
|
|
self.assertIn("results", data)
|
|
self.assertIn("next", data)
|
|
self.assertIn("previous", data)
|
|
|
|
program = data["results"][0]
|
|
for field in ("id", "title", "start_time", "end_time", "tvg_id", "channels", "streams"):
|
|
self.assertIn(field, program)
|
|
|
|
# ------------------------------------------------------------------
|
|
# No filter — returns all programs
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_no_filters_returns_all(self):
|
|
"""Omitting filters returns all seeded programs."""
|
|
response = self.client.get(SEARCH_URL)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.json()["count"], 4)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Title search
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_title_simple_match(self):
|
|
"""Simple title search returns matching programs."""
|
|
response = self.client.get(SEARCH_URL, {"title": "football"})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
data = response.json()
|
|
self.assertEqual(data["count"], 1)
|
|
self.assertEqual(data["results"][0]["title"], "Premier League Football")
|
|
|
|
def test_title_multi_word_is_phrase_not_implicit_and(self):
|
|
"""Space-separated words without AND/OR are matched as a phrase, not as implicit AND.
|
|
|
|
'Premier Football' contains both words from 'Premier League Football' but
|
|
not as a consecutive phrase — so it should return 0 results.
|
|
Use 'Premier AND Football' to match both words independently.
|
|
"""
|
|
phrase = self.client.get(SEARCH_URL, {"title": "Premier Football"}).json()
|
|
explicit_and = self.client.get(SEARCH_URL, {"title": "Premier AND Football"}).json()
|
|
# Phrase match: "Premier Football" is not a substring of "Premier League Football"
|
|
self.assertEqual(phrase["count"], 0)
|
|
# Explicit AND: both words present → matches
|
|
self.assertEqual(explicit_and["count"], 1)
|
|
|
|
def test_title_quoted_phrase(self):
|
|
"""Double-quoted phrases are matched literally; 'and'/'or' inside quotes are not operators.
|
|
|
|
This is the standard way to search for program titles that contain conjunctions,
|
|
e.g. "Law and Order". Without quotes, lowercase 'and'/'or' are still treated as
|
|
case-insensitive boolean operators — so quoting is the reliable way to do a phrase match.
|
|
"""
|
|
prog = ProgramData.objects.create(
|
|
epg=self.epg,
|
|
title="Law and Order",
|
|
description="Crime drama.",
|
|
start_time=self.now + timedelta(hours=10),
|
|
end_time=self.now + timedelta(hours=11),
|
|
)
|
|
try:
|
|
# Quoted phrase → exact substring match → finds the program
|
|
quoted = self.client.get(SEARCH_URL, {"title": '"Law and Order"'}).json()
|
|
self.assertEqual(quoted["count"], 1)
|
|
self.assertEqual(quoted["results"][0]["title"], "Law and Order")
|
|
|
|
# Quoted phrase that is not a substring → no match
|
|
non_phrase = self.client.get(SEARCH_URL, {"title": '"Law Order"'}).json()
|
|
self.assertEqual(non_phrase["count"], 0)
|
|
|
|
# Mix: quoted phrase AND bare term present in the title → matches
|
|
mixed_match = self.client.get(SEARCH_URL, {"title": '"Law and Order" AND order'}).json()
|
|
self.assertEqual(mixed_match["count"], 1)
|
|
|
|
# Mix: quoted phrase AND bare term NOT in the title → no match
|
|
mixed_no_match = self.client.get(SEARCH_URL, {"title": '"Law and Order" AND crime'}).json()
|
|
self.assertEqual(mixed_no_match["count"], 0)
|
|
finally:
|
|
prog.delete()
|
|
def test_title_case_insensitive(self):
|
|
"""Title search is case-insensitive."""
|
|
lower = self.client.get(SEARCH_URL, {"title": "football"}).json()
|
|
upper = self.client.get(SEARCH_URL, {"title": "FOOTBALL"}).json()
|
|
self.assertEqual(lower["count"], 1)
|
|
self.assertEqual(upper["count"], 1)
|
|
self.assertEqual(lower["results"][0]["title"], upper["results"][0]["title"])
|
|
|
|
def test_title_and_operator(self):
|
|
"""AND operator (case-insensitive) requires both terms to be present in the title."""
|
|
upper = self.client.get(SEARCH_URL, {"title": "Premier AND League"}).json()
|
|
lower = self.client.get(SEARCH_URL, {"title": "Premier and League"}).json()
|
|
self.assertEqual(upper["count"], 1)
|
|
self.assertEqual(lower["count"], 1)
|
|
self.assertIn("Premier", upper["results"][0]["title"])
|
|
|
|
def test_title_or_operator(self):
|
|
"""OR operator (case-insensitive) returns programs matching either term."""
|
|
upper = self.client.get(SEARCH_URL, {"title": "Newcastle OR Football"})
|
|
lower = self.client.get(SEARCH_URL, {"title": "Newcastle or Football"})
|
|
self.assertEqual(upper.status_code, status.HTTP_200_OK)
|
|
for response in (upper, lower):
|
|
titles = [r["title"] for r in response.json()["results"]]
|
|
self.assertIn("Premier League Football", titles)
|
|
self.assertIn("Newcastle vs Villa", titles)
|
|
|
|
def test_title_no_match_returns_empty(self):
|
|
"""Search with no matching title returns empty results."""
|
|
response = self.client.get(SEARCH_URL, {"title": "XYZNONEXISTENT999"})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
data = response.json()
|
|
self.assertEqual(data["count"], 0)
|
|
self.assertEqual(data["results"], [])
|
|
|
|
def test_title_whole_word_matching(self):
|
|
"""title_whole_words=true matches complete words but not partial words."""
|
|
# 'new' as substring matches 'Newcastle vs Villa' and 'BBC News at Ten'
|
|
partial = self.client.get(SEARCH_URL, {"title": "new"}).json()
|
|
# Whole-word \bnew\b matches neither 'Newcastle' nor 'News' (partial matches)
|
|
whole_no_match = self.client.get(SEARCH_URL, {"title": "new", "title_whole_words": "true"}).json()
|
|
self.assertEqual(partial["count"], 2)
|
|
self.assertEqual(whole_no_match["count"], 0)
|
|
|
|
# 'football' is a complete word in 'Premier League Football' — must still match
|
|
whole_match = self.client.get(SEARCH_URL, {"title": "football", "title_whole_words": "true"}).json()
|
|
self.assertEqual(whole_match["count"], 1)
|
|
self.assertEqual(whole_match["results"][0]["title"], "Premier League Football")
|
|
|
|
# 'league' is also a complete word — and 'Premier AND league' with whole_words works
|
|
both_words = self.client.get(SEARCH_URL, {"title": "Premier AND league", "title_whole_words": "true"}).json()
|
|
self.assertEqual(both_words["count"], 1)
|
|
|
|
def test_title_regex(self):
|
|
"""title_regex=true applies the query as a regex pattern."""
|
|
response = self.client.get(
|
|
SEARCH_URL, {"title": "^Premier", "title_regex": "true"}
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
for program in response.json()["results"]:
|
|
self.assertTrue(program["title"].startswith("Premier"))
|
|
|
|
def test_title_parenthetical_grouping(self):
|
|
"""Parenthetical groups with AND/OR are evaluated correctly."""
|
|
# (Newcastle OR Football) AND (Villa OR League) should match both seeded programs:
|
|
# "Premier League Football" matches Football AND League
|
|
# "Newcastle vs Villa" matches Newcastle AND Villa
|
|
response = self.client.get(
|
|
SEARCH_URL, {"title": "(Newcastle OR Football) AND (Villa OR League)"}
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
titles = {r["title"] for r in response.json()["results"]}
|
|
self.assertIn("Premier League Football", titles)
|
|
self.assertIn("Newcastle vs Villa", titles)
|
|
self.assertNotIn("BBC News at Ten", titles)
|
|
self.assertNotIn("Nature Documentary", titles)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Description search
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_description_simple_match(self):
|
|
"""Description search returns programs whose description contains the term."""
|
|
response = self.client.get(SEARCH_URL, {"description": "Premier League"})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
data = response.json()
|
|
self.assertEqual(data["count"], 1)
|
|
self.assertEqual(data["results"][0]["title"], "Premier League Football")
|
|
|
|
def test_description_and_operator(self):
|
|
"""AND operator in description requires both terms."""
|
|
response = self.client.get(SEARCH_URL, {"description": "latest AND news"})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
data = response.json()
|
|
self.assertEqual(data["count"], 1)
|
|
self.assertEqual(data["results"][0]["title"], "BBC News at Ten")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Time filters
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_airing_at_returns_current_programs(self):
|
|
"""airing_at returns programs where start_time <= t < end_time."""
|
|
ts = self.now.isoformat()
|
|
response = self.client.get(SEARCH_URL, {"airing_at": ts})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
titles = [r["title"] for r in response.json()["results"]]
|
|
self.assertIn("Premier League Football", titles)
|
|
self.assertIn("Newcastle vs Villa", titles)
|
|
self.assertNotIn("BBC News at Ten", titles)
|
|
|
|
def test_start_after_filter(self):
|
|
"""start_after excludes programs that start before the given time."""
|
|
cutoff = (self.now + timedelta(hours=4)).isoformat()
|
|
response = self.client.get(SEARCH_URL, {"start_after": cutoff})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
titles = [r["title"] for r in response.json()["results"]]
|
|
self.assertIn("Nature Documentary", titles)
|
|
self.assertNotIn("Premier League Football", titles)
|
|
self.assertNotIn("BBC News at Ten", titles)
|
|
|
|
def test_start_before_filter(self):
|
|
"""start_before excludes programs that start after the given time."""
|
|
cutoff = (self.now + timedelta(hours=1)).isoformat()
|
|
response = self.client.get(SEARCH_URL, {"start_before": cutoff})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
titles = [r["title"] for r in response.json()["results"]]
|
|
self.assertIn("Premier League Football", titles)
|
|
self.assertIn("Newcastle vs Villa", titles)
|
|
self.assertNotIn("Nature Documentary", titles)
|
|
|
|
def test_invalid_datetime_returns_400(self):
|
|
"""An unparseable datetime value returns a 400 error."""
|
|
response = self.client.get(SEARCH_URL, {"airing_at": "not-a-date"})
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn("error", response.json())
|
|
|
|
# ------------------------------------------------------------------
|
|
# Field selection
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_field_selection_limits_response_keys(self):
|
|
"""fields param restricts the keys present in each result."""
|
|
response = self.client.get(SEARCH_URL, {"fields": "title,start_time,end_time"})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
for program in response.json()["results"]:
|
|
self.assertIn("title", program)
|
|
self.assertIn("start_time", program)
|
|
self.assertIn("end_time", program)
|
|
self.assertNotIn("description", program)
|
|
self.assertNotIn("channels", program)
|
|
self.assertNotIn("streams", program)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Pagination
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_pagination_page_size(self):
|
|
"""page_size limits the number of results returned."""
|
|
response = self.client.get(SEARCH_URL, {"page_size": 2})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
data = response.json()
|
|
self.assertEqual(len(data["results"]), 2)
|
|
self.assertEqual(data["count"], 4)
|
|
self.assertIsNotNone(data["next"])
|
|
|
|
def test_pagination_second_page(self):
|
|
"""Page 2 returns different results from page 1."""
|
|
page1 = self.client.get(SEARCH_URL, {"page": 1, "page_size": 2}).json()
|
|
page2 = self.client.get(SEARCH_URL, {"page": 2, "page_size": 2}).json()
|
|
ids_p1 = {r["id"] for r in page1["results"]}
|
|
ids_p2 = {r["id"] for r in page2["results"]}
|
|
self.assertTrue(ids_p1.isdisjoint(ids_p2))
|
|
|
|
def test_page_size_capped_at_maximum(self):
|
|
"""page_size beyond the 500 maximum is clamped, not rejected."""
|
|
response = self.client.get(SEARCH_URL, {"page_size": 10000})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
data = response.json()
|
|
# All 4 seeded programs are returned — request was accepted and clamped
|
|
self.assertEqual(data["count"], 4)
|
|
self.assertEqual(len(data["results"]), 4)
|