Codex
75 lines
4.2 KiB
Python
75 lines
4.2 KiB
Python
import io
|
|
import os
|
|
import time
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
import playback
|
|
|
|
|
|
class PlaybackCacheTests(unittest.TestCase):
|
|
def test_private_cache_removed_after_use_and_abort(self):
|
|
for abort in (False, True):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
with patch.dict('os.environ', {'VISION_DATA':folder}), patch('playback.shutil.disk_usage', return_value=SimpleNamespace(free=10**12)), patch('playback.urllib.request.urlopen', return_value=io.BytesIO(b'video')) as opening:
|
|
try:
|
|
with playback.cached_source('http://127.0.0.1/source', 'Basic private', 8) as path:
|
|
self.assertEqual(Path(path).read_bytes(), b'video')
|
|
self.assertEqual(opening.call_args.args[0].get_header('Authorization'), 'Basic private')
|
|
if abort:
|
|
raise RuntimeError('browser left')
|
|
except RuntimeError:
|
|
self.assertTrue(abort)
|
|
self.assertEqual(list((Path(folder)/'playback-tmp').iterdir()), [])
|
|
|
|
def test_oversize_and_empty_sources_leave_no_cache(self):
|
|
for contents in (b'', b'123456789'):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
with patch.dict('os.environ', {'VISION_DATA':folder}), patch('playback.MAX_SOURCE_BYTES', 8), patch('playback.shutil.disk_usage', return_value=SimpleNamespace(free=1000)), patch('playback.urllib.request.urlopen', return_value=io.BytesIO(contents)):
|
|
with self.assertRaises(playback.PlaybackError):
|
|
with playback.cached_source('http://127.0.0.1/source', 'private', 8):
|
|
self.fail('invalid cache accepted')
|
|
self.assertEqual(list((Path(folder)/'playback-tmp').iterdir()), [])
|
|
|
|
def test_disk_reserve_checked_before_download(self):
|
|
with patch('playback.shutil.disk_usage', return_value=SimpleNamespace(free=1)), patch('playback.urllib.request.urlopen') as opening:
|
|
with self.assertRaises(playback.PlaybackError):
|
|
with playback.cached_source('unused', 'private', 8):
|
|
self.fail('low space accepted')
|
|
opening.assert_not_called()
|
|
|
|
def test_stale_buffers_are_removed_but_current_and_unrelated_files_remain(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
base=Path(folder)/'playback-tmp';base.mkdir()
|
|
old=base/'buffer-old';old.mkdir();(old/'source.mp4').write_bytes(b'old')
|
|
os.utime(old,(time.time()-7200,time.time()-7200))
|
|
fresh=base/'buffer-active';fresh.mkdir();(fresh/'source.mp4').write_bytes(b'active')
|
|
unrelated=base/'keep.txt';unrelated.write_text('keep')
|
|
with patch.dict('os.environ',{'VISION_DATA':folder}), patch('playback.shutil.disk_usage',return_value=SimpleNamespace(free=10**12)), patch('playback.urllib.request.urlopen',return_value=io.BytesIO(b'video')):
|
|
with playback.cached_source('http://127.0.0.1/source','private',8):
|
|
self.assertFalse(old.exists())
|
|
self.assertEqual((fresh/'source.mp4').read_bytes(),b'active')
|
|
self.assertEqual(unrelated.read_text(),'keep')
|
|
|
|
def test_local_converter_is_paced_and_has_no_credentials(self):
|
|
with patch('playback.shutil.which', return_value='ffmpeg'), patch('playback.os.access', return_value=False):
|
|
cmd=playback.command('/tmp/private/source.mp4', '', 300)
|
|
self.assertIn('-readrate', cmd)
|
|
self.assertNotIn('-headers', cmd)
|
|
self.assertIn('libx264', cmd)
|
|
self.assertEqual(cmd[cmd.index('-threads')+1], '1')
|
|
|
|
def test_sessions_are_bounded_before_consuming_a_slot(self):
|
|
for duration in (0, 301, float('nan')):
|
|
with self.assertRaises(playback.PlaybackError):
|
|
with playback.stream('unused', 'private', duration):
|
|
self.fail('unbounded playback accepted')
|
|
self.assertTrue(playback.SLOTS.acquire(blocking=False))
|
|
playback.SLOTS.release()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|