60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
import os
|
|
|
|
import pytest
|
|
|
|
from ucast import queue, tests
|
|
from ucast.models import Channel, Video
|
|
from ucast.service import storage
|
|
from ucast.tasks import download
|
|
|
|
CHANNEL_ID_THETADEV = "UCGiJh0NZ52wRhYKYnuZI08Q"
|
|
VIDEO_ID_INTRO = "I0RRENheeTo"
|
|
VIDEO_SLUG_INTRO = "20211010_No_copyright_intro_free_fire_intro_no_text_free_copy_right_free_templates_free_download"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_download_video(download_dir, rq_queue):
|
|
video = Video.objects.get(video_id=VIDEO_ID_INTRO)
|
|
job = queue.enqueue(download.download_video, video)
|
|
|
|
store = storage.Storage()
|
|
cf = store.get_or_create_channel_folder(video.channel.slug)
|
|
|
|
assert job.is_finished
|
|
|
|
assert os.path.isfile(cf.get_audio(VIDEO_SLUG_INTRO))
|
|
assert os.path.isfile(cf.get_cover(VIDEO_SLUG_INTRO))
|
|
assert os.path.isfile(cf.get_thumbnail(VIDEO_SLUG_INTRO))
|
|
assert os.path.isfile(cf.get_thumbnail(VIDEO_SLUG_INTRO, True))
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_update_channel(
|
|
download_dir, rq_queue, mock_get_video_details, mock_download_audio
|
|
):
|
|
# Remove 2 videos from the database so they can be imported
|
|
Video.objects.get(video_id="ZPxEr4YdWt8").delete()
|
|
Video.objects.get(video_id="_I5IFObm_-k").delete()
|
|
|
|
channel = Channel.objects.get(channel_id=CHANNEL_ID_THETADEV)
|
|
job = rq_queue.enqueue(download.update_channel, channel)
|
|
assert job.is_finished
|
|
|
|
mock_download_audio.assert_any_call(
|
|
"_I5IFObm_-k",
|
|
download_dir / "ThetaDev" / "20180331_Easter_special_3D_printed_Bunny.mp3",
|
|
)
|
|
mock_download_audio.assert_any_call(
|
|
"ZPxEr4YdWt8",
|
|
download_dir / "ThetaDev" / "20190602_ThetaDev_Embedded_World_2019.mp3",
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_update_channels(rq_queue, mocker):
|
|
update_channel_mock = tests.GlobalMock()
|
|
mocker.patch.object(download, "update_channel", update_channel_mock)
|
|
job = rq_queue.enqueue(download.update_channels)
|
|
assert job.is_finished
|
|
|
|
assert update_channel_mock.n_calls == 3
|