1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
import sys
import tempfile
import os.path
import datetime
import streamlink.cache
from shutil import rmtree
import time
import unittest
from tests.mock import patch
is_py2 = (sys.version_info[0] == 2)
class TestCache(unittest.TestCase):
def setUp(self):
self.tmp_dir = tempfile.mkdtemp("streamlink-test")
streamlink.cache.cache_dir = self.tmp_dir
self.cache = streamlink.cache.Cache("cache.json")
def tearDown(self):
rmtree(self.tmp_dir)
def test_get_no_file(self):
self.assertEqual(self.cache.get("missing-value"), None)
self.assertEqual(self.cache.get("missing-value", default="default"), "default")
def test_put_get(self):
self.cache.set("value", 1)
self.assertEqual(self.cache.get("value"), 1)
def test_put_get_prefix(self):
self.cache.key_prefix = "test"
self.cache.set("value", 1)
self.assertEqual(self.cache.get("value"), 1)
def test_key_prefix(self):
self.cache.key_prefix = "test"
self.cache.set("value", 1)
self.assertTrue("test:value" in self.cache._cache)
self.assertEqual(1, self.cache._cache["test:value"]["value"])
@patch('os.path.exists', return_value=True)
def test_load_fail(self, exists_mock):
if is_py2:
patch('__builtin__.open', side_effect=IOError)
else:
patch('streamlink.cache.open', side_effect=IOError)
self.cache._load()
self.assertEqual({}, self.cache._cache)
def test_expired(self):
self.cache.set("value", 10, expires=-20)
self.assertEqual(None, self.cache.get("value"))
def test_expired_at(self):
self.cache.set("value", 10, expires_at=datetime.datetime.now() - datetime.timedelta(seconds=20))
self.assertEqual(None, self.cache.get("value"))
def test_not_expired(self):
self.cache.set("value", 10, expires=20)
self.assertEqual(10, self.cache.get("value"))
def test_expired_at(self):
self.cache.set("value", 10, expires_at=datetime.datetime.now() + datetime.timedelta(seconds=20))
self.assertEqual(10, self.cache.get("value"))
def test_create_directory(self):
try:
streamlink.cache.cache_dir = os.path.join(tempfile.gettempdir(), "streamlink-test")
cache = streamlink.cache.Cache("cache.json")
self.assertFalse(os.path.exists(cache.filename))
cache.set("value", 10)
self.assertTrue(os.path.exists(cache.filename))
finally:
rmtree(streamlink.cache.cache_dir, ignore_errors=True)
@patch('os.makedirs', side_effect=OSError)
def test_create_directory_fail(self, makedirs):
try:
streamlink.cache.cache_dir = os.path.join(tempfile.gettempdir(), "streamlink-test")
cache = streamlink.cache.Cache("cache.json")
self.assertFalse(os.path.exists(cache.filename))
cache.set("value", 10)
self.assertFalse(os.path.exists(cache.filename))
finally:
rmtree(streamlink.cache.cache_dir, ignore_errors=True)
def test_get_all(self):
self.cache.set("test1", 1)
self.cache.set("test2", 2)
self.assertDictEqual(
{"test1": 1, "test2": 2},
self.cache.get_all())
def test_get_all_prefix(self):
self.cache.set("test1", 1)
self.cache.set("test2", 2)
self.cache.key_prefix = "test"
self.cache.set("test3", 3)
self.cache.set("test4", 4)
self.assertDictEqual(
{"test3": 3, "test4": 4},
self.cache.get_all())
def test_get_all_prune(self):
self.cache.set("test1", 1)
self.cache.set("test2", 2, -1)
self.assertDictEqual(
{"test1": 1},
self.cache.get_all())
|
