unit tests

pytube/__main__.py

@@ -19,7 +19,6 @@ from pytube import Stream
 from pytube import StreamQuery
 from pytube.compat import parse_qsl
 from pytube.helpers import apply_mixin
-from pytube.helpers import memoize
 
 
 logger = logging.getLogger(__name__)
@@ -159,7 +158,6 @@ class YouTube(object):
             self.fmt_streams.append(video)
 
     @property
-    @memoize
     def streams(self):
         """Interface to query both adaptive (DASH) and progressive streams."""
         return StreamQuery([s for s in self.fmt_streams])

pytube/cipher.py

@@ -23,7 +23,6 @@ import re
 from itertools import chain
 
 from pytube.exceptions import RegexMatchError
-from pytube.helpers import memoize
 from pytube.helpers import regex_search
 
 
@@ -104,7 +103,6 @@ def get_transform_object(js, var):
     )
 
 
-@memoize
 def get_transform_map(js, var):
     """Build a transform function lookup.
 
@@ -235,7 +233,6 @@ def parse_function(js_func):
     return regex_search(pattern, js_func, groups=True)
 
 
-@memoize
 def get_signature(js, ciphered_signature):
     """Decipher the signature.
 

pytube/extract.py

@@ -5,7 +5,6 @@ from collections import OrderedDict
 
 from pytube.compat import quote
 from pytube.compat import urlencode
-from pytube.helpers import memoize
 from pytube.helpers import regex_search
 
 
@@ -88,7 +87,6 @@ def mime_type_codec(mime_type_codec):
     return mime_type, [c.strip() for c in codecs.split(',')]
 
 
-@memoize
 def get_ytplayer_config(watch_html):
     """Get the YouTube player configuration data from the watch html.
 

pytube/helpers.py

@@ -2,7 +2,6 @@
 """Various helper functions implemented by pytube."""
 from __future__ import absolute_import
 
-import functools
 import logging
 import pprint
 import re
@@ -88,16 +87,3 @@ def safe_filename(s, max_length=255):
     regex = re.compile(pattern, re.UNICODE)
     filename = regex.sub('', s)
     return filename[:max_length].rsplit(' ', 0)[0]
-
-
-def memoize(func):
-    """Decorate that caches input arguments for return values."""
-    cache = func.cache = {}
-
-    @functools.wraps(func)
-    def wrapper(*args, **kwargs):
-        key = str(args) + str(kwargs)
-        if key not in cache:
-            cache[key] = func(*args, **kwargs)
-        return cache[key]
-    return wrapper

pytube/mixins.py

@@ -34,6 +34,7 @@ def apply_signature(config_args, fmt, js):
             # For certain videos, YouTube will just provide them pre-signed, in
             # which case there's no real magic to download them and we can skip
             # the whole signature decrambling entirely.
+            logger.debug('signature found, skip decipher')
             continue
 
         signature = cipher.get_signature(js, stream['s'])
@@ -63,4 +64,7 @@ def apply_descrambler(stream_data, key):
         {k: unquote(v) for k, v in parse_qsl(i)}
         for i in stream_data[key].split(',')
     ]
-    logger.debug('applying descrambler\n%s', pprint.pformat(stream_data[key]))
+    logger.debug(
+        'applying descrambler\n%s',
+        pprint.pformat(stream_data[key], indent=2),
+    )

pytube/streams.py

@@ -15,7 +15,6 @@ import pprint
 
 from pytube import extract
 from pytube import request
-from pytube.helpers import memoize
 from pytube.helpers import safe_filename
 from pytube.itags import get_format_profile
 
@@ -133,7 +132,6 @@ class Stream(object):
         return video, audio
 
     @property
-    @memoize
     def filesize(self):
         """File size of the media stream in bytes."""
         headers = request.get(self.url, headers=True)

tests/conftest.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 """Reusable dependency injected testing components."""
+import gzip
 import json
 import os
 
@@ -8,13 +9,11 @@ import pytest
 from pytube import YouTube
 
 
-@pytest.fixture
-def gangnam_style():
-    """Youtube instance initialized with video id 9bZkp7q19f0."""
+def load_from_playback_file(filename):
     cur_dir = os.path.dirname(os.path.realpath(__file__))
-    fp = os.path.join(cur_dir, 'mocks', 'yt-video-9bZkp7q19f0.json')
+    fp = os.path.join(cur_dir, 'mocks', filename)
     video = None
-    with open(fp, 'r') as fh:
+    with gzip.open(fp, 'rb') as fh:
         video = json.loads(fh.read())
     yt = YouTube(
         url='https://www.youtube.com/watch?v=9bZkp7q19f0',
@@ -25,3 +24,17 @@ def gangnam_style():
     yt.vid_info = video['video_info']
     yt.init()
     return yt
+
+
+@pytest.fixture
+def gangnam_style():
+    """Youtube instance initialized with video id 9bZkp7q19f0."""
+    filename = 'yt-video-9bZkp7q19f0-1507588332.json.tar.gz'
+    return load_from_playback_file(filename)
+
+
+@pytest.fixture
+def youtube_captions_and_subtitles():
+    """Youtube instance initialized with video id QRS8MkLhQmM."""
+    filename = 'yt-video-QRS8MkLhQmM-1507588031.json.tar.gz'
+    return load_from_playback_file(filename)

tests/test_mixins.py

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

tests/test_query.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+"""Unit tests for the :class:`StreamQuery <StreamQuery>` class."""
+import pytest
+
+
+def test_count(gangnam_style):
+    assert gangnam_style.streams.count() == 22
+
+
+@pytest.mark.parametrize(
+    'test_input,expected', [
+        ({'progressive': True}, ['22', '43', '18', '36', '17']),
+        ({'resolution': '720p'}, ['22', '136', '247']),
+        ({'res': '720p'}, ['22', '136', '247']),
+        ({'fps': 30, 'resolution': '480p'}, ['135', '244']),
+        ({'mime_type': 'audio/mp4'}, ['140']),
+        ({'type': 'audio'}, ['140', '171', '249', '250', '251']),
+        ({'subtype': '3gpp'}, ['36', '17']),
+        ({'abr': '128kbps'}, ['43', '140', '171']),
+        ({'bitrate': '128kbps'}, ['43', '140', '171']),
+        ({'audio_codec': 'vorbis'}, ['43', '171']),
+        ({'video_codec': 'vp9'}, ['248', '247', '244', '243', '242', '278']),
+        ({'only_audio': True}, ['140', '171', '249', '250', '251']),
+        ({'only_video': True, 'video_codec': 'avc1.4d4015'}, ['133']),
+        ({'progressive': True}, ['22', '43', '18', '36', '17']),
+        ({'adaptive': True, 'resolution': '1080p'}, ['137', '248']),
+        ({'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()]
+    assert result == expected
+
+
+def test_get_last(gangnam_style):
+    assert gangnam_style.streams.last().itag == '251'
+
+
+def test_get_last_empty_results(gangnam_style):
+    assert not gangnam_style.streams.filter(video_codec='vp20').last()
+
+
+def test_get_first(gangnam_style):
+    assert gangnam_style.streams.first().itag == '22'
+
+
+def test_get_first_empty_results(gangnam_style):
+    assert not gangnam_style.streams.filter(video_codec='vp20').first()
+
+
+def test_order_by(gangnam_style):
+    itags = [
+        s.itag for s in gangnam_style.streams
+        .filter(progressive=True)
+        .order_by('itag')
+        .all()
+    ]
+
+    assert itags == ['17', '18', '22', '36', '43']
+
+
+def test_order_by_descending(gangnam_style):
+    itags = [
+        s.itag for s in gangnam_style.streams
+        .filter(progressive=True)
+        .order_by('itag')
+        .desc()
+        .all()
+    ]
+
+    assert itags == ['43', '36', '22', '18', '17']
+
+
+def test_order_by_ascending(gangnam_style):
+    itags = [
+        s.itag for s in gangnam_style.streams
+        .filter(progressive=True)
+        .order_by('itag')
+        .asc()
+        .all()
+    ]
+
+    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_non_existent_itag(gangnam_style):
+    assert not gangnam_style.streams.get_by_itag(22983)