# -*- coding: utf-8 -*- import os import random from datetime import datetime from unittest import mock from unittest.mock import MagicMock from pytube import request from pytube import Stream, streams @mock.patch("pytube.streams.request") def test_stream_to_buffer(mock_request, cipher_signature): # Given stream_bytes = iter( [ bytes(os.urandom(8 * 1024)), bytes(os.urandom(8 * 1024)), bytes(os.urandom(8 * 1024)), ] ) mock_request.stream.return_value = stream_bytes buffer = MagicMock() # When cipher_signature.streams[0].stream_to_buffer(buffer) # Then assert buffer.write.call_count == 3 def test_filesize(cipher_signature, mocker): mocker.patch.object(request, "head") request.head.return_value = {"content-length": "6796391"} assert cipher_signature.streams[0].filesize == 6796391 def test_filesize_approx(cipher_signature, mocker): mocker.patch.object(request, "head") request.head.return_value = {"content-length": "6796391"} stream = cipher_signature.streams[0] assert stream.filesize_approx == 22350604 stream.bitrate = None assert stream.filesize_approx == 6796391 def test_default_filename(cipher_signature): expected = "PSY - GANGNAM STYLE(강남스타일) MV.mp4" stream = cipher_signature.streams[0] assert stream.default_filename == expected def test_title(cipher_signature): expected = "title" cipher_signature.player_config_args["title"] = expected assert cipher_signature.title == expected expected = "title2" del cipher_signature.player_config_args["title"] cipher_signature.player_response = {"videoDetails": {"title": expected}} assert cipher_signature.title == expected def test_expiration(cipher_signature): assert cipher_signature.streams[0].expiration == datetime(2020, 1, 15, 21, 12, 5) def test_caption_tracks(presigned_video): assert len(presigned_video.caption_tracks) == 13 def test_captions(presigned_video): assert len(presigned_video.captions) == 13 def test_description(cipher_signature): expected = ( "PSY - ‘I LUV IT’ M/V @ https://youtu.be/Xvjnoagk6GU\n" "PSY - ‘New Face’ M/V @https://youtu.be/OwJPPaEyqhI\n\n" "PSY - 8TH ALBUM '4X2=8' on iTunes @\n" "https://smarturl.it/PSY_8thAlbum\n\n" "PSY - GANGNAM STYLE(강남스타일) on iTunes @ http://smarturl.it/PsyGangnam\n\n" "#PSY #싸이 #GANGNAMSTYLE #강남스타일\n\n" "More about PSY@\nhttp://www.youtube.com/officialpsy\n" "http://www.facebook.com/officialpsy\n" "http://twitter.com/psy_oppa\n" "https://www.instagram.com/42psy42\n" "http://iTunes.com/PSY\n" "http://sptfy.com/PSY\n" "http://weibo.com/psyoppa" ) assert cipher_signature.description == expected cipher_signature.player_response = {} expected = ( "PSY - ‘I LUV IT’ M/V @ https://youtu.be/Xvjnoagk6GU\n" "PSY - ‘New Face’ M/V @https://youtu.be/OwJPPaEyqhI\n" "PSY - 8TH ALBUM '4X2=8' on iTunes @\n" "https://smarturl.it/PSY_8thAlbum\n" "PSY - GANGNAM STYLE(강남스타일) on iTunes @ http://smarturl.it/PsyGangnam\n" "#PSY #싸이 #GANGNAMSTYLE #강남스타일\n" "More about PSY@\nhttp://www.youtube.com/officialpsy\n" "http://www.facebook.com/officialpsy\n" "http://twitter.com/psy_oppa\n" "https://www.instagram.com/42psy42\n" "http://iTunes.com/PSY\n" "http://sptfy.com/PSY\n" "http://weibo.com/psyoppa" ) assert cipher_signature.description == expected def test_rating(cipher_signature): assert cipher_signature.rating == 4.522203 def test_length(cipher_signature): assert cipher_signature.length == 252 def test_views(cipher_signature): assert cipher_signature.views == 3494704859 def test_download(cipher_signature, mocker): mocker.patch.object(request, "head") request.head.return_value = {"content-length": "16384"} mocker.patch.object(request, "stream") request.stream.return_value = iter([str(random.getrandbits(8 * 1024))]) with mock.patch("pytube.streams.open", mock.mock_open(), create=True): stream = cipher_signature.streams[0] stream.download() def test_download_with_prefix(cipher_signature, mocker): mocker.patch.object(request, "head") request.head.return_value = {"content-length": "16384"} mocker.patch.object(request, "stream") request.stream.return_value = iter([str(random.getrandbits(8 * 1024))]) streams.target_directory = MagicMock(return_value="/target") with mock.patch("pytube.streams.open", mock.mock_open(), create=True): stream = cipher_signature.streams[0] file_path = stream.download(filename_prefix="prefix") assert file_path == "/target/prefixPSY - GANGNAM STYLE(강남스타일) MV.mp4" def test_download_with_filename(cipher_signature, mocker): mocker.patch.object(request, "head") request.head.return_value = {"content-length": "16384"} mocker.patch.object(request, "stream") request.stream.return_value = iter([str(random.getrandbits(8 * 1024))]) streams.target_directory = MagicMock(return_value="/target") with mock.patch("pytube.streams.open", mock.mock_open(), create=True): stream = cipher_signature.streams[0] file_path = stream.download(filename="cool name bro") assert file_path == "/target/cool name bro.mp4" def test_download_with_existing(cipher_signature, mocker): mocker.patch.object(request, "head") request.head.return_value = {"content-length": "16384"} mocker.patch.object(request, "stream") streams.target_directory = MagicMock(return_value="/target") mocker.patch.object(os.path, "isfile") os.path.isfile.return_value = True with mock.patch("pytube.streams.open", mock.mock_open(), create=True): stream = cipher_signature.streams[0] mocker.patch.object(os.path, "getsize") os.path.getsize.return_value = stream.filesize file_path = stream.download() assert file_path == "/target/PSY - GANGNAM STYLE(강남스타일) MV.mp4" assert not request.stream.called def test_download_with_existing_no_skip(cipher_signature, mocker): mocker.patch.object(request, "head") request.head.return_value = {"content-length": "16384"} mocker.patch.object(request, "stream") request.stream.return_value = iter([str(random.getrandbits(8 * 1024))]) streams.target_directory = MagicMock(return_value="/target") mocker.patch.object(os.path, "isfile") os.path.isfile.return_value = True with mock.patch("pytube.streams.open", mock.mock_open(), create=True): stream = cipher_signature.streams[0] mocker.patch.object(os.path, "getsize") os.path.getsize.return_value = stream.filesize file_path = stream.download(skip_existing=False) assert file_path == "/target/PSY - GANGNAM STYLE(강남스타일) MV.mp4" assert request.stream.called def test_progressive_streams_return_includes_audio_track(cipher_signature): stream = cipher_signature.streams.filter(progressive=True)[0] assert stream.includes_audio_track def test_progressive_streams_return_includes_video_track(cipher_signature): stream = cipher_signature.streams.filter(progressive=True)[0] assert stream.includes_video_track def test_on_progress_hook(cipher_signature, mocker): callback_fn = mock.MagicMock() cipher_signature.register_on_progress_callback(callback_fn) mocker.patch.object(request, "head") request.head.return_value = {"content-length": "16384"} mocker.patch.object(request, "stream") request.stream.return_value = iter([str(random.getrandbits(8 * 1024))]) with mock.patch("pytube.streams.open", mock.mock_open(), create=True): stream = cipher_signature.streams[0] stream.download() assert callback_fn.called args, _ = callback_fn.call_args assert len(args) == 3 stream, _, _ = args assert isinstance(stream, Stream) def test_on_complete_hook(cipher_signature, mocker): callback_fn = mock.MagicMock() cipher_signature.register_on_complete_callback(callback_fn) mocker.patch.object(request, "head") request.head.return_value = {"content-length": "16384"} mocker.patch.object(request, "stream") request.stream.return_value = iter([str(random.getrandbits(8 * 1024))]) with mock.patch("pytube.streams.open", mock.mock_open(), create=True): stream = cipher_signature.streams[0] stream.download() assert callback_fn.called def test_author(cipher_signature): expected = "Test author" cipher_signature.player_response = {"videoDetails": {"author": expected}} assert cipher_signature.author == expected expected = "unknown" cipher_signature.player_response = {} assert cipher_signature.author == expected def test_thumbnail_when_in_details(cipher_signature): expected = "some url" cipher_signature.player_response = { "videoDetails": {"thumbnail": {"thumbnails": [{"url": expected}]}} } assert cipher_signature.thumbnail_url == expected def test_thumbnail_when_not_in_details(cipher_signature): expected = "https://img.youtube.com/vi/9bZkp7q19f0/maxresdefault.jpg" cipher_signature.player_response = {} assert cipher_signature.thumbnail_url == expected def test_repr_for_audio_streams(cipher_signature): stream = str(cipher_signature.streams.filter(only_audio=True)[0]) expected = ( '' ) assert stream == expected def test_repr_for_video_streams(cipher_signature): stream = str(cipher_signature.streams.filter(only_video=True)[0]) expected = ( '' ) assert stream == expected def test_repr_for_progressive_streams(cipher_signature): stream = str(cipher_signature.streams.filter(progressive=True)[0]) expected = ( '' ) assert stream == expected def test_repr_for_adaptive_streams(cipher_signature): stream = str(cipher_signature.streams.filter(adaptive=True)[0]) expected = ( '' ) assert stream == expected