本文整理汇总了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；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论