ucast/ucast/tests/conftest.py

import json
import shutil
import tempfile
from datetime import datetime
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Tuple
from unittest import mock

import pytest
import rq
from django.conf import settings
from django.core.management import call_command
from fakeredis import FakeRedis

from ucast import queue, tests
from ucast.models import User
from ucast.service import cover, storage, util, videoutil, youtube


@pytest.fixture(scope="session", autouse=True)
def default_config():
    settings.DOWNLOAD_ROOT = Path("does/not/exist")
    settings.REDIS_URL = "no redis"


@pytest.fixture(scope="session")
def django_db_setup(django_db_setup, django_db_blocker):
    with django_db_blocker.unblock():
        fixture_path = tests.DIR_TESTFILES / "fixture" / "models.json"
        call_command("loaddata", fixture_path)


def _create_download_dir() -> Tuple[Path, TemporaryDirectory]:
    tmpdir_o = tempfile.TemporaryDirectory()
    tmpdir = Path(tmpdir_o.name)
    settings.DOWNLOAD_ROOT = tmpdir

    # Copy channel avatars
    store = storage.Storage()

    for slug, avatar in (
        ("ThetaDev", "a1"),
        ("media_ccc_de", "a3"),
        ("Creative_Commons", "a4"),
    ):
        cf = store.get_or_create_channel_folder(slug)
        shutil.copyfile(
            tests.DIR_TESTFILES / "avatar" / f"{avatar}.jpg", cf.file_avatar
        )
        videoutil.resize_avatar(cf.file_avatar, cf.file_avatar_sm)

    return tmpdir, tmpdir_o


def _add_download_dir_content():
    store = storage.Storage()

    with open(tests.DIR_TESTFILES / "object" / "videodetails.json") as f:
        videodetails = json.load(f)

    for vid in ("ZPxEr4YdWt8", "_I5IFObm_-k", "mmEDPbbSnaY", "Cda4zS-1j-k"):
        video_detail = videodetails[vid]
        channel_name = video_detail["channel_name"]
        channel_slug = util.get_slug(channel_name)
        published = datetime.fromisoformat(video_detail["published"])
        title = video_detail["title"]
        video_slug = util.get_slug(f"{published.strftime('%Y%m%d')}_{title}")
        description = video_detail["description"]

        cf = store.get_or_create_channel_folder(channel_slug)
        file_audio = cf.get_audio(video_slug)
        file_tn = cf.get_thumbnail(video_slug)
        file_cover = cf.get_cover(video_slug)

        shutil.copyfile(tests.DIR_TESTFILES / "audio" / "audio1.mp3", file_audio)
        shutil.copyfile(tests.DIR_TESTFILES / "thumbnail" / f"{vid}.webp", file_tn)
        videoutil.resize_thumbnail(file_tn, cf.get_thumbnail(video_slug, True))
        cover.create_cover_file(
            file_tn,
            cf.file_avatar,
            title,
            channel_name,
            cover.COVER_STYLE_BLUR,
            file_cover,
        )
        videoutil.tag_audio(
            file_audio,
            title,
            channel_name,
            published.date(),
            f"{description}\n\nhttps://youtu.be/{vid}",
            file_cover,
        )


@pytest.fixture
def download_dir() -> Path:
    tmpdir, tmpdir_o = _create_download_dir()
    yield tmpdir


@pytest.fixture(scope="session")
def download_dir_content() -> Path:
    tmpdir, tmpdir_o = _create_download_dir()
    settings.DOWNLOAD_ROOT = tmpdir
    _add_download_dir_content()
    yield tmpdir


@pytest.fixture
def download_dir_content_mut() -> Path:
    tmpdir, tmpdir_o = _create_download_dir()
    settings.DOWNLOAD_ROOT = tmpdir
    _add_download_dir_content()
    yield tmpdir


@pytest.fixture
def mock_redis(mocker) -> FakeRedis:
    redis = FakeRedis()
    mocker.patch.object(queue, "get_redis_connection")
    queue.get_redis_connection.return_value = redis
    return redis


@pytest.fixture
def rq_queue(mocker) -> rq.Queue:
    test_queue = rq.Queue(is_async=False, connection=FakeRedis())
    mocker.patch.object(queue, "get_queue")
    queue.get_queue.return_value = test_queue
    return test_queue


@pytest.fixture
def mock_download_audio(mocker) -> mock.Mock:
    def mockfn_download_audio(
        video_id: str, download_path: Path, sponsorblock=False
    ) -> youtube.VideoDetails:
        shutil.copyfile(tests.DIR_TESTFILES / "audio" / "audio1.mp3", download_path)
        return tests.get_video_details(video_id)

    download_mock: mock.Mock = mocker.patch.object(youtube, "download_audio")
    download_mock.side_effect = mockfn_download_audio
    return download_mock


@pytest.fixture
def mock_get_video_details(mocker) -> mock.Mock:
    video_details_mock: mock.Mock = mocker.patch.object(youtube, "get_video_details")
    video_details_mock.side_effect = tests.get_video_details
    return video_details_mock


@pytest.fixture
def mock_get_channel_metadata(mocker) -> mock.Mock:
    channel_meta_mock: mock.Mock = mocker.patch.object(youtube, "get_channel_metadata")
    channel_meta_mock.side_effect = tests.get_channel_metadata
    return channel_meta_mock


@pytest.fixture
def user_admin(db):
    return User.objects.get(id=1)


@pytest.fixture
def feed_key(user_admin) -> str:
    return user_admin.get_feed_key()