Server IP : 85.214.239.14 / Your IP : 3.145.72.55 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /proc/2/root/usr/lib/python3/dist-packages/supervisor/tests/ |
Upload File : |
import base64 import os import stat import sys import socket import tempfile import unittest from supervisor.compat import as_bytes from supervisor.compat import as_string from supervisor.compat import sha1 from supervisor.tests.base import DummySupervisor from supervisor.tests.base import PopulatedDummySupervisor from supervisor.tests.base import DummyRPCInterfaceFactory from supervisor.tests.base import DummyPConfig from supervisor.tests.base import DummyOptions from supervisor.tests.base import DummyRequest from supervisor.tests.base import DummyLogger from supervisor.http import NOT_DONE_YET class HandlerTests: def _makeOne(self, supervisord): return self._getTargetClass()(supervisord) def test_match(self): class FakeRequest: def __init__(self, uri): self.uri = uri supervisor = DummySupervisor() handler = self._makeOne(supervisor) self.assertEqual(handler.match(FakeRequest(handler.path)), True) class LogtailHandlerTests(HandlerTests, unittest.TestCase): def _getTargetClass(self): from supervisor.http import logtail_handler return logtail_handler def test_handle_request_stdout_logfile_none(self): options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig) handler = self._makeOne(supervisord) request = DummyRequest('/logtail/process1', None, None, None) handler.handle_request(request) self.assertEqual(request._error, 404) def test_handle_request_stdout_logfile_missing(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', 'foo', 'it/is/missing') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) handler = self._makeOne(supervisord) request = DummyRequest('/logtail/foo', None, None, None) handler.handle_request(request) self.assertEqual(request._error, 404) def test_handle_request(self): with tempfile.NamedTemporaryFile() as f: t = f.name options = DummyOptions() pconfig = DummyPConfig(options, 'foo', 'foo', stdout_logfile=t) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) handler = self._makeOne(supervisord) request = DummyRequest('/logtail/foo', None, None, None) handler.handle_request(request) self.assertEqual(request._error, None) from supervisor.medusa import http_date self.assertEqual(request.headers['Last-Modified'], http_date.build_http_date(os.stat(t)[stat.ST_MTIME])) self.assertEqual(request.headers['Content-Type'], 'text/plain;charset=utf-8') self.assertEqual(request.headers['X-Accel-Buffering'], 'no') self.assertEqual(len(request.producers), 1) self.assertEqual(request._done, True) class MainLogTailHandlerTests(HandlerTests, unittest.TestCase): def _getTargetClass(self): from supervisor.http import mainlogtail_handler return mainlogtail_handler def test_handle_request_stdout_logfile_none(self): supervisor = DummySupervisor() handler = self._makeOne(supervisor) request = DummyRequest('/mainlogtail', None, None, None) handler.handle_request(request) self.assertEqual(request._error, 404) def test_handle_request_stdout_logfile_missing(self): supervisor = DummySupervisor() supervisor.options.logfile = '/not/there' request = DummyRequest('/mainlogtail', None, None, None) handler = self._makeOne(supervisor) handler.handle_request(request) self.assertEqual(request._error, 404) def test_handle_request(self): supervisor = DummySupervisor() with tempfile.NamedTemporaryFile() as f: t = f.name supervisor.options.logfile = t handler = self._makeOne(supervisor) request = DummyRequest('/mainlogtail', None, None, None) handler.handle_request(request) self.assertEqual(request._error, None) from supervisor.medusa import http_date self.assertEqual(request.headers['Last-Modified'], http_date.build_http_date(os.stat(t)[stat.ST_MTIME])) self.assertEqual(request.headers['Content-Type'], 'text/plain;charset=utf-8') self.assertEqual(len(request.producers), 1) self.assertEqual(request._done, True) class TailFProducerTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import tail_f_producer return tail_f_producer def _makeOne(self, request, filename, head): return self._getTargetClass()(request, filename, head) def test_handle_more(self): request = DummyRequest('/logtail/foo', None, None, None) from supervisor import http f = tempfile.NamedTemporaryFile() f.write(b'a' * 80) f.flush() producer = self._makeOne(request, f.name, 80) result = producer.more() self.assertEqual(result, b'a' * 80) f.write(as_bytes(b'w' * 100)) f.flush() result = producer.more() self.assertEqual(result, b'w' * 100) result = producer.more() self.assertEqual(result, http.NOT_DONE_YET) f.truncate(0) f.flush() result = producer.more() self.assertEqual(result, '==> File truncated <==\n') def test_handle_more_fd_closed(self): request = DummyRequest('/logtail/foo', None, None, None) with tempfile.NamedTemporaryFile() as f: f.write(as_bytes('a' * 80)) f.flush() producer = self._makeOne(request, f.name, 80) producer.file.close() result = producer.more() self.assertEqual(result, producer.more()) def test_handle_more_follow_file_recreated(self): request = DummyRequest('/logtail/foo', None, None, None) f = tempfile.NamedTemporaryFile() f.write(as_bytes('a' * 80)) f.flush() producer = self._makeOne(request, f.name, 80) result = producer.more() self.assertEqual(result, b'a' * 80) f.close() f2 = open(f.name, 'wb') try: f2.write(as_bytes(b'b' * 80)) f2.close() result = producer.more() finally: os.unlink(f2.name) self.assertEqual(result, b'b' * 80) def test_handle_more_follow_file_gone(self): request = DummyRequest('/logtail/foo', None, None, None) with tempfile.NamedTemporaryFile(delete=False) as f: filename = f.name f.write(b'a' * 80) try: producer = self._makeOne(request, f.name, 80) finally: os.unlink(f.name) result = producer.more() self.assertEqual(result, b'a' * 80) with open(filename, 'wb') as f: f.write(as_bytes(b'b' * 80)) try: result = producer.more() # should open in new file self.assertEqual(result, b'b' * 80) finally: os.unlink(f.name) class DeferringChunkedProducerTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import deferring_chunked_producer return deferring_chunked_producer def _makeOne(self, producer, footers=None): return self._getTargetClass()(producer, footers) def test_more_not_done_yet(self): wrapped = DummyProducer(NOT_DONE_YET) producer = self._makeOne(wrapped) self.assertEqual(producer.more(), NOT_DONE_YET) def test_more_string(self): wrapped = DummyProducer(b'hello') producer = self._makeOne(wrapped) self.assertEqual(producer.more(), b'5\r\nhello\r\n') def test_more_nodata(self): wrapped = DummyProducer() producer = self._makeOne(wrapped, footers=[b'a', b'b']) self.assertEqual(producer.more(), b'0\r\na\r\nb\r\n\r\n') def test_more_nodata_footers(self): wrapped = DummyProducer(b'') producer = self._makeOne(wrapped, footers=[b'a', b'b']) self.assertEqual(producer.more(), b'0\r\na\r\nb\r\n\r\n') def test_more_nodata_nofooters(self): wrapped = DummyProducer(b'') producer = self._makeOne(wrapped) self.assertEqual(producer.more(), b'0\r\n\r\n') def test_more_noproducer(self): producer = self._makeOne(None) self.assertEqual(producer.more(), b'') class DeferringCompositeProducerTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import deferring_composite_producer return deferring_composite_producer def _makeOne(self, producers): return self._getTargetClass()(producers) def test_more_not_done_yet(self): wrapped = DummyProducer(NOT_DONE_YET) producer = self._makeOne([wrapped]) self.assertEqual(producer.more(), NOT_DONE_YET) def test_more_string(self): wrapped1 = DummyProducer('hello') wrapped2 = DummyProducer('goodbye') producer = self._makeOne([wrapped1, wrapped2]) self.assertEqual(producer.more(), 'hello') self.assertEqual(producer.more(), 'goodbye') self.assertEqual(producer.more(), b'') def test_more_nodata(self): wrapped = DummyProducer() producer = self._makeOne([wrapped]) self.assertEqual(producer.more(), b'') class DeferringGlobbingProducerTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import deferring_globbing_producer return deferring_globbing_producer def _makeOne(self, producer, buffer_size=1<<16): return self._getTargetClass()(producer, buffer_size) def test_more_not_done_yet(self): wrapped = DummyProducer(NOT_DONE_YET) producer = self._makeOne(wrapped) self.assertEqual(producer.more(), NOT_DONE_YET) def test_more_string(self): wrapped = DummyProducer('hello', 'there', 'guy') producer = self._makeOne(wrapped, buffer_size=1) self.assertEqual(producer.more(), b'hello') wrapped = DummyProducer('hello', 'there', 'guy') producer = self._makeOne(wrapped, buffer_size=50) self.assertEqual(producer.more(), b'hellothereguy') def test_more_nodata(self): wrapped = DummyProducer() producer = self._makeOne(wrapped) self.assertEqual(producer.more(), b'') class DeferringHookedProducerTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import deferring_hooked_producer return deferring_hooked_producer def _makeOne(self, producer, function): return self._getTargetClass()(producer, function) def test_more_not_done_yet(self): wrapped = DummyProducer(NOT_DONE_YET) producer = self._makeOne(wrapped, None) self.assertEqual(producer.more(), NOT_DONE_YET) def test_more_string(self): wrapped = DummyProducer('hello') L = [] def callback(bytes): L.append(bytes) producer = self._makeOne(wrapped, callback) self.assertEqual(producer.more(), 'hello') self.assertEqual(L, []) producer.more() self.assertEqual(L, [5]) def test_more_nodata(self): wrapped = DummyProducer() L = [] def callback(bytes): L.append(bytes) producer = self._makeOne(wrapped, callback) self.assertEqual(producer.more(), b'') self.assertEqual(L, [0]) def test_more_noproducer(self): producer = self._makeOne(None, None) self.assertEqual(producer.more(), b'') class DeferringHttpRequestTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import deferring_http_request return deferring_http_request def _makeOne( self, channel=None, req='GET / HTTP/1.0', command='GET', uri='/', version='1.0', header=(), ): return self._getTargetClass()( channel, req, command, uri, version, header ) def _makeChannel(self): class Channel: closed = False def close_when_done(self): self.closed = True def push_with_producer(self, producer): self.producer = producer return Channel() def test_done_http_10_nokeepalive(self): channel = self._makeChannel() inst = self._makeOne(channel=channel, version='1.0') inst.done() self.assertTrue(channel.closed) def test_done_http_10_keepalive_no_content_length(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version='1.0', header=['Connection: Keep-Alive'], ) inst.done() self.assertTrue(channel.closed) def test_done_http_10_keepalive_and_content_length(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version='1.0', header=['Connection: Keep-Alive'], ) inst.reply_headers['Content-Length'] = 1 inst.done() self.assertEqual(inst['Connection'], 'Keep-Alive') self.assertFalse(channel.closed) def test_done_http_11_connection_close(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version='1.1', header=['Connection: close'] ) inst.done() self.assertTrue(channel.closed) def test_done_http_11_unknown_transfer_encoding(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version='1.1', ) inst.reply_headers['Transfer-Encoding'] = 'notchunked' inst.done() self.assertTrue(channel.closed) def test_done_http_11_chunked_transfer_encoding(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version='1.1', ) inst.reply_headers['Transfer-Encoding'] = 'chunked' inst.done() self.assertFalse(channel.closed) def test_done_http_11_use_chunked(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version='1.1', ) inst.use_chunked = True inst.done() self.assertTrue('Transfer-Encoding' in inst) self.assertFalse(channel.closed) def test_done_http_11_wo_content_length_no_te_no_use_chunked_close(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version='1.1', ) inst.use_chunked = False inst.done() self.assertTrue(channel.closed) def test_done_http_09(self): channel = self._makeChannel() inst = self._makeOne( channel=channel, version=None, ) inst.done() self.assertTrue(channel.closed) class DeferringHttpChannelTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import deferring_http_channel return deferring_http_channel def _makeOne(self): return self._getTargetClass()( server=None, conn=None, addr=None ) def test_defaults_delay_and_last_writable_check_time(self): channel = self._makeOne() self.assertEqual(channel.delay, 0) self.assertEqual(channel.last_writable_check, 0) def test_writable_with_delay_is_False_if_elapsed_lt_delay(self): channel = self._makeOne() channel.delay = 2 channel.last_writable_check = _NOW later = _NOW + 1 self.assertFalse(channel.writable(now=later)) self.assertEqual(channel.last_writable_check, _NOW) def test_writable_with_delay_is_False_if_elapsed_eq_delay(self): channel = self._makeOne() channel.delay = 2 channel.last_writable_check = _NOW later = _NOW + channel.delay self.assertFalse(channel.writable(now=later)) self.assertEqual(channel.last_writable_check, _NOW) def test_writable_with_delay_is_True_if_elapsed_gt_delay(self): channel = self._makeOne() channel.delay = 2 channel.last_writable_check = _NOW later = _NOW + channel.delay + 0.1 self.assertTrue(channel.writable(now=later)) self.assertEqual(channel.last_writable_check, later) def test_writable_with_delay_is_True_if_system_time_goes_backwards(self): channel = self._makeOne() channel.delay = 2 channel.last_writable_check = _NOW later = _NOW - 3600 # last check was in the future self.assertTrue(channel.writable(now=later)) self.assertEqual(channel.last_writable_check, later) _NOW = 1470085990 class EncryptedDictionaryAuthorizedTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import encrypted_dictionary_authorizer return encrypted_dictionary_authorizer def _makeOne(self, dict): return self._getTargetClass()(dict) def test_authorize_baduser(self): authorizer = self._makeOne({}) self.assertFalse(authorizer.authorize(('foo', 'bar'))) def test_authorize_gooduser_badpassword(self): authorizer = self._makeOne({'foo':'password'}) self.assertFalse(authorizer.authorize(('foo', 'bar'))) def test_authorize_gooduser_goodpassword(self): authorizer = self._makeOne({'foo':'password'}) self.assertTrue(authorizer.authorize(('foo', 'password'))) def test_authorize_gooduser_goodpassword_with_colon(self): authorizer = self._makeOne({'foo':'pass:word'}) self.assertTrue(authorizer.authorize(('foo', 'pass:word'))) def test_authorize_gooduser_badpassword_sha(self): password = '{SHA}' + sha1(as_bytes('password')).hexdigest() authorizer = self._makeOne({'foo':password}) self.assertFalse(authorizer.authorize(('foo', 'bar'))) def test_authorize_gooduser_goodpassword_sha(self): password = '{SHA}' + sha1(as_bytes('password')).hexdigest() authorizer = self._makeOne({'foo':password}) self.assertTrue(authorizer.authorize(('foo', 'password'))) class SupervisorAuthHandlerTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import supervisor_auth_handler return supervisor_auth_handler def _makeOne(self, dict, handler): return self._getTargetClass()(dict, handler) def test_ctor(self): handler = self._makeOne({'a':1}, None) from supervisor.http import encrypted_dictionary_authorizer self.assertEqual(handler.authorizer.__class__, encrypted_dictionary_authorizer) def test_handle_request_authorizes_good_credentials(self): request = DummyRequest('/logtail/process1', None, None, None) encoded = base64.b64encode(as_bytes("user:password")) request.header = ["Authorization: Basic %s" % as_string(encoded)] handler = DummyHandler() auth_handler = self._makeOne({'user':'password'}, handler) auth_handler.handle_request(request) self.assertTrue(handler.handled_request) def test_handle_request_authorizes_good_password_with_colon(self): request = DummyRequest('/logtail/process1', None, None, None) # password contains colon encoded = base64.b64encode(as_bytes("user:pass:word")) request.header = ["Authorization: Basic %s" % as_string(encoded)] handler = DummyHandler() auth_handler = self._makeOne({'user':'pass:word'}, handler) auth_handler.handle_request(request) self.assertTrue(handler.handled_request) def test_handle_request_does_not_authorize_bad_credentials(self): request = DummyRequest('/logtail/process1', None, None, None) encoded = base64.b64encode(as_bytes("wrong:wrong")) request.header = ["Authorization: Basic %s" % as_string(encoded)] handler = DummyHandler() auth_handler = self._makeOne({'user':'password'}, handler) auth_handler.handle_request(request) self.assertFalse(handler.handled_request) class LogWrapperTests(unittest.TestCase): def _getTargetClass(self): from supervisor.http import LogWrapper return LogWrapper def _makeOne(self, logger): return self._getTargetClass()(logger) def test_strips_trailing_newlines_from_msgs(self): logger = DummyLogger() log_wrapper = self._makeOne(logger) log_wrapper.log("foo\n") logdata = logger.data self.assertEqual(len(logdata), 1) self.assertEqual(logdata[0], "foo") def test_logs_msgs_with_error_at_error_level(self): logger = DummyLogger() log_wrapper = self._makeOne(logger) errors = [] logger.error = errors.append log_wrapper.log("Server Error") self.assertEqual(len(errors), 1) self.assertEqual(errors[0], "Server Error") def test_logs_other_messages_at_trace_level(self): logger = DummyLogger() log_wrapper = self._makeOne(logger) traces = [] logger.trace = traces.append log_wrapper.log("GET /") self.assertEqual(len(traces), 1) self.assertEqual(traces[0], "GET /") class TopLevelFunctionTests(unittest.TestCase): def _make_http_servers(self, sconfigs): options = DummyOptions() options.server_configs = sconfigs options.rpcinterface_factories = [('dummy',DummyRPCInterfaceFactory,{})] supervisord = DummySupervisor() from supervisor.http import make_http_servers servers = make_http_servers(options, supervisord) try: for config, s in servers: s.close() socketfile = config.get('file') if socketfile is not None: os.unlink(socketfile) finally: from supervisor.medusa.asyncore_25 import socket_map socket_map.clear() return servers def test_make_http_servers_socket_type_error(self): config = {'family':999, 'host':'localhost', 'port':17735, 'username':None, 'password':None, 'section':'inet_http_server'} try: self._make_http_servers([config]) self.fail('nothing raised') except ValueError as exc: self.assertEqual(exc.args[0], 'Cannot determine socket type 999') def test_make_http_servers_noauth(self): with tempfile.NamedTemporaryFile(delete=True) as f: socketfile = f.name self.assertFalse(os.path.exists(socketfile)) inet = {'family':socket.AF_INET, 'host':'localhost', 'port':17735, 'username':None, 'password':None, 'section':'inet_http_server'} unix = {'family':socket.AF_UNIX, 'file':socketfile, 'chmod':0o700, 'chown':(-1, -1), 'username':None, 'password':None, 'section':'unix_http_server'} servers = self._make_http_servers([inet, unix]) self.assertEqual(len(servers), 2) inetdata = servers[0] self.assertEqual(inetdata[0], inet) server = inetdata[1] idents = [ 'Supervisor XML-RPC Handler', 'Logtail HTTP Request Handler', 'Main Logtail HTTP Request Handler', 'Supervisor Web UI HTTP Request Handler', 'Default HTTP Request Handler' ] self.assertEqual([x.IDENT for x in server.handlers], idents) unixdata = servers[1] self.assertEqual(unixdata[0], unix) server = unixdata[1] self.assertEqual([x.IDENT for x in server.handlers], idents) def test_make_http_servers_withauth(self): with tempfile.NamedTemporaryFile(delete=True) as f: socketfile = f.name self.assertFalse(os.path.exists(socketfile)) inet = {'family':socket.AF_INET, 'host':'localhost', 'port':17736, 'username':'username', 'password':'password', 'section':'inet_http_server'} unix = {'family':socket.AF_UNIX, 'file':socketfile, 'chmod':0o700, 'chown':(-1, -1), 'username':'username', 'password':'password', 'section':'unix_http_server'} servers = self._make_http_servers([inet, unix]) self.assertEqual(len(servers), 2) from supervisor.http import supervisor_auth_handler for config, server in servers: for handler in server.handlers: self.assertTrue(isinstance(handler, supervisor_auth_handler), handler) class DummyHandler: def __init__(self): self.handled_request = False def handle_request(self, request): self.handled_request = True class DummyProducer: def __init__(self, *data): self.data = list(data) def more(self): if self.data: return self.data.pop(0) else: return b'' def test_suite(): return unittest.findTestCases(sys.modules[__name__]) if __name__ == '__main__': unittest.main(defaultTest='test_suite')