ucvl-home-vision/test_playback_cache.py

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_can_buffer_ahead_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.assertNotIn('-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()