move coverage

README.rst

@@ -14,7 +14,7 @@ pytube
   :alt: Documentation Status
   :target: http://python-pytube.readthedocs.io/en/latest/?badge=latest
 
-.. image:: https://coveralls.io/repos/github/nficano/pytube/badge.svg?branch=master#23e6f7ac5623e96f
+.. image:: https://coveralls.io/repos/github/nficano/pytube/badge.svg?branch=master#23e6f7ac5623e
   :alt: Code Coverage
   :target: https://coveralls.io/github/nficano/pytube?branch=master
 

pytube/request.py

@@ -4,7 +4,7 @@ from pytube.compat import urlopen
 
 
 def get(
-    url=None, processes=2, headers=False,
+    url=None, headers=False,
     streaming=False, chunk_size=8 * 1024,
 ):
     """Send an http GET request.
@@ -13,8 +13,6 @@ def get(
         The URL to perform the GET request for.
     :param list urls:
         List of URLs to perform the GET request for concurrently.
-    :param int processes:
-        How many worker processes to start.
     :param bool headers:
         Only return the http headers.
     :param bool streaming:

tests/conftest.py

@@ -11,32 +11,41 @@ import pytest
 from pytube import YouTube
 
 
-def load_from_playback_file(filename):
-    cur_dir = os.path.dirname(os.path.realpath(__file__))
+def load_playback_file(filename):
+    cur_fp = os.path.realpath(__file__)
+    cur_dir = os.path.dirname(cur_fp)
     fp = os.path.join(cur_dir, 'mocks', filename)
-    video = None
     with gzip.open(fp, 'rb') as fh:
-        video = json.loads(fh.read().decode('utf-8'))
-    yt = YouTube(
-        url='https://www.youtube.com/watch?v=9bZkp7q19f0',
-        defer_prefetch_init=True,
-    )
-    yt.watch_html = video['watch_html']
-    yt.js = video['js']
-    yt.vid_info = video['video_info']
+        content = fh.read().decode('utf-8')
+        return json.loads(content)
+
+
+def load_and_init_from_playback_file(filename):
+    pb = load_playback_file(filename)
+    yt = YouTube(pb['url'], defer_prefetch_init=True)
+    yt.watch_html = pb['watch_html']
+    yt.js = pb['js']
+    yt.vid_info = pb['video_info']
     yt.init()
     return yt
 
 
 @pytest.fixture
-def gangnam_style():
+def cipher_signature():
     """Youtube instance initialized with video id 9bZkp7q19f0."""
     filename = 'yt-video-9bZkp7q19f0-1507588332.json.tar.gz'
-    return load_from_playback_file(filename)
+    return load_and_init_from_playback_file(filename)
 
 
 @pytest.fixture
-def youtube_captions_and_subtitles():
+def presigned_video():
     """Youtube instance initialized with video id QRS8MkLhQmM."""
     filename = 'yt-video-QRS8MkLhQmM-1507588031.json.tar.gz'
-    return load_from_playback_file(filename)
+    return load_and_init_from_playback_file(filename)
+
+
+@pytest.fixture
+def age_restricted():
+    """Youtube instance initialized with video id zRbsm3e2ltw."""
+    filename = 'yt-video-zRbsm3e2ltw-1507777044.json.tar.gz'
+    return load_playback_file(filename)

tests/test_extract.py

@@ -15,11 +15,11 @@ def test_extract_watch_url():
     assert watch_url == 'https://youtube.com/watch?v=9bZkp7q19f0'
 
 
-def test_info_url(gangnam_style):
+def test_info_url(cipher_signature):
     video_info_url = extract.video_info_url(
-        video_id=gangnam_style.video_id,
-        watch_url=gangnam_style.watch_url,
-        watch_html=gangnam_style.watch_html,
+        video_id=cipher_signature.video_id,
+        watch_url=cipher_signature.watch_url,
+        watch_html=cipher_signature.watch_html,
     )
     expected = (
         'https://youtube.com/get_video_info?video_id=9bZkp7q19f0&el=%24el'
@@ -29,7 +29,15 @@ def test_info_url(gangnam_style):
     assert video_info_url == expected
 
 
-def test_js_url(gangnam_style):
+def test_js_url(cipher_signature):
     expected = 'https://youtube.com/yts/jsbin/player-vflOdyxa4/en_US/base.js'
-    result = extract.js_url(gangnam_style.watch_html)
+    result = extract.js_url(cipher_signature.watch_html)
     assert expected == result
+
+
+def test_age_restricted(age_restricted):
+    assert extract.is_age_restricted(age_restricted['watch_html'])
+
+
+def test_non_age_restricted(cipher_signature):
+    assert not extract.is_age_restricted(cipher_signature.watch_html)

tests/test_mixins.py

@@ -1,3 +1,3 @@
 # -*- coding: utf-8 -*-
-def test_pre_signed_video(youtube_captions_and_subtitles):
-    assert youtube_captions_and_subtitles.streams.count() == 15
+def test_pre_signed_video(presigned_video):
+    assert presigned_video.streams.count() == 15

tests/test_query.py

@@ -3,8 +3,8 @@
 import pytest
 
 
-def test_count(gangnam_style):
-    assert gangnam_style.streams.count() == 22
+def test_count(cipher_signature):
+    assert cipher_signature.streams.count() == 22
 
 
 @pytest.mark.parametrize(
@@ -27,29 +27,31 @@ def test_count(gangnam_style):
         ({'custom_filter_functions': [lambda s: s.itag == '22']}, ['22']),
     ],
 )
-def test_filters(test_input, expected, gangnam_style):
-    result = [s.itag for s in gangnam_style.streams.filter(**test_input).all()]
+def test_filters(test_input, expected, cipher_signature):
+    result = [s.itag for s in cipher_signature.streams.filter(
+        **test_input
+    ).all()]
     assert result == expected
 
 
 @pytest.mark.parametrize('test_input', ['first', 'last'])
-def test_empty(test_input, gangnam_style):
-    query = gangnam_style.streams.filter(video_codec='vp20')
+def test_empty(test_input, cipher_signature):
+    query = cipher_signature.streams.filter(video_codec='vp20')
     fn = getattr(query, test_input)
     assert fn() is None
 
 
-def test_get_last(gangnam_style):
-    assert gangnam_style.streams.last().itag == '251'
+def test_get_last(cipher_signature):
+    assert cipher_signature.streams.last().itag == '251'
 
 
-def test_get_first(gangnam_style):
-    assert gangnam_style.streams.first().itag == '22'
+def test_get_first(cipher_signature):
+    assert cipher_signature.streams.first().itag == '22'
 
 
-def test_order_by(gangnam_style):
+def test_order_by(cipher_signature):
     itags = [
-        s.itag for s in gangnam_style.streams
+        s.itag for s in cipher_signature.streams
         .filter(progressive=True)
         .order_by('itag')
         .all()
@@ -58,9 +60,9 @@ def test_order_by(gangnam_style):
     assert itags == ['17', '18', '22', '36', '43']
 
 
-def test_order_by_descending(gangnam_style):
+def test_order_by_descending(cipher_signature):
     itags = [
-        s.itag for s in gangnam_style.streams
+        s.itag for s in cipher_signature.streams
         .filter(progressive=True)
         .order_by('itag')
         .desc()
@@ -70,9 +72,9 @@ def test_order_by_descending(gangnam_style):
     assert itags == ['43', '36', '22', '18', '17']
 
 
-def test_order_by_ascending(gangnam_style):
+def test_order_by_ascending(cipher_signature):
     itags = [
-        s.itag for s in gangnam_style.streams
+        s.itag for s in cipher_signature.streams
         .filter(progressive=True)
         .order_by('itag')
         .asc()
@@ -82,9 +84,9 @@ def test_order_by_ascending(gangnam_style):
     assert itags == ['17', '18', '22', '36', '43']
 
 
-def test_get_by_itag(gangnam_style):
-    assert gangnam_style.streams.get_by_itag(22).itag == '22'
+def test_get_by_itag(cipher_signature):
+    assert cipher_signature.streams.get_by_itag(22).itag == '22'
 
 
-def test_get_by_non_existent_itag(gangnam_style):
-    assert not gangnam_style.streams.get_by_itag(22983)
+def test_get_by_non_existent_itag(cipher_signature):
+    assert not cipher_signature.streams.get_by_itag(22983)

tests/test_request.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+import os
+
+import mock
+
+from pytube import request
+
+
+@mock.patch('pytube.request.urlopen')
+def test_get_streaming(mock_urlopen):
+    fake_stream_binary = [
+        iter(os.urandom(8 * 1024)),
+        iter(os.urandom(8 * 1024)),
+        iter(os.urandom(8 * 1024)),
+        None,
+    ]
+    response = mock.Mock()
+    response.read.side_effect = fake_stream_binary
+    mock_urlopen.return_value = response
+    response = request.get('fakeassurl.gov', streaming=True)
+    call_count = 0
+    for i in response:
+        call_count += 1
+    assert call_count == 3
+
+
+@mock.patch('pytube.request.urlopen')
+def test_get_headers(mock_urlopen):
+    response = mock.Mock()
+    response.info.return_value = {'Content-Length': '16384'}
+    mock_urlopen.return_value = response
+    response = request.get('fakeassurl.gov', headers=True)
+    assert response == {'Content-Length': '16384'}
+
+
+@mock.patch('pytube.request.urlopen')
+def test_get(mock_urlopen):
+    response = mock.Mock()
+    response.read.return_value = '<html></html>'.encode('utf-8')
+    mock_urlopen.return_value = response
+    response = request.get('fakeassurl.gov')
+    assert response == '<html></html>'

tests/test_streams.py

@@ -7,19 +7,19 @@ from pytube import request
 from pytube import Stream
 
 
-def test_filesize(gangnam_style, mocker):
+def test_filesize(cipher_signature, mocker):
     mocker.patch.object(request, 'get')
     request.get.return_value = {'Content-Length': '6796391'}
-    assert gangnam_style.streams.first().filesize == 6796391
+    assert cipher_signature.streams.first().filesize == 6796391
 
 
-def test_default_filename(gangnam_style):
+def test_default_filename(cipher_signature):
     expected = 'PSY - GANGNAM STYLE(강남스타일) MV.mp4'
-    stream = gangnam_style.streams.first()
+    stream = cipher_signature.streams.first()
     assert stream.default_filename == expected
 
 
-def test_download(gangnam_style, mocker):
+def test_download(cipher_signature, mocker):
     mocker.patch.object(request, 'get')
     request.get.side_effect = [
         {'Content-Length': '16384'},
@@ -27,23 +27,23 @@ def test_download(gangnam_style, mocker):
         iter([str(random.getrandbits(8 * 1024))]),
     ]
     with mock.patch('pytube.streams.open', mock.mock_open(), create=True):
-        stream = gangnam_style.streams.first()
+        stream = cipher_signature.streams.first()
         stream.download()
 
 
-def test_progressive_streams_return_includes_audio_track(gangnam_style):
-    stream = gangnam_style.streams.filter(progressive=True).first()
+def test_progressive_streams_return_includes_audio_track(cipher_signature):
+    stream = cipher_signature.streams.filter(progressive=True).first()
     assert stream.includes_audio_track
 
 
-def test_progressive_streams_return_includes_video_track(gangnam_style):
-    stream = gangnam_style.streams.filter(progressive=True).first()
+def test_progressive_streams_return_includes_video_track(cipher_signature):
+    stream = cipher_signature.streams.filter(progressive=True).first()
     assert stream.includes_video_track
 
 
-def test_on_progress_hook(gangnam_style, mocker):
+def test_on_progress_hook(cipher_signature, mocker):
     callback_fn = mock.MagicMock()
-    gangnam_style.register_on_progress_callback(callback_fn)
+    cipher_signature.register_on_progress_callback(callback_fn)
 
     mocker.patch.object(request, 'get')
     request.get.side_effect = [
@@ -52,7 +52,7 @@ def test_on_progress_hook(gangnam_style, mocker):
         iter([str(random.getrandbits(8 * 1024))]),
     ]
     with mock.patch('pytube.streams.open', mock.mock_open(), create=True):
-        stream = gangnam_style.streams.first()
+        stream = cipher_signature.streams.first()
         stream.download()
     assert callback_fn.called
     args, _ = callback_fn.call_args
@@ -61,9 +61,9 @@ def test_on_progress_hook(gangnam_style, mocker):
     assert isinstance(stream, Stream)
 
 
-def test_on_complete_hook(gangnam_style, mocker):
+def test_on_complete_hook(cipher_signature, mocker):
     callback_fn = mock.MagicMock()
-    gangnam_style.register_on_complete_callback(callback_fn)
+    cipher_signature.register_on_complete_callback(callback_fn)
 
     mocker.patch.object(request, 'get')
     request.get.side_effect = [
@@ -72,13 +72,13 @@ def test_on_complete_hook(gangnam_style, mocker):
         iter([str(random.getrandbits(8 * 1024))]),
     ]
     with mock.patch('pytube.streams.open', mock.mock_open(), create=True):
-        stream = gangnam_style.streams.first()
+        stream = cipher_signature.streams.first()
         stream.download()
     assert callback_fn.called
 
 
-def test_repr_for_audio_streams(gangnam_style):
-    stream = str(gangnam_style.streams.filter(only_audio=True).first())
+def test_repr_for_audio_streams(cipher_signature):
+    stream = str(cipher_signature.streams.filter(only_audio=True).first())
     expected = (
         '<Stream: itag="140" mime_type="audio/mp4" abr="128kbps" '
         'acodec="mp4a.40.2">'
@@ -86,8 +86,8 @@ def test_repr_for_audio_streams(gangnam_style):
     assert stream == expected
 
 
-def test_repr_for_video_streams(gangnam_style):
-    stream = str(gangnam_style.streams.filter(only_video=True).first())
+def test_repr_for_video_streams(cipher_signature):
+    stream = str(cipher_signature.streams.filter(only_video=True).first())
     expected = (
         '<Stream: itag="137" mime_type="video/mp4" res="1080p" '
         'fps="30fps" vcodec="avc1.640028">'
@@ -95,8 +95,8 @@ def test_repr_for_video_streams(gangnam_style):
     assert stream == expected
 
 
-def test_repr_for_progressive_streams(gangnam_style):
-    stream = str(gangnam_style.streams.filter(progressive=True).first())
+def test_repr_for_progressive_streams(cipher_signature):
+    stream = str(cipher_signature.streams.filter(progressive=True).first())
     expected = (
         '<Stream: itag="22" mime_type="video/mp4" res="720p" '
         'fps="30fps" vcodec="avc1.64001F" acodec="mp4a.40.2">'
@@ -104,8 +104,8 @@ def test_repr_for_progressive_streams(gangnam_style):
     assert stream == expected
 
 
-def test_repr_for_adaptive_streams(gangnam_style):
-    stream = str(gangnam_style.streams.filter(adaptive=True).first())
+def test_repr_for_adaptive_streams(cipher_signature):
+    stream = str(cipher_signature.streams.filter(adaptive=True).first())
     expected = (
         '<Stream: itag="137" mime_type="video/mp4" res="1080p" '
         'fps="30fps" vcodec="avc1.640028">'