Python test_sessions.FakeTransport类代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Python中tftp.test.test_sessions.FakeTransport类的典型用法代码示例。如果您正苦于以下问题：Python FakeTransport类的具体用法？Python FakeTransport怎么用？Python FakeTransport使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了FakeTransport类的16个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: RemoteOriginReadOptionNegotiation

class RemoteOriginReadOptionNegotiation(unittest.TestCase):
    test_data = """line1
line2
anotherline"""
    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        with self.target.open('wb') as temp_fd:
            temp_fd.write(self.test_data)
        self.reader = DelayedReader(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.rs = RemoteOriginReadSession(('127.0.0.1', 65465), self.reader,
                                          options={'blksize':'9'}, _clock=self.clock)
        self.rs.transport = self.transport

    def test_option_normal(self):
        self.rs.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram({'blksize':'9'}).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)

        self.transport.clear()
        self.rs.datagramReceived(ACKDatagram(0).to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), DATADatagram(1, self.test_data[:9]).to_wire())

        self.addCleanup(self.rs.cancel)

    def test_option_timeout(self):
        self.rs.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram({'blksize':'9'}).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failUnless(self.transport.disconnecting)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:rvbad，项目名称:python-tx-tftp，代码行数:54，代码来源:test_bootstrap.py

示例2: BootstrapRemoteOriginRead

class BootstrapRemoteOriginRead(unittest.TestCase):
    test_data = """line1
line2
anotherline"""
    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        with self.target.open('wb') as temp_fd:
            temp_fd.write(self.test_data)
        self.reader = DelayedReader(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.rs = RemoteOriginReadSession(('127.0.0.1', 65465), self.reader, _clock=self.clock)
        self.rs.transport = self.transport

    @inlineCallbacks
    def test_invalid_tid(self):
        self.rs.startProtocol()
        data_datagram = DATADatagram(1, 'foobar')
        yield self.rs.datagramReceived(data_datagram, ('127.0.0.1', 11111))
        err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(err_dgram.errorcode, ERR_TID_UNKNOWN)
        self.addCleanup(self.rs.cancel)

    def test_remote_origin_read_bootstrap(self):
        # First datagram
        self.rs.session.block_size = 5
        self.rs.startProtocol()
        self.clock.pump((1,)*3)

        data_datagram_1 = DATADatagram(1, self.test_data[:5])

        self.assertEqual(self.transport.value(), data_datagram_1.to_wire())
        self.failIf(self.transport.disconnecting)

        # Normal exchange continues
        self.transport.clear()
        self.rs.datagramReceived(ACKDatagram(1).to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        data_datagram_2 = DATADatagram(2, self.test_data[5:10])
        self.assertEqual(self.transport.value(), data_datagram_2.to_wire())
        self.failIf(self.transport.disconnecting)
        self.addCleanup(self.rs.cancel)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:deepakhajare，项目名称:maas，代码行数:48，代码来源:test_bootstrap.py

示例3: BootstrapRemoteOriginWrite

class BootstrapRemoteOriginWrite(unittest.TestCase):

    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child("foo")
        self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=("127.0.0.1", self.port))
        self.ws = RemoteOriginWriteSession(("127.0.0.1", 65465), self.writer, _clock=self.clock)
        self.ws.transport = self.transport
        self.ws.startProtocol()

    @inlineCallbacks
    def test_invalid_tid(self):
        bad_tid_dgram = ACKDatagram(123)
        yield self.ws.datagramReceived(bad_tid_dgram.to_wire(), ("127.0.0.1", 1111))
        err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(err_dgram.errorcode, ERR_TID_UNKNOWN)
        self.addCleanup(self.ws.cancel)

    def test_remote_origin_write_bootstrap(self):
        # Initial ACK
        ack_datagram_0 = ACKDatagram(0)
        self.clock.advance(0.1)
        self.assertEqual(self.transport.value(), ack_datagram_0.to_wire())
        self.failIf(self.transport.disconnecting)

        # Normal exchange
        self.transport.clear()
        d = self.ws.datagramReceived(DATADatagram(1, "foobar").to_wire(), ("127.0.0.1", 65465))

        def cb(res):
            self.clock.advance(0.1)
            ack_datagram_1 = ACKDatagram(1)
            self.assertEqual(self.transport.value(), ack_datagram_1.to_wire())
            self.assertEqual(self.target.open("r").read(), "foobar")
            self.failIf(self.transport.disconnecting)
            self.addCleanup(self.ws.cancel)

        d.addCallback(cb)
        self.clock.advance(3)
        return d

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:chronidev，项目名称:python-tx-tftp，代码行数:47，代码来源:test_bootstrap.py

示例4: BootstrapLocalOriginWrite

class BootstrapLocalOriginWrite(unittest.TestCase):

    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child("foo")
        self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=("127.0.0.1", self.port))
        self.ws = LocalOriginWriteSession(("127.0.0.1", 65465), self.writer, _clock=self.clock)
        self.wd = MockHandshakeWatchdog(4, self.ws.timedOut, _clock=self.clock)
        self.ws.timeout_watchdog = self.wd
        self.ws.transport = self.transport

    def test_invalid_tid(self):
        self.ws.startProtocol()
        bad_tid_dgram = ACKDatagram(123)
        self.ws.datagramReceived(bad_tid_dgram.to_wire(), ("127.0.0.1", 1111))

        err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(err_dgram.errorcode, ERR_TID_UNKNOWN)
        self.addCleanup(self.ws.cancel)

    # test_invalid_tid.skip = 'Will go to another test case'

    def test_local_origin_write_session_handshake_timeout(self):
        self.ws.startProtocol()
        self.clock.advance(5)
        self.failIf(self.transport.value())
        self.failUnless(self.transport.disconnecting)

    def test_local_origin_write_session_handshake_success(self):
        self.ws.session.block_size = 6
        self.ws.startProtocol()
        self.clock.advance(1)
        data_datagram = DATADatagram(1, "foobar")
        self.ws.datagramReceived(data_datagram.to_wire(), ("127.0.0.1", 65465))
        self.clock.pump((1,) * 3)
        self.assertEqual(self.transport.value(), ACKDatagram(1).to_wire())
        self.failIf(self.transport.disconnecting)
        self.failIf(self.wd.active())
        self.addCleanup(self.ws.cancel)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:chronidev，项目名称:python-tx-tftp，代码行数:46，代码来源:test_bootstrap.py

示例5: LocalOriginReadOptionNegotiation

class LocalOriginReadOptionNegotiation(unittest.TestCase):
    test_data = """line1
line2
anotherline"""
    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        with self.target.open('wb') as temp_fd:
            temp_fd.write(self.test_data)
        self.reader = DelayedReader(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.rs = LocalOriginReadSession(('127.0.0.1', 65465), self.reader, _clock=self.clock)
        self.wd = MockHandshakeWatchdog(4, self.rs.timedOut, _clock=self.clock)
        self.rs.timeout_watchdog = self.wd
        self.rs.transport = self.transport

    def test_option_normal(self):
        self.rs.startProtocol()
        self.rs.datagramReceived(OACKDatagram({'blksize':'9'}).to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(0.1)
        self.assertEqual(self.rs.session.block_size, 9)
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), DATADatagram(1, self.test_data[:9]).to_wire())

        self.rs.datagramReceived(OACKDatagram({'blksize':'12'}).to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(0.1)
        self.assertEqual(self.rs.session.block_size, 9)

        self.transport.clear()
        self.rs.datagramReceived(ACKDatagram(1).to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), DATADatagram(2, self.test_data[9:18]).to_wire())

        self.addCleanup(self.rs.cancel)

    def test_local_origin_read_option_timeout(self):
        self.rs.startProtocol()
        self.clock.advance(5)
        self.failUnless(self.transport.disconnecting)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:deepakhajare，项目名称:maas，代码行数:45，代码来源:test_bootstrap.py

示例6: BootstrapLocalOriginRead

class BootstrapLocalOriginRead(unittest.TestCase):
    test_data = """line1
line2
anotherline"""
    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        with self.target.open('wb') as temp_fd:
            temp_fd.write(self.test_data)
        self.reader = DelayedReader(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.rs = LocalOriginReadSession(('127.0.0.1', 65465), self.reader, _clock=self.clock)
        self.wd = MockHandshakeWatchdog(4, self.rs.timedOut, _clock=self.clock)
        self.rs.timeout_watchdog = self.wd
        self.rs.transport = self.transport
        self.rs.startProtocol()

    def test_invalid_tid(self):
        data_datagram = DATADatagram(1, 'foobar')
        self.rs.datagramReceived(data_datagram, ('127.0.0.1', 11111))
        self.clock.advance(0.1)
        err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(err_dgram.errorcode, ERR_TID_UNKNOWN)
        self.addCleanup(self.rs.cancel)

    def test_local_origin_read_session_handshake_timeout(self):
        self.clock.advance(5)
        self.failIf(self.transport.value())
        self.failUnless(self.transport.disconnecting)

    def test_local_origin_read_session_handshake_success(self):
        self.clock.advance(1)
        ack_datagram = ACKDatagram(0)
        self.rs.datagramReceived(ack_datagram.to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(2)
        self.failUnless(self.transport.value())
        self.failIf(self.transport.disconnecting)
        self.failIf(self.wd.active())
        self.addCleanup(self.rs.cancel)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:deepakhajare，项目名称:maas，代码行数:45，代码来源:test_bootstrap.py

示例7: setUp

 def setUp(self):
     self.clock = Clock()
     self.tmp_dir_path = tempfile.mkdtemp()
     self.target = FilePath(self.tmp_dir_path).child('foo')
     self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
     self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
     self.ws = RemoteOriginWriteSession(('127.0.0.1', 65465), self.writer, _clock=self.clock)
     self.ws.transport = self.transport
     self.ws.startProtocol()

开发者ID:deepakhajare，项目名称:maas，代码行数:9，代码来源:test_bootstrap.py

示例8: BootstrapLocalOriginWrite

class BootstrapLocalOriginWrite(unittest.TestCase):

    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.temp_dir = FilePath(tempfile.mkdtemp()).asBytesMode()
        self.target = self.temp_dir.child(b'foo')
        self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.ws = LocalOriginWriteSession(('127.0.0.1', 65465), self.writer, _clock=self.clock)
        self.wd = timedCaller([5], None, self.ws.timedOut, self.clock)
        self.ws.timeout_watchdog = self.wd
        self.ws.transport = self.transport

    def test_invalid_tid(self):
        self.ws.startProtocol()
        bad_tid_dgram = ACKDatagram(123)
        self.ws.datagramReceived(bad_tid_dgram.to_wire(), ('127.0.0.1', 1111))

        err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(err_dgram.errorcode, ERR_TID_UNKNOWN)
        self.addCleanup(self.ws.cancel)
    #test_invalid_tid.skip = 'Will go to another test case'

    def test_local_origin_write_session_handshake_timeout(self):
        self.ws.startProtocol()
        self.clock.advance(5)
        self.assertFalse(self.transport.value())
        self.assertTrue(self.transport.disconnecting)

    def test_local_origin_write_session_handshake_success(self):
        self.ws.session.block_size = 6
        self.ws.startProtocol()
        self.clock.advance(1)
        data_datagram = DATADatagram(1, b'foobar')
        self.ws.datagramReceived(data_datagram.to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), ACKDatagram(1).to_wire())
        self.assertFalse(self.transport.disconnecting)
        self.addCleanup(self.ws.cancel)

    def tearDown(self):
        self.temp_dir.remove()

开发者ID:mont5piques，项目名称:python-tx-tftp，代码行数:44，代码来源:test_bootstrap.py

示例9: setUp

 def setUp(self):
     self.clock = Clock()
     self.temp_dir = FilePath(tempfile.mkdtemp()).asBytesMode()
     self.target = self.temp_dir.child(b'foo')
     self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
     self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
     self.ws = LocalOriginWriteSession(('127.0.0.1', 65465), self.writer, _clock=self.clock)
     self.wd = timedCaller([5], None, self.ws.timedOut, self.clock)
     self.ws.timeout_watchdog = self.wd
     self.ws.transport = self.transport

开发者ID:mont5piques，项目名称:python-tx-tftp，代码行数:10，代码来源:test_bootstrap.py

示例10: setUp

 def setUp(self):
     self.clock = Clock()
     self.temp_dir = FilePath(tempfile.mkdtemp()).asBytesMode()
     self.target = self.temp_dir.child(b'foo')
     self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
     self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
     self.ws = LocalOriginWriteSession(('127.0.0.1', 65465), self.writer,
                                       options={b'blksize':b'123'}, _clock=self.clock)
     self.wd = MockHandshakeWatchdog(4, self.ws.timedOut, _clock=self.clock)
     self.ws.timeout_watchdog = self.wd
     self.ws.transport = self.transport

开发者ID:aivins，项目名称:python-tx-tftp，代码行数:11，代码来源:test_bootstrap.py

示例11: setUp

 def setUp(self):
     self.clock = Clock()
     self.tmp_dir_path = tempfile.mkdtemp()
     self.target = FilePath(self.tmp_dir_path).child("foo")
     self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
     self.transport = FakeTransport(hostAddress=("127.0.0.1", self.port))
     self.options = OrderedDict()
     self.options["blksize"] = "9"
     self.options["tsize"] = "45"
     self.ws = RemoteOriginWriteSession(("127.0.0.1", 65465), self.writer, options=self.options, _clock=self.clock)
     self.ws.transport = self.transport

开发者ID:chronidev，项目名称:python-tx-tftp，代码行数:11，代码来源:test_bootstrap.py

示例12: BootstrapRemoteOriginRead

class BootstrapRemoteOriginRead(unittest.TestCase):
    test_data = b"""line1
line2
anotherline"""
    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.temp_dir = FilePath(tempfile.mkdtemp()).asBytesMode()
        self.target = self.temp_dir.child(b'foo')
        with self.target.open('wb') as temp_fd:
            temp_fd.write(self.test_data)
        self.reader = DelayedReader(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.rs = RemoteOriginReadSession(('127.0.0.1', 65465), self.reader, _clock=self.clock)
        self.rs.transport = self.transport

    @inlineCallbacks
    def test_invalid_tid(self):
        self.rs.startProtocol()
        data_datagram = DATADatagram(1, b'foobar')
        yield self.rs.datagramReceived(data_datagram, ('127.0.0.1', 11111))
        err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(err_dgram.errorcode, ERR_TID_UNKNOWN)
        self.addCleanup(self.rs.cancel)

    def test_remote_origin_read_bootstrap(self):
        # First datagram
        self.rs.session.block_size = 5
        self.rs.startProtocol()
        self.clock.pump((1,)*3)

        data_datagram_1 = DATADatagram(1, self.test_data[:5])

        self.assertEqual(self.transport.value(), data_datagram_1.to_wire())
        self.assertFalse(self.transport.disconnecting)

        # Normal exchange continues
        self.transport.clear()
        self.rs.datagramReceived(ACKDatagram(1).to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        data_datagram_2 = DATADatagram(2, self.test_data[5:10])
        self.assertEqual(self.transport.value(), data_datagram_2.to_wire())
        self.assertFalse(self.transport.disconnecting)
        self.addCleanup(self.rs.cancel)

    def test_remote_origin_read_session_not_started_rollover(self):
        # if a rollover is done, we reach blocknum 0 again. But this time
        # session is already started.
        # Here we test the case where rollover has not happened yet

        data_datagram = DATADatagram(1, self.test_data[:5])
        data_datagram.opcode = OP_ACK
        data_datagram.blocknum = 0
        self.rs.session.block_size = 5
        self.clock.pump((1,)*3)

        self.rs.session.transport = b"Fake"
        self.rs.session.started = False
        self.rs._datagramReceived(data_datagram)

        self.assertEqual(self.rs.session.started, True)
        self.assertEqual(self.rs.session.transport, self.rs.transport)

        self.addCleanup(self.rs.cancel)

    def test_remote_origin_read_session_started_rollover(self):
        # if a rollover is done, we reach blocknum 0 again. But this time
        # session is already started.
        # Here we test the case where rollover has already happened
        data_datagram = DATADatagram(1, self.test_data[:5])
        data_datagram.opcode = OP_ACK
        data_datagram.blocknum = 0
        self.rs.session.block_size = 5
        self.rs.startProtocol()
        self.clock.pump((1,)*3)

        self.assertEqual(
            self.rs._datagramReceived(data_datagram),
            self.rs.session.datagramReceived(data_datagram))
        self.addCleanup(self.rs.cancel)

    def tearDown(self):
        self.temp_dir.remove()

开发者ID:aivins，项目名称:python-tx-tftp，代码行数:84，代码来源:test_bootstrap.py

示例13: LocalOriginWriteOptionNegotiation

class LocalOriginWriteOptionNegotiation(unittest.TestCase):

    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.ws = LocalOriginWriteSession(('127.0.0.1', 65465), self.writer,
                                          options={'blksize':'123'}, _clock=self.clock)
        self.wd = MockHandshakeWatchdog(4, self.ws.timedOut, _clock=self.clock)
        self.ws.timeout_watchdog = self.wd
        self.ws.transport = self.transport


    def test_option_normal(self):
        self.ws.startProtocol()
        self.ws.datagramReceived(OACKDatagram({'blksize':'12'}).to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(0.1)
        self.assertEqual(self.ws.session.block_size, WriteSession.block_size)
        self.assertEqual(self.transport.value(), ACKDatagram(0).to_wire())

        self.transport.clear()
        self.ws.datagramReceived(OACKDatagram({'blksize':'9'}).to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(0.1)
        self.assertEqual(self.ws.session.block_size, WriteSession.block_size)
        self.assertEqual(self.transport.value(), ACKDatagram(0).to_wire())

        self.transport.clear()
        self.ws.datagramReceived(DATADatagram(1, 'foobarbaz').to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(3)
        self.failUnless(self.ws.session.started)
        self.clock.advance(0.1)
        self.assertEqual(self.ws.session.block_size, 9)
        self.assertEqual(self.transport.value(), ACKDatagram(1).to_wire())

        self.transport.clear()
        self.ws.datagramReceived(DATADatagram(2, 'asdfghjkl').to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(3)
        self.assertEqual(self.transport.value(), ACKDatagram(2).to_wire())
        self.writer.finish()
        self.assertEqual(self.writer.file_path.open('r').read(), 'foobarbazasdfghjkl')

        self.transport.clear()
        self.ws.datagramReceived(OACKDatagram({'blksize':'12'}).to_wire(), ('127.0.0.1', 65465))
        self.clock.advance(0.1)
        self.assertEqual(self.ws.session.block_size, 9)
        self.assertEqual(self.transport.value(), ACKDatagram(0).to_wire())

    def test_option_timeout(self):
        self.ws.startProtocol()
        self.clock.advance(5)
        self.failUnless(self.transport.disconnecting)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:deepakhajare，项目名称:maas，代码行数:58，代码来源:test_bootstrap.py

示例14: RemoteOriginWriteOptionNegotiation

class RemoteOriginWriteOptionNegotiation(unittest.TestCase):

    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.options = OrderedDict()
        self.options['blksize'] = '9'
        self.options['tsize'] = '45'
        self.ws = RemoteOriginWriteSession(
            ('127.0.0.1', 65465), self.writer, options=self.options,
            _clock=self.clock)
        self.ws.transport = self.transport

    def test_option_normal(self):
        self.ws.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram(self.options).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)

        self.transport.clear()
        self.ws.datagramReceived(DATADatagram(1, 'foobarbaz').to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), ACKDatagram(1).to_wire())
        self.assertEqual(self.ws.session.block_size, 9)

        self.transport.clear()
        self.ws.datagramReceived(DATADatagram(2, 'smthng').to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), ACKDatagram(2).to_wire())
        self.clock.pump((1,)*10)
        self.writer.finish()
        self.assertEqual(self.writer.file_path.open('r').read(), 'foobarbazsmthng')
        self.failUnless(self.transport.disconnecting)

    def test_option_timeout(self):
        self.ws.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram(self.options).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failUnless(self.transport.disconnecting)

    def test_option_tsize(self):
        # A tsize option sent as part of a write session is recorded.
        self.ws.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram(self.options).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.failIf(self.transport.disconnecting)
        self.assertIsInstance(self.ws.session, WriteSession)
        # Options are not applied to the WriteSession until the first DATA
        # datagram is received,
        self.assertIsNone(self.ws.session.tsize)
        self.ws.datagramReceived(
            DATADatagram(1, 'foobarbaz').to_wire(), ('127.0.0.1', 65465))
        # The tsize option has been applied to the WriteSession.
        self.assertEqual(45, self.ws.session.tsize)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:deepakhajare，项目名称:maas，代码行数:78，代码来源:test_bootstrap.py

示例15: RemoteOriginReadOptionNegotiation

class RemoteOriginReadOptionNegotiation(unittest.TestCase):
    test_data = """line1
line2
anotherline"""
    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        with self.target.open('wb') as temp_fd:
            temp_fd.write(self.test_data)
        self.reader = DelayedReader(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.options = OrderedDict()
        self.options['blksize'] = '9'
        self.options['tsize'] = '34'
        self.rs = RemoteOriginReadSession(('127.0.0.1', 65465), self.reader,
                                          options=self.options, _clock=self.clock)
        self.rs.transport = self.transport

    def test_option_normal(self):
        self.rs.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram(self.options).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)

        self.transport.clear()
        self.rs.datagramReceived(ACKDatagram(0).to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), DATADatagram(1, self.test_data[:9]).to_wire())

        self.addCleanup(self.rs.cancel)

    def test_option_timeout(self):
        self.rs.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram(self.options).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failUnless(self.transport.disconnecting)

    def test_option_tsize(self):
        # A tsize option of 0 sent as part of a read session prompts a tsize
        # response with the actual size of the file.
        self.options['tsize'] = '0'
        self.rs.startProtocol()
        self.clock.advance(0.1)
        self.transport.clear()
        self.clock.advance(3)
        # The response contains the size of the test data.
        self.options['tsize'] = str(len(self.test_data))
        oack_datagram = OACKDatagram(self.options).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:deepakhajare，项目名称:maas，代码行数:70，代码来源:test_bootstrap.py

示例16: RemoteOriginWriteOptionNegotiation

class RemoteOriginWriteOptionNegotiation(unittest.TestCase):

    port = 65466

    def setUp(self):
        self.clock = Clock()
        self.tmp_dir_path = tempfile.mkdtemp()
        self.target = FilePath(self.tmp_dir_path).child('foo')
        self.writer = DelayedWriter(self.target, _clock=self.clock, delay=2)
        self.transport = FakeTransport(hostAddress=('127.0.0.1', self.port))
        self.ws = RemoteOriginWriteSession(('127.0.0.1', 65465), self.writer,
                                           options={'blksize':'9'}, _clock=self.clock)
        self.ws.transport = self.transport

    def test_option_normal(self):
        self.ws.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram({'blksize':'9'}).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)

        self.transport.clear()
        self.ws.datagramReceived(DATADatagram(1, 'foobarbaz').to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), ACKDatagram(1).to_wire())
        self.assertEqual(self.ws.session.block_size, 9)

        self.transport.clear()
        self.ws.datagramReceived(DATADatagram(2, 'smthng').to_wire(), ('127.0.0.1', 65465))
        self.clock.pump((1,)*3)
        self.assertEqual(self.transport.value(), ACKDatagram(2).to_wire())
        self.clock.pump((1,)*10)
        self.writer.finish()
        self.assertEqual(self.writer.file_path.open('r').read(), 'foobarbazsmthng')
        self.failUnless(self.transport.disconnecting)

    def test_option_timeout(self):
        self.ws.startProtocol()
        self.clock.advance(0.1)
        oack_datagram = OACKDatagram({'blksize':'9'}).to_wire()
        self.assertEqual(self.transport.value(), oack_datagram)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(3)
        self.assertEqual(self.transport.value(), oack_datagram * 2)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failIf(self.transport.disconnecting)

        self.clock.advance(2)
        self.assertEqual(self.transport.value(), oack_datagram * 3)
        self.failUnless(self.transport.disconnecting)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir_path)

开发者ID:rvbad，项目名称:python-tx-tftp，代码行数:58，代码来源:test_bootstrap.py

注：本文中的tftp.test.test_sessions.FakeTransport类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python tfx.pose函数代码示例发布时间：2022-05-27

Python protocol.TFTP类代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13769|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：9571|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4064|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4039|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3817|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3499|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3022|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2637|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2624|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2290|2022-01-22

客服电话

电子邮件

Python test_sessions.FakeTransport类代码示例

示例1: RemoteOriginReadOptionNegotiation

示例2: BootstrapRemoteOriginRead

示例3: BootstrapRemoteOriginWrite

示例4: BootstrapLocalOriginWrite

示例5: LocalOriginReadOptionNegotiation

示例6: BootstrapLocalOriginRead

示例7: setUp

示例8: BootstrapLocalOriginWrite

示例9: setUp

示例10: setUp

示例11: setUp

示例12: BootstrapRemoteOriginRead

示例13: LocalOriginWriteOptionNegotiation

示例14: RemoteOriginWriteOptionNegotiation

示例15: RemoteOriginReadOptionNegotiation

示例16: RemoteOriginWriteOptionNegotiation

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053