• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python datagram.split_opcode函数代码示例

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

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



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

示例1: test_access_violation

    def test_access_violation(self):
        tftp = TFTP(BackendFactory(AccessViolation("No!")), _clock=self.clock)
        tftp.transport = self.transport
        wrq_datagram = WRQDatagram('foobar', 'netascii', {})
        tftp.datagramReceived(wrq_datagram.to_wire(), ('127.0.0.1', 1111))
        error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(error_datagram.errorcode, ERR_ACCESS_VIOLATION)

        self.transport.clear()
        rrq_datagram = RRQDatagram('foobar', 'octet', {})
        tftp.datagramReceived(rrq_datagram.to_wire(), ('127.0.0.1', 1111))
        error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(error_datagram.errorcode, ERR_ACCESS_VIOLATION)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:13,代码来源:test_protocol.py


示例2: test_generic_backend_error

    def test_generic_backend_error(self):
        tftp = TFTP(BackendFactory(BackendError("A backend that couldn't")), _clock=self.clock)
        tftp.transport = self.transport
        rrq_datagram = RRQDatagram('foobar', 'netascii', {})
        tftp.datagramReceived(rrq_datagram.to_wire(), ('127.0.0.1', 1111))
        error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(error_datagram.errorcode, ERR_NOT_DEFINED)

        self.transport.clear()
        rrq_datagram = RRQDatagram('foobar', 'octet', {})
        tftp.datagramReceived(rrq_datagram.to_wire(), ('127.0.0.1', 1111))
        error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(error_datagram.errorcode, ERR_NOT_DEFINED)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:13,代码来源:test_protocol.py


示例3: test_unsupported

    def test_unsupported(self):
        tftp = TFTP(BackendFactory(Unsupported("I don't support you")), _clock=self.clock)
        tftp.transport = self.transport
        wrq_datagram = WRQDatagram('foobar', 'netascii', {})
        tftp.datagramReceived(wrq_datagram.to_wire(), ('127.0.0.1', 1111))
        error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(error_datagram.errorcode, ERR_ILLEGAL_OP)

        self.transport.clear()
        rrq_datagram = RRQDatagram('foobar', 'octet', {})
        tftp.datagramReceived(rrq_datagram.to_wire(), ('127.0.0.1', 1111))
        error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
        self.assertEqual(error_datagram.errorcode, ERR_ILLEGAL_OP)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:13,代码来源:test_protocol.py


示例4: test_file_exists

 def test_file_exists(self):
     tftp = TFTP(BackendFactory(FileExists("Already have one")), _clock=self.clock)
     tftp.transport = self.transport
     wrq_datagram = WRQDatagram('foobar', 'netascii', {})
     tftp.datagramReceived(wrq_datagram.to_wire(), ('127.0.0.1', 1111))
     error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assertEqual(error_datagram.errorcode, ERR_FILE_EXISTS)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:7,代码来源:test_protocol.py


示例5: test_ACK_invalid_blocknum

 def test_ACK_invalid_blocknum(self):
     ack_datagram = ACKDatagram(3)
     yield self.rs.datagramReceived(ack_datagram)
     self.failIf(self.transport.disconnecting)
     err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assert_(isinstance(err_dgram, ERRORDatagram))
     self.addCleanup(self.rs.cancel)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:7,代码来源:test_sessions.py


示例6: test_file_not_found

 def test_file_not_found(self):
     tftp = TFTP(BackendFactory(FileNotFound("Not found")), _clock=self.clock)
     tftp.transport = self.transport
     rrq_datagram = RRQDatagram('foobar', 'netascii', {})
     tftp.datagramReceived(rrq_datagram.to_wire(), ('127.0.0.1', 1111))
     error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assertEqual(error_datagram.errorcode, ERR_FILE_NOT_FOUND)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:7,代码来源:test_protocol.py


示例7: test_invalid_tid

 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)
开发者ID:deepakhajare,项目名称:maas,代码行数:7,代码来源:test_bootstrap.py


示例8: test_bad_mode

 def test_bad_mode(self):
     tftp = TFTP(DummyBackend(), _clock=self.clock)
     tftp.transport = self.transport
     wrq_datagram = WRQDatagram('foobar', 'badmode', {})
     tftp.datagramReceived(wrq_datagram.to_wire(), ('127.0.0.1', 1111))
     error_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assertEqual(error_datagram.errorcode, ERR_ILLEGAL_OP)
开发者ID:deepakhajare,项目名称:maas,代码行数:7,代码来源:test_protocol.py


示例9: test_invalid_tid

 def test_invalid_tid(self):
     data_datagram = DATADatagram(1, b'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)
开发者ID:aivins,项目名称:python-tx-tftp,代码行数:7,代码来源:test_bootstrap.py


示例10: cb

 def cb(ign):
     self.clock.advance(0.1)
     self.failIf(self.transport.disconnecting)
     data_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assertEqual(data_datagram.data, 'line1')
     self.failIf(self.rs.completed,
                 "Got enough bytes from the reader, there is no reason to stop")
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:7,代码来源:test_sessions.py


示例11: cb_

 def cb_(ign):
     self.clock.advance(0.1)
     self.failIf(self.transport.disconnecting)
     ack_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assertEqual(ack_dgram.blocknum, 2)
     self.failIf(self.ws.completed,
                 "Data length is equal to blocksize, no reason to stop")
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:7,代码来源:test_sessions.py


示例12: datagramReceived

 def datagramReceived(self, datagram, addr):
     if self.remote[1] != addr[1]:
         self.transport.write(ERRORDatagram.from_code(ERR_TID_UNKNOWN).to_wire())
         return# Does not belong to this transfer
     datagram = TFTPDatagramFactory(*split_opcode(datagram))
     if datagram.opcode == OP_ERROR:
         return self.tftp_ERROR(datagram)
     return self._datagramReceived(datagram)
开发者ID:deepakhajare,项目名称:maas,代码行数:8,代码来源:bootstrap.py


示例13: test_failed_write

 def test_failed_write(self):
     self.writer.cancel()
     self.ws.writer = FailingWriter()
     data_datagram = DATADatagram(1, 'foobar')
     yield self.ws.datagramReceived(data_datagram)
     self.flushLoggedErrors()
     self.clock.advance(0.1)
     err_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.failUnless(isinstance(err_datagram, ERRORDatagram))
     self.failUnless(self.transport.disconnecting)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:10,代码来源:test_sessions.py


示例14: test_DATA_invalid_blocknum

 def test_DATA_invalid_blocknum(self):
     self.ws.block_size = 6
     data_datagram = DATADatagram(3, 'foobar')
     yield self.ws.datagramReceived(data_datagram)
     self.writer.finish()
     self.failIf(self.target.open('r').read())
     self.failIf(self.transport.disconnecting)
     err_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assert_(isinstance(err_dgram, ERRORDatagram))
     self.addCleanup(self.ws.cancel)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:10,代码来源:test_sessions.py


示例15: test_failed_read

 def test_failed_read(self):
     self.reader.finish()
     self.rs.reader = FailingReader()
     self.rs.blocknum = 1
     ack_datagram = ACKDatagram(1)
     yield self.rs.datagramReceived(ack_datagram)
     self.flushLoggedErrors()
     self.clock.advance(0.1)
     err_datagram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.failUnless(isinstance(err_datagram, ERRORDatagram))
     self.failUnless(self.transport.disconnecting)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:11,代码来源:test_sessions.py


示例16: test_DATA_stale_blocknum

 def test_DATA_stale_blocknum(self):
     self.ws.block_size = 6
     self.ws.blocknum = 2
     data_datagram = DATADatagram(1, 'foobar')
     yield self.ws.datagramReceived(data_datagram)
     self.writer.finish()
     self.failIf(self.target.open('r').read())
     self.failIf(self.transport.disconnecting)
     ack_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
     self.assertEqual(ack_dgram.blocknum, 1)
     self.addCleanup(self.ws.cancel)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:11,代码来源:test_sessions.py


示例17: datagramReceived

    def datagramReceived(self, datagram, addr):
        datagram = TFTPDatagramFactory(*split_opcode(datagram))
        log.msg("Datagram received from %s: %s" % (addr, datagram))

        mode = datagram.mode.lower()
        if datagram.mode not in ('netascii', 'octet'):
            return self.transport.write(ERRORDatagram.from_code(ERR_ILLEGAL_OP,
                "Unknown transfer mode %s, - expected "
                "'netascii' or 'octet' (case-insensitive)" % mode).to_wire(), addr)

        self._clock.callLater(0, self._startSession, datagram, addr, mode)
开发者ID:chronidev,项目名称:python-tx-tftp,代码行数:11,代码来源:protocol.py


示例18: datagramReceived

    def datagramReceived(self, datagram, addr):
        datagram = TFTPDatagramFactory(*split_opcode(datagram))
        log.msg("Datagram received from %s: %s" % (addr, datagram))

        mode = datagram.mode.lower()
        if mode not in (b'netascii', b'octet'):
            errmsg = (
                u"Unknown transfer mode '%s', - expected 'netascii' or 'octet'"
                u"(case-insensitive)" % mode.decode("ascii"))
            return self.transport.write(ERRORDatagram.from_code(
                ERR_ILLEGAL_OP, errmsg.encode("ascii", "replace")).to_wire(), addr)

        self._clock.callLater(0, self._startSession, datagram, addr, mode)
开发者ID:aivins,项目名称:python-tx-tftp,代码行数:13,代码来源:protocol.py


示例19: cb

        def cb(ign):
            self.clock.advance(0.1)
            # self.writer.finish()
            # self.assertEqual(self.target.open('r').read(), 'foobar')
            self.failIf(self.transport.disconnecting)
            ack_dgram = TFTPDatagramFactory(*split_opcode(self.transport.value()))
            self.assertEqual(ack_dgram.blocknum, 1)
            self.failIf(self.ws.completed, "Data length is equal to blocksize, no reason to stop")
            data_datagram = DATADatagram(2, "barbaz")

            self.transport.clear()
            d = self.ws.datagramReceived(data_datagram)
            d.addCallback(cb_)
            self.clock.advance(3)
            return d
开发者ID:karlzheng,项目名称:bashrc,代码行数:15,代码来源:test_sessions.py


示例20: datagramReceived

    def datagramReceived(self, datagram, addr):
        datagram = TFTPDatagramFactory(*split_opcode(datagram))
        log.msg("Datagram received from %s: %s" % (addr, datagram))

        mode = datagram.mode.lower()
        if datagram.mode not in ('netascii', 'octet'):
            return self.transport.write(ERRORDatagram.from_code(ERR_ILLEGAL_OP,
                "Unknown transfer mode %s, - expected "
                "'netascii' or 'octet' (case-insensitive)" % mode).to_wire(), addr)
        try:
            if datagram.opcode == OP_WRQ:
                fs_interface = self.backend.get_writer(datagram.filename)
            elif datagram.opcode == OP_RRQ:
                fs_interface = self.backend.get_reader(datagram.filename)
        except Unsupported, e:
            return self.transport.write(ERRORDatagram.from_code(ERR_ILLEGAL_OP,
                                    str(e)).to_wire(), addr)
开发者ID:rvbad,项目名称:python-tx-tftp,代码行数:17,代码来源:protocol.py



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


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python datagram.ERRORDatagram类代码示例发布时间:2022-05-27
下一篇:
Python generators.fmlin函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap