chore: rename libretimeapi dir to libretime_api
This commit is contained in:
jo
Kyle Robbertze
parent
02efadc3d0
commit
6de242db65
36 changed files with 0 additions and 0 deletions
0
api/libretime_api/tests/__init__.py
Normal file
0
api/libretime_api/tests/__init__.py
Normal file
0
api/libretime_api/tests/models/__init__.py
Normal file
0
api/libretime_api/tests/models/__init__.py
Normal file
57
api/libretime_api/tests/models/test_schedule.py
Normal file
57
api/libretime_api/tests/models/test_schedule.py
Normal file
|
|
@ -0,0 +1,57 @@
|
|||
from datetime import datetime, timedelta
|
||||
|
||||
from django.test import TestCase
|
||||
|
||||
from libretimeapi.models import Schedule, ShowInstance
|
||||
|
||||
|
||||
class TestSchedule(TestCase):
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
cls.show_instance = ShowInstance(
|
||||
created=datetime(year=2021, month=10, day=1, hour=12),
|
||||
starts=datetime(year=2021, month=10, day=2, hour=1),
|
||||
ends=datetime(year=2021, month=10, day=2, hour=2),
|
||||
)
|
||||
cls.length = timedelta(minutes=10)
|
||||
cls.cue_in = timedelta(seconds=1)
|
||||
cls.cue_out = cls.length - timedelta(seconds=4)
|
||||
|
||||
def create_schedule(self, starts):
|
||||
return Schedule(
|
||||
starts=starts,
|
||||
ends=starts + self.length,
|
||||
cue_in=self.cue_in,
|
||||
cue_out=self.cue_out,
|
||||
instance=self.show_instance,
|
||||
)
|
||||
|
||||
def test_get_cueout(self):
|
||||
# No overlapping schedule datetimes, normal usecase:
|
||||
s1_starts = datetime(year=2021, month=10, day=2, hour=1, minute=30)
|
||||
s1 = self.create_schedule(s1_starts)
|
||||
self.assertEqual(s1.get_cueout(), self.cue_out)
|
||||
self.assertEqual(s1.get_ends(), s1_starts + self.length)
|
||||
|
||||
# Mixed overlapping schedule datetimes (only ends is overlapping):
|
||||
s2_starts = datetime(year=2021, month=10, day=2, hour=1, minute=55)
|
||||
s2 = self.create_schedule(s2_starts)
|
||||
self.assertEqual(s2.get_cueout(), timedelta(minutes=5))
|
||||
self.assertEqual(s2.get_ends(), self.show_instance.ends)
|
||||
|
||||
# Fully overlapping schedule datetimes (starts and ends are overlapping):
|
||||
s3_starts = datetime(year=2021, month=10, day=2, hour=2, minute=1)
|
||||
s3 = self.create_schedule(s3_starts)
|
||||
self.assertEqual(s3.get_cueout(), self.cue_out)
|
||||
self.assertEqual(s3.get_ends(), self.show_instance.ends)
|
||||
|
||||
def test_is_valid(self):
|
||||
# Starts before the schedule ends
|
||||
s1_starts = datetime(year=2021, month=10, day=2, hour=1, minute=30)
|
||||
s1 = self.create_schedule(s1_starts)
|
||||
self.assertTrue(s1.is_valid)
|
||||
|
||||
# Starts after the schedule ends
|
||||
s2_starts = datetime(year=2021, month=10, day=2, hour=3)
|
||||
s2 = self.create_schedule(s2_starts)
|
||||
self.assertFalse(s2.is_valid)
|
||||
BIN
api/libretime_api/tests/resources/song.mp3
Normal file
BIN
api/libretime_api/tests/resources/song.mp3
Normal file
Binary file not shown.
23
api/libretime_api/tests/runners.py
Normal file
23
api/libretime_api/tests/runners.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
from django.test.runner import DiscoverRunner
|
||||
|
||||
|
||||
class ManagedModelTestRunner(DiscoverRunner):
|
||||
"""
|
||||
Test runner that automatically makes all unmanaged models in your Django
|
||||
project managed for the duration of the test run, so that one doesn't need
|
||||
to execute the SQL manually to create them.
|
||||
"""
|
||||
|
||||
def setup_test_environment(self, *args, **kwargs):
|
||||
from django.apps import apps
|
||||
|
||||
self.unmanaged_models = [m for m in apps.get_models() if not m._meta.managed]
|
||||
for m in self.unmanaged_models:
|
||||
m._meta.managed = True
|
||||
super(ManagedModelTestRunner, self).setup_test_environment(*args, **kwargs)
|
||||
|
||||
def teardown_test_environment(self, *args, **kwargs):
|
||||
super(ManagedModelTestRunner, self).teardown_test_environment(*args, **kwargs)
|
||||
# reset unmanaged models
|
||||
for m in self.unmanaged_models:
|
||||
m._meta.managed = False
|
||||
48
api/libretime_api/tests/test_models.py
Normal file
48
api/libretime_api/tests/test_models.py
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
from django.apps import apps
|
||||
from django.contrib.auth.models import Group
|
||||
from rest_framework.test import APITestCase
|
||||
|
||||
from libretimeapi.models import User
|
||||
from libretimeapi.models.user_constants import DJ, GUEST
|
||||
from libretimeapi.permission_constants import GROUPS
|
||||
|
||||
|
||||
class TestUserManager(APITestCase):
|
||||
def test_create_user(self):
|
||||
user = User.objects.create_user(
|
||||
"test",
|
||||
email="test@example.com",
|
||||
password="test",
|
||||
type=DJ,
|
||||
first_name="test",
|
||||
last_name="user",
|
||||
)
|
||||
db_user = User.objects.get(pk=user.pk)
|
||||
self.assertEqual(db_user.username, user.username)
|
||||
|
||||
def test_create_superuser(self):
|
||||
user = User.objects.create_superuser(
|
||||
"test",
|
||||
email="test@example.com",
|
||||
password="test",
|
||||
first_name="test",
|
||||
last_name="user",
|
||||
)
|
||||
db_user = User.objects.get(pk=user.pk)
|
||||
self.assertEqual(db_user.username, user.username)
|
||||
|
||||
|
||||
class TestUser(APITestCase):
|
||||
def test_guest_get_group_perms(self):
|
||||
user = User.objects.create_user(
|
||||
"test",
|
||||
email="test@example.com",
|
||||
password="test",
|
||||
type=GUEST,
|
||||
first_name="test",
|
||||
last_name="user",
|
||||
)
|
||||
permissions = user.get_group_permissions()
|
||||
# APIRoot permission hardcoded in the check as it isn't a Permission object
|
||||
str_perms = [p.codename for p in permissions] + ["view_apiroot"]
|
||||
self.assertCountEqual(str_perms, GROUPS[GUEST])
|
||||
133
api/libretime_api/tests/test_permissions.py
Normal file
133
api/libretime_api/tests/test_permissions.py
Normal file
|
|
@ -0,0 +1,133 @@
|
|||
import os
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import AnonymousUser
|
||||
from model_bakery import baker
|
||||
from rest_framework.test import APIRequestFactory, APITestCase
|
||||
|
||||
from libretimeapi.models.user_constants import ADMIN, DJ, GUEST, PROGRAM_MANAGER
|
||||
from libretimeapi.permission_constants import (
|
||||
DJ_PERMISSIONS,
|
||||
GUEST_PERMISSIONS,
|
||||
PROGRAM_MANAGER_PERMISSIONS,
|
||||
)
|
||||
from libretimeapi.permissions import IsSystemTokenOrUser
|
||||
|
||||
|
||||
class TestIsSystemTokenOrUser(APITestCase):
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
cls.path = "/api/v2/files/"
|
||||
|
||||
def test_unauthorized(self):
|
||||
response = self.client.get(self.path.format("files"))
|
||||
self.assertEqual(response.status_code, 403)
|
||||
|
||||
def test_token_incorrect(self):
|
||||
token = "doesnotexist"
|
||||
request = APIRequestFactory().get(self.path)
|
||||
request.user = AnonymousUser()
|
||||
request.META["Authorization"] = "Api-Key {token}".format(token=token)
|
||||
allowed = IsSystemTokenOrUser().has_permission(request, None)
|
||||
self.assertFalse(allowed)
|
||||
|
||||
def test_token_correct(self):
|
||||
token = settings.CONFIG.get("general", "api_key")
|
||||
request = APIRequestFactory().get(self.path)
|
||||
request.user = AnonymousUser()
|
||||
request.META["Authorization"] = "Api-Key {token}".format(token=token)
|
||||
allowed = IsSystemTokenOrUser().has_permission(request, None)
|
||||
self.assertTrue(allowed)
|
||||
|
||||
|
||||
class TestPermissions(APITestCase):
|
||||
URLS = [
|
||||
"schedule",
|
||||
"shows",
|
||||
"show-days",
|
||||
"show-hosts",
|
||||
"show-instances",
|
||||
"show-rebroadcasts",
|
||||
"files",
|
||||
"playlists",
|
||||
"playlist-contents",
|
||||
"smart-blocks",
|
||||
"smart-block-contents",
|
||||
"smart-block-criteria",
|
||||
"webstreams",
|
||||
]
|
||||
|
||||
def logged_in_test_model(self, model, name, user_type, fn):
|
||||
path = self.path.format(model)
|
||||
user_created = get_user_model().objects.filter(username=name)
|
||||
if not user_created:
|
||||
user = get_user_model().objects.create_user(
|
||||
name,
|
||||
email="test@example.com",
|
||||
password="test",
|
||||
type=user_type,
|
||||
first_name="test",
|
||||
last_name="user",
|
||||
)
|
||||
self.client.login(username=name, password="test")
|
||||
return fn(path)
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
cls.path = "/api/v2/{}/"
|
||||
|
||||
def test_guest_permissions_success(self):
|
||||
for model in self.URLS:
|
||||
response = self.logged_in_test_model(model, "guest", GUEST, self.client.get)
|
||||
self.assertEqual(
|
||||
response.status_code, 200, msg="Invalid for model {}".format(model)
|
||||
)
|
||||
|
||||
def test_guest_permissions_failure(self):
|
||||
for model in self.URLS:
|
||||
response = self.logged_in_test_model(
|
||||
model, "guest", GUEST, self.client.post
|
||||
)
|
||||
self.assertEqual(
|
||||
response.status_code, 403, msg="Invalid for model {}".format(model)
|
||||
)
|
||||
|
||||
def test_dj_get_permissions(self):
|
||||
for model in self.URLS:
|
||||
response = self.logged_in_test_model(model, "dj", DJ, self.client.get)
|
||||
self.assertEqual(
|
||||
response.status_code, 200, msg="Invalid for model {}".format(model)
|
||||
)
|
||||
|
||||
def test_dj_post_permissions(self):
|
||||
user = get_user_model().objects.create_user(
|
||||
"test-dj",
|
||||
email="test@example.com",
|
||||
password="test",
|
||||
type=DJ,
|
||||
first_name="test",
|
||||
last_name="user",
|
||||
)
|
||||
f = baker.make("libretimeapi.File", owner=user)
|
||||
model = "files/{}".format(f.id)
|
||||
path = self.path.format(model)
|
||||
self.client.login(username="test-dj", password="test")
|
||||
response = self.client.patch(path, {"name": "newFilename"})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_dj_post_permissions_failure(self):
|
||||
user = get_user_model().objects.create_user(
|
||||
"test-dj",
|
||||
email="test@example.com",
|
||||
password="test",
|
||||
type=DJ,
|
||||
first_name="test",
|
||||
last_name="user",
|
||||
)
|
||||
f = baker.make("libretimeapi.File")
|
||||
model = "files/{}".format(f.id)
|
||||
path = self.path.format(model)
|
||||
self.client.login(username="test-dj", password="test")
|
||||
response = self.client.patch(path, {"name": "newFilename"})
|
||||
self.assertEqual(response.status_code, 403)
|
||||
220
api/libretime_api/tests/test_views.py
Normal file
220
api/libretime_api/tests/test_views.py
Normal file
|
|
@ -0,0 +1,220 @@
|
|||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import AnonymousUser
|
||||
from django.utils import dateparse
|
||||
from model_bakery import baker
|
||||
from rest_framework.test import APIRequestFactory, APITestCase
|
||||
|
||||
from libretimeapi.views import FileViewSet
|
||||
|
||||
|
||||
class TestFileViewSet(APITestCase):
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
cls.path = "/api/v2/files/{id}/download/"
|
||||
cls.token = settings.CONFIG.get("general", "api_key")
|
||||
|
||||
def test_invalid(self):
|
||||
path = self.path.format(id="a")
|
||||
self.client.credentials(HTTP_AUTHORIZATION="Api-Key {}".format(self.token))
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_does_not_exist(self):
|
||||
path = self.path.format(id="1")
|
||||
self.client.credentials(HTTP_AUTHORIZATION="Api-Key {}".format(self.token))
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
def test_exists(self):
|
||||
music_dir = baker.make(
|
||||
"libretimeapi.MusicDir",
|
||||
directory=os.path.join(os.path.dirname(__file__), "resources"),
|
||||
)
|
||||
f = baker.make(
|
||||
"libretimeapi.File",
|
||||
directory=music_dir,
|
||||
mime="audio/mp3",
|
||||
filepath="song.mp3",
|
||||
)
|
||||
path = self.path.format(id=str(f.pk))
|
||||
self.client.credentials(HTTP_AUTHORIZATION="Api-Key {}".format(self.token))
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
|
||||
class TestScheduleViewSet(APITestCase):
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
cls.path = "/api/v2/schedule/"
|
||||
cls.token = settings.CONFIG.get("general", "api_key")
|
||||
|
||||
def test_schedule_item_full_length(self):
|
||||
music_dir = baker.make(
|
||||
"libretimeapi.MusicDir",
|
||||
directory=os.path.join(os.path.dirname(__file__), "resources"),
|
||||
)
|
||||
f = baker.make(
|
||||
"libretimeapi.File",
|
||||
directory=music_dir,
|
||||
mime="audio/mp3",
|
||||
filepath="song.mp3",
|
||||
length=timedelta(seconds=40.86),
|
||||
cuein=timedelta(seconds=0),
|
||||
cueout=timedelta(seconds=40.8131),
|
||||
)
|
||||
show = baker.make(
|
||||
"libretimeapi.ShowInstance",
|
||||
starts=datetime.now(tz=timezone.utc) - timedelta(minutes=5),
|
||||
ends=datetime.now(tz=timezone.utc) + timedelta(minutes=5),
|
||||
)
|
||||
scheduleItem = baker.make(
|
||||
"libretimeapi.Schedule",
|
||||
starts=datetime.now(tz=timezone.utc),
|
||||
ends=datetime.now(tz=timezone.utc) + f.length,
|
||||
cue_out=f.cueout,
|
||||
instance=show,
|
||||
file=f,
|
||||
)
|
||||
self.client.credentials(HTTP_AUTHORIZATION="Api-Key {}".format(self.token))
|
||||
response = self.client.get(self.path)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
result = response.json()
|
||||
self.assertEqual(dateparse.parse_datetime(result[0]["ends"]), scheduleItem.ends)
|
||||
self.assertEqual(dateparse.parse_duration(result[0]["cue_out"]), f.cueout)
|
||||
|
||||
def test_schedule_item_trunc(self):
|
||||
music_dir = baker.make(
|
||||
"libretimeapi.MusicDir",
|
||||
directory=os.path.join(os.path.dirname(__file__), "resources"),
|
||||
)
|
||||
f = baker.make(
|
||||
"libretimeapi.File",
|
||||
directory=music_dir,
|
||||
mime="audio/mp3",
|
||||
filepath="song.mp3",
|
||||
length=timedelta(seconds=40.86),
|
||||
cuein=timedelta(seconds=0),
|
||||
cueout=timedelta(seconds=40.8131),
|
||||
)
|
||||
show = baker.make(
|
||||
"libretimeapi.ShowInstance",
|
||||
starts=datetime.now(tz=timezone.utc) - timedelta(minutes=5),
|
||||
ends=datetime.now(tz=timezone.utc) + timedelta(seconds=20),
|
||||
)
|
||||
scheduleItem = baker.make(
|
||||
"libretimeapi.Schedule",
|
||||
starts=datetime.now(tz=timezone.utc),
|
||||
ends=datetime.now(tz=timezone.utc) + f.length,
|
||||
instance=show,
|
||||
file=f,
|
||||
)
|
||||
self.client.credentials(HTTP_AUTHORIZATION="Api-Key {}".format(self.token))
|
||||
response = self.client.get(self.path)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
result = response.json()
|
||||
self.assertEqual(dateparse.parse_datetime(result[0]["ends"]), show.ends)
|
||||
expected = show.ends - scheduleItem.starts
|
||||
self.assertEqual(dateparse.parse_duration(result[0]["cue_out"]), expected)
|
||||
self.assertNotEqual(
|
||||
dateparse.parse_datetime(result[0]["ends"]), scheduleItem.ends
|
||||
)
|
||||
|
||||
def test_schedule_item_invalid(self):
|
||||
music_dir = baker.make(
|
||||
"libretimeapi.MusicDir",
|
||||
directory=os.path.join(os.path.dirname(__file__), "resources"),
|
||||
)
|
||||
f = baker.make(
|
||||
"libretimeapi.File",
|
||||
directory=music_dir,
|
||||
mime="audio/mp3",
|
||||
filepath="song.mp3",
|
||||
length=timedelta(seconds=40.86),
|
||||
cuein=timedelta(seconds=0),
|
||||
cueout=timedelta(seconds=40.8131),
|
||||
)
|
||||
show = baker.make(
|
||||
"libretimeapi.ShowInstance",
|
||||
starts=datetime.now(tz=timezone.utc) - timedelta(minutes=5),
|
||||
ends=datetime.now(tz=timezone.utc) + timedelta(minutes=5),
|
||||
)
|
||||
scheduleItem = baker.make(
|
||||
"libretimeapi.Schedule",
|
||||
starts=datetime.now(tz=timezone.utc),
|
||||
ends=datetime.now(tz=timezone.utc) + f.length,
|
||||
cue_out=f.cueout,
|
||||
instance=show,
|
||||
file=f,
|
||||
)
|
||||
invalidScheduleItem = baker.make(
|
||||
"libretimeapi.Schedule",
|
||||
starts=show.ends + timedelta(minutes=1),
|
||||
ends=show.ends + timedelta(minutes=1) + f.length,
|
||||
cue_out=f.cueout,
|
||||
instance=show,
|
||||
file=f,
|
||||
)
|
||||
self.client.credentials(HTTP_AUTHORIZATION="Api-Key {}".format(self.token))
|
||||
response = self.client.get(self.path, {"is_valid": True})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
result = response.json()
|
||||
# The invalid item should be filtered out and not returned
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertEqual(dateparse.parse_datetime(result[0]["ends"]), scheduleItem.ends)
|
||||
self.assertEqual(dateparse.parse_duration(result[0]["cue_out"]), f.cueout)
|
||||
|
||||
def test_schedule_item_range(self):
|
||||
music_dir = baker.make(
|
||||
"libretimeapi.MusicDir",
|
||||
directory=os.path.join(os.path.dirname(__file__), "resources"),
|
||||
)
|
||||
f = baker.make(
|
||||
"libretimeapi.File",
|
||||
directory=music_dir,
|
||||
mime="audio/mp3",
|
||||
filepath="song.mp3",
|
||||
length=timedelta(seconds=40.86),
|
||||
cuein=timedelta(seconds=0),
|
||||
cueout=timedelta(seconds=40.8131),
|
||||
)
|
||||
filter_point = datetime.now(tz=timezone.utc)
|
||||
|
||||
show = baker.make(
|
||||
"libretimeapi.ShowInstance",
|
||||
starts=filter_point - timedelta(minutes=5),
|
||||
ends=filter_point + timedelta(minutes=5),
|
||||
)
|
||||
schedule_item = baker.make(
|
||||
"libretimeapi.Schedule",
|
||||
starts=filter_point,
|
||||
ends=filter_point + f.length,
|
||||
cue_out=f.cueout,
|
||||
instance=show,
|
||||
file=f,
|
||||
)
|
||||
previous_item = baker.make(
|
||||
"libretimeapi.Schedule",
|
||||
starts=filter_point - timedelta(minutes=5),
|
||||
ends=filter_point - timedelta(minutes=5) + f.length,
|
||||
cue_out=f.cueout,
|
||||
instance=show,
|
||||
file=f,
|
||||
)
|
||||
self.client.credentials(HTTP_AUTHORIZATION="Api-Key {}".format(self.token))
|
||||
range_start = (filter_point - timedelta(minutes=1)).isoformat(
|
||||
timespec="seconds"
|
||||
)
|
||||
range_end = (filter_point + timedelta(minutes=1)).isoformat(timespec="seconds")
|
||||
response = self.client.get(
|
||||
self.path, {"starts__range": "{},{}".format(range_start, range_end)}
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
result = response.json()
|
||||
# The previous_item should be filtered out and not returned
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertEqual(
|
||||
dateparse.parse_datetime(result[0]["starts"]), schedule_item.starts
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue