added test_stream_to_buffer
Browse files- tests/test_request.py +3 -1
- tests/test_streams.py +18 -0
tests/test_request.py
CHANGED
|
@@ -10,6 +10,7 @@ from pytube import request
|
|
| 10 |
@mock.patch("pytube.request.filesize", return_value=3 * 8 * 1024)
|
| 11 |
@mock.patch("pytube.request.urlopen")
|
| 12 |
def test_streaming(mock_urlopen, filesize):
|
|
|
|
| 13 |
fake_stream_binary = [
|
| 14 |
os.urandom(8 * 1024),
|
| 15 |
os.urandom(8 * 1024),
|
|
@@ -19,9 +20,10 @@ def test_streaming(mock_urlopen, filesize):
|
|
| 19 |
response = mock.Mock()
|
| 20 |
response.read.side_effect = fake_stream_binary
|
| 21 |
mock_urlopen.return_value = response
|
|
|
|
| 22 |
response = request.stream("http://fakeassurl.gov")
|
|
|
|
| 23 |
call_count = len(list(response))
|
| 24 |
-
|
| 25 |
assert call_count == 3
|
| 26 |
|
| 27 |
|
|
|
|
| 10 |
@mock.patch("pytube.request.filesize", return_value=3 * 8 * 1024)
|
| 11 |
@mock.patch("pytube.request.urlopen")
|
| 12 |
def test_streaming(mock_urlopen, filesize):
|
| 13 |
+
# Given
|
| 14 |
fake_stream_binary = [
|
| 15 |
os.urandom(8 * 1024),
|
| 16 |
os.urandom(8 * 1024),
|
|
|
|
| 20 |
response = mock.Mock()
|
| 21 |
response.read.side_effect = fake_stream_binary
|
| 22 |
mock_urlopen.return_value = response
|
| 23 |
+
# When
|
| 24 |
response = request.stream("http://fakeassurl.gov")
|
| 25 |
+
# Then
|
| 26 |
call_count = len(list(response))
|
|
|
|
| 27 |
assert call_count == 3
|
| 28 |
|
| 29 |
|
tests/test_streams.py
CHANGED
|
@@ -8,6 +8,24 @@ from pytube import request
|
|
| 8 |
from pytube import Stream, streams
|
| 9 |
|
| 10 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 11 |
def test_filesize(cipher_signature, mocker):
|
| 12 |
mocker.patch.object(request, "head")
|
| 13 |
request.head.return_value = {"content-length": "6796391"}
|
|
|
|
| 8 |
from pytube import Stream, streams
|
| 9 |
|
| 10 |
|
| 11 |
+
@mock.patch("pytube.streams.request")
|
| 12 |
+
def test_stream_to_buffer(mock_request, cipher_signature):
|
| 13 |
+
# Given
|
| 14 |
+
stream_bytes = iter(
|
| 15 |
+
[
|
| 16 |
+
bytes(os.urandom(8 * 1024)),
|
| 17 |
+
bytes(os.urandom(8 * 1024)),
|
| 18 |
+
bytes(os.urandom(8 * 1024)),
|
| 19 |
+
]
|
| 20 |
+
)
|
| 21 |
+
mock_request.stream.return_value = stream_bytes
|
| 22 |
+
buffer = MagicMock()
|
| 23 |
+
# When
|
| 24 |
+
cipher_signature.streams[0].stream_to_buffer(buffer)
|
| 25 |
+
# Then
|
| 26 |
+
assert buffer.write.call_count == 3
|
| 27 |
+
|
| 28 |
+
|
| 29 |
def test_filesize(cipher_signature, mocker):
|
| 30 |
mocker.patch.object(request, "head")
|
| 31 |
request.head.return_value = {"content-length": "6796391"}
|