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
120
121
122
|
# This file is part of Supysonic.
# Supysonic is a Python implementation of the Subsonic server API.
#
# Copyright (C) 2017-2020 Alban 'spl0k' Féron
#
# Distributed under terms of the GNU AGPLv3 license.
import inspect
import os
import os.path
import shutil
import sys
import tempfile
import unittest
from pony.orm import db_session
from supysonic.db import init_database, release_database
from supysonic.config import DefaultConfig
from supysonic.managers.user import UserManager
from supysonic.web import create_application
class TestConfig(DefaultConfig):
TESTING = True
LOGGER_HANDLER_POLICY = "never"
MIMETYPES = {"mp3": "audio/mpeg", "weirdextension": "application/octet-stream"}
TRANSCODING = {
"transcoder_mp3_mp3": "echo -n %srcpath %outrate",
"transcoder_mp3_rnd": "dd if=/dev/urandom bs=1kB count=52 status=none",
"decoder_mp3": "echo -n Pushing out some mp3 data...",
"encoder_cat": "cat -",
"encoder_md5": "md5sum",
}
def __init__(self, with_webui, with_api):
super().__init__()
for cls in reversed(inspect.getmro(self.__class__)):
for attr, value in cls.__dict__.items():
if attr.startswith("_") or attr != attr.upper():
continue
if isinstance(value, dict):
setattr(self, attr, value.copy())
else:
setattr(self, attr, value)
self.WEBAPP.update({"mount_webui": with_webui, "mount_api": with_api})
with tempfile.NamedTemporaryFile() as tf:
if sys.platform == "win32":
self.DAEMON["socket"] = "\\\\.\\pipe\\" + os.path.basename(tf.name)
else:
self.DAEMON["socket"] = tf.name
class MockResponse:
def __init__(self, response):
self.__status_code = response.status_code
self.__data = response.get_data(as_text=True)
self.__mimetype = response.mimetype
@property
def status_code(self):
return self.__status_code
@property
def data(self):
return self.__data
@property
def mimetype(self):
return self.__mimetype
def patch_method(f):
original = f
def patched(*args, **kwargs):
rv = original(*args, **kwargs)
return MockResponse(rv)
return patched
class TestBase(unittest.TestCase):
__with_webui__ = False
__with_api__ = False
def setUp(self):
self.__db = tempfile.mkstemp()
self.__dir = tempfile.mkdtemp()
self.config = TestConfig(self.__with_webui__, self.__with_api__)
self.config.BASE["database_uri"] = "sqlite:///" + self.__db[1]
self.config.WEBAPP["cache_dir"] = self.__dir
init_database(self.config.BASE["database_uri"])
release_database()
self.__app = create_application(self.config)
self.client = self.__app.test_client()
with db_session:
UserManager.add("alice", "Alic3", admin=True)
UserManager.add("bob", "B0b")
def _patch_client(self):
self.client.get = patch_method(self.client.get)
self.client.post = patch_method(self.client.post)
def app_context(self, *args, **kwargs):
return self.__app.app_context(*args, **kwargs)
def request_context(self, *args, **kwargs):
return self.__app.test_request_context(*args, **kwargs)
def tearDown(self):
release_database()
shutil.rmtree(self.__dir)
os.close(self.__db[0])
os.remove(self.__db[1])
|
