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

Python interfaces.ISession类代码示例

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

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



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

示例1: test_avatarAdaptsToRestrictedExecOnlySession

    def test_avatarAdaptsToRestrictedExecOnlySession(self):
        # When Conch tries to adapt the SSH server avatar to ISession, it
        # adapts to a RestrictedExecOnlySession. This means that a
        # RestrictedExecOnlySession handles any requests to execute a command.
        session = ISession(self.avatar)
        self.failUnless(
            isinstance(session, RestrictedExecOnlySession),
            "ISession(avatar) doesn't adapt to ExecOnlySession. "
            "Got %r instead." % (session,))
        self.assertEqual(
            get_BZR_PLUGIN_PATH_for_subprocess(),
            session.environment['BZR_PLUGIN_PATH'])
        self.assertEqual(
            '%[email protected]' % self.avatar.username,
            session.environment['BZR_EMAIL'])

        executable, arguments = session.getCommandToRun(
            'bzr serve --inet --directory=/ --allow-writes')
        interpreter = '%s/bin/py' % config.root
        self.assertEqual(interpreter, executable)
        self.assertEqual(
            [interpreter, get_bzr_path(), 'lp-serve',
             '--inet', str(self.avatar.user_id)],
            list(arguments))
        self.assertRaises(
            ForbiddenCommand, session.getCommandToRun, 'rm -rf /')
开发者ID:pombreda，项目名称:UnnaturalCodeFork，代码行数:26，代码来源:test_session.py


示例2: test_openShell

    def test_openShell(self):
        """
        The L{ISession} adapter of the L{IConchUser} powerup implements
        C{openShell} so as to associate the given L{IProcessProtocol} with a
        transport.
        """
        proto = ProcessProtocol()
        session = ISession(IConchUser(self.store))

        # XXX See Twisted ticket #3864
        proto.session = session
        proto.write = lambda bytes: None

        # XXX See #2895.
        session.getPty(None, (123, 456, 789, 1000), None)
        session.openShell(proto)
        self.assertNotIdentical(proto.transport, None)
开发者ID:fusionapp，项目名称:mantissa，代码行数:17，代码来源:test_terminal.py


示例3: test_avatarAdaptsToForkingRestrictedExecOnlySession

 def test_avatarAdaptsToForkingRestrictedExecOnlySession(self):
     config.push('codehosting-forking',
         "[codehosting]\nuse_forking_daemon: True\n")
     self.addCleanup(config.pop, 'codehosting-forking')
     session = ISession(self.avatar)
     self.failUnless(
         isinstance(session, ForkingRestrictedExecOnlySession),
         "ISession(avatar) doesn't adapt to "
         " ForkingRestrictedExecOnlySession. "
         "Got %r instead." % (session,))
     executable, arguments = session.getCommandToRun(
         'bzr serve --inet --directory=/ --allow-writes')
     executable, arguments, env = session.getCommandToFork(
         executable, arguments, session.environment)
     self.assertEqual('bzr', executable)
     self.assertEqual(
          ['bzr', 'lp-serve',
           '--inet', str(self.avatar.user_id)],
          list(arguments))
开发者ID:pombreda，项目名称:UnnaturalCodeFork，代码行数:19，代码来源:test_session.py


示例4: request_window_change

 def request_window_change(self, data):
     if not self.session:
         self.session = ISession(self.avatar)
     winSize = parseRequest_window_change(data)
     try:
         self.session.windowChanged(winSize)
     except:
         log.msg('error changing window size')
         log.err()
         return 0
     else:
         return 1
开发者ID:AnthonyNystrom，项目名称:YoGoMee，代码行数:12，代码来源:session.py


示例5: request_pty_req

 def request_pty_req(self, data):
     if not self.session:
         self.session = ISession(self.avatar)
     term, windowSize, modes = parseRequest_pty_req(data)
     log.msg('pty request: %s %s' % (term, windowSize))
     try:
         self.session.getPty(term, windowSize, modes)
     except:
         log.err()
         return 0
     else:
         return 1
开发者ID:AnthonyNystrom，项目名称:YoGoMee，代码行数:12，代码来源:session.py


示例6: request_shell

 def request_shell(self, data):
     log.msg('getting shell')
     if not self.session:
         self.session = ISession(self.avatar)
     try:
         pp = SSHSessionProcessProtocol(self)
         self.session.openShell(pp)
     except:
         log.deferr()
         return 0
     else:
         self.client = pp
         return 1
开发者ID:AnthonyNystrom，项目名称:YoGoMee，代码行数:13，代码来源:session.py


示例7: request_exec

 def request_exec(self, data):
     if not self.session:
         self.session = ISession(self.avatar)
     f,data = common.getNS(data)
     log.msg('executing command "%s"' % f)
     try:
         pp = SSHSessionProcessProtocol(self)
         self.session.execCommand(pp, f)
     except:
         log.deferr()
         return 0
     else:
         self.client = pp
         return 1
开发者ID:AnthonyNystrom，项目名称:YoGoMee，代码行数:14，代码来源:session.py


示例8: SSHSession

class SSHSession(channel.SSHChannel):

    name = 'session'
    def __init__(self, *args, **kw):
        channel.SSHChannel.__init__(self, *args, **kw)
        self.buf = ''
        self.client = None
        self.session = None

    def request_subsystem(self, data):
        subsystem, ignored= common.getNS(data)
        log.msg('asking for subsystem "%s"' % subsystem)
        client = self.avatar.lookupSubsystem(subsystem, data)
        if client:
            pp = SSHSessionProcessProtocol(self)
            proto = wrapProcessProtocol(pp)
            client.makeConnection(proto)
            pp.makeConnection(wrapProtocol(client))
            self.client = pp
            return 1
        else:
            log.msg('failed to get subsystem')
            return 0

    def request_shell(self, data):
        log.msg('getting shell')
        if not self.session:
            self.session = ISession(self.avatar)
        try:
            pp = SSHSessionProcessProtocol(self)
            self.session.openShell(pp)
        except:
            log.deferr()
            return 0
        else:
            self.client = pp
            return 1

    def request_exec(self, data):
        if not self.session:
            self.session = ISession(self.avatar)
        f,data = common.getNS(data)
        log.msg('executing command "%s"' % f)
        try:
            pp = SSHSessionProcessProtocol(self)
            self.session.execCommand(pp, f)
        except:
            log.deferr()
            return 0
        else:
            self.client = pp
            return 1

    def request_pty_req(self, data):
        if not self.session:
            self.session = ISession(self.avatar)
        term, windowSize, modes = parseRequest_pty_req(data)
        log.msg('pty request: %s %s' % (term, windowSize))
        try:
            self.session.getPty(term, windowSize, modes)
        except:
            log.err()
            return 0
        else:
            return 1

    def request_window_change(self, data):
        if not self.session:
            self.session = ISession(self.avatar)
        winSize = parseRequest_window_change(data)
        try:
            self.session.windowChanged(winSize)
        except:
            log.msg('error changing window size')
            log.err()
            return 0
        else:
            return 1

    def dataReceived(self, data):
        if not self.client:
            #self.conn.sendClose(self)
            self.buf += data
            return
        self.client.transport.write(data)

    def extReceived(self, dataType, data):
        if dataType == connection.EXTENDED_DATA_STDERR:
            if self.client and hasattr(self.client.transport, 'writeErr'):
                self.client.transport.writeErr(data)
        else:
            log.msg('weird extended data: %s'%dataType)

    def eofReceived(self):
        if self.session:
            self.session.eofReceived()
        elif self.client:
            self.conn.sendClose(self)

    def closed(self):
#.........这里部分代码省略.........
开发者ID:AnthonyNystrom，项目名称:YoGoMee，代码行数:101，代码来源:session.py


示例9: test_providesISession

 def test_providesISession(self):
     # ExecOnlySession must provide ISession.
     self.failUnless(ISession.providedBy(self.session),
                     "ExecOnlySession doesn't implement ISession")
开发者ID:pombreda，项目名称:UnnaturalCodeFork，代码行数:4，代码来源:test_session.py


示例10: test_providesISession

 def test_providesISession(self):
     # DoNothingSession must provide ISession.
     self.failUnless(ISession.providedBy(self.session),
                     "DoNothingSession doesn't implement ISession")
开发者ID:pombreda，项目名称:UnnaturalCodeFork，代码行数:4，代码来源:test_session.py



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

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
Python manhole.lastColorizedLine函数代码示例发布时间：2022-05-27
下一篇：
Python endpoints.SSHCommandClientEndpoint类代码示例发布时间：2022-05-27
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

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

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