本文整理汇总了Python中qutebrowser.misc.ipc.send_or_listen函数的典型用法代码示例。如果您正苦于以下问题:Python send_or_listen函数的具体用法?Python send_or_listen怎么用?Python send_or_listen使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了send_or_listen函数的10个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_normal_connection
def test_normal_connection(self, caplog, qtbot, args):
ret_server = ipc.send_or_listen(args)
assert isinstance(ret_server, ipc.IPCServer)
assert "Starting IPC server..." in caplog.messages
assert ret_server is ipc.server
with qtbot.waitSignal(ret_server.got_args):
ret_client = ipc.send_or_listen(args)
assert ret_client is None
开发者ID:mehak,项目名称:qutebrowser,代码行数:10,代码来源:test_ipc.py
示例2: test_normal_connection
def test_normal_connection(self, caplog, qtbot, args):
ret_server = ipc.send_or_listen(args)
assert isinstance(ret_server, ipc.IPCServer)
msgs = [e.message for e in caplog.records]
assert "Starting IPC server..." in msgs
objreg_server = objreg.get('ipc-server')
assert objreg_server is ret_server
with qtbot.waitSignal(ret_server.got_args, raising=True):
ret_client = ipc.send_or_listen(args)
assert ret_client is None
开发者ID:axs221,项目名称:qutebrowser,代码行数:12,代码来源:test_ipc.py
示例3: test_address_in_use_ok
def test_address_in_use_ok(self, qlocalserver_mock, qlocalsocket_mock,
stubs, caplog, args):
"""Test the following scenario:
- First call to send_to_running_instance:
-> could not connect (server not found)
- Trying to set up a server and listen
-> AddressInUseError
- Second call to send_to_running_instance:
-> success
"""
qlocalserver_mock().listen.return_value = False
err = QAbstractSocket.AddressInUseError
qlocalserver_mock().serverError.return_value = err
qlocalsocket_mock().waitForConnected.side_effect = [False, True]
qlocalsocket_mock().error.side_effect = [
QLocalSocket.ServerNotFoundError, # legacy name
QLocalSocket.ServerNotFoundError,
QLocalSocket.ServerNotFoundError, # legacy name
QLocalSocket.UnknownSocketError,
QLocalSocket.UnknownSocketError, # error() gets called twice
]
ret = ipc.send_or_listen(args)
assert ret is None
msgs = [e.message for e in caplog.records]
assert "Got AddressInUseError, trying again." in msgs
开发者ID:axs221,项目名称:qutebrowser,代码行数:28,代码来源:test_ipc.py
示例4: test_address_in_use_error
def test_address_in_use_error(self, qlocalserver_mock, qlocalsocket_mock,
stubs, caplog, args, has_error, exc_name,
exc_msg):
"""Test the following scenario:
- First call to send_to_running_instance:
-> could not connect (server not found)
- Trying to set up a server and listen
-> AddressInUseError
- Second call to send_to_running_instance:
-> not sent / error
"""
qlocalserver_mock().listen.return_value = False
err = QAbstractSocket.AddressInUseError
qlocalserver_mock().serverError.return_value = err
# If the second connection succeeds, we will have an error later.
# If it fails, that's the "not sent" case above.
qlocalsocket_mock().waitForConnected.side_effect = [False, has_error]
qlocalsocket_mock().error.side_effect = [
QLocalSocket.ServerNotFoundError, # legacy name
QLocalSocket.ServerNotFoundError,
QLocalSocket.ServerNotFoundError,
QLocalSocket.ServerNotFoundError, # legacy name
QLocalSocket.ConnectionRefusedError,
QLocalSocket.ConnectionRefusedError, # error() gets called twice
]
with caplog.atLevel(logging.ERROR):
with pytest.raises(ipc.Error):
ipc.send_or_listen(args)
records = caplog.records()
assert len(records) == 1
error_msgs = [
'Handling fatal misc.ipc.{} with --no-err-windows!'.format(
exc_name),
'',
'title: Error while connecting to running instance!',
'pre_text: ',
'post_text: Maybe another instance is running but frozen?',
'exception text: {}'.format(exc_msg),
]
assert records[0].msg == '\n'.join(error_msgs)
开发者ID:mdlx,项目名称:qutebrowser,代码行数:45,代码来源:test_ipc.py
示例5: run
def run(args):
"""Initialize everything and run the application."""
if args.temp_basedir:
args.basedir = tempfile.mkdtemp(prefix='qutebrowser-basedir-')
quitter = Quitter(args)
objreg.register('quitter', quitter)
log.init.debug("Initializing directories...")
standarddir.init(args)
utils.preload_resources()
log.init.debug("Initializing config...")
configinit.early_init(args)
global q_app
q_app = Application(args)
q_app.setOrganizationName("qutebrowser")
q_app.setApplicationName("qutebrowser")
q_app.setDesktopFileName("qutebrowser")
q_app.setApplicationVersion(qutebrowser.__version__)
q_app.lastWindowClosed.connect(quitter.on_last_window_closed)
if args.version:
print(version.version())
sys.exit(usertypes.Exit.ok)
crash_handler = crashsignal.CrashHandler(
app=q_app, quitter=quitter, args=args, parent=q_app)
crash_handler.activate()
objreg.register('crash-handler', crash_handler)
signal_handler = crashsignal.SignalHandler(app=q_app, quitter=quitter,
parent=q_app)
signal_handler.activate()
objreg.register('signal-handler', signal_handler)
try:
server = ipc.send_or_listen(args)
except ipc.Error:
# ipc.send_or_listen already displays the error message for us.
# We didn't really initialize much so far, so we just quit hard.
sys.exit(usertypes.Exit.err_ipc)
if server is None:
if args.backend is not None:
log.init.warning(
"Backend from the running instance will be used")
sys.exit(usertypes.Exit.ok)
else:
server.got_args.connect(lambda args, target_arg, cwd:
process_pos_args(args, cwd=cwd, via_ipc=True,
target_arg=target_arg))
init(args, crash_handler)
ret = qt_mainloop()
return ret
开发者ID:fiete201,项目名称:qutebrowser,代码行数:57,代码来源:app.py
示例6: test_error_while_listening
def test_error_while_listening(self, qlocalserver_mock, caplog, args):
"""Test an error with the first listen call."""
qlocalserver_mock().listen.return_value = False
err = QAbstractSocket.SocketResourceError
qlocalserver_mock().serverError.return_value = err
with caplog.atLevel(logging.ERROR):
with pytest.raises(ipc.Error):
ipc.send_or_listen(args)
msgs = [e.message for e in caplog.records()]
error_msgs = [
'Handling fatal misc.ipc.ListenError with --no-err-windows!',
'title: Error while connecting to running instance!',
'pre_text: ',
'post_text: Maybe another instance is running but frozen?',
'exception text: Error while listening to IPC server: Error '
'string (error 4)',
]
assert msgs[-5:] == error_msgs
开发者ID:r8b7xy,项目名称:qutebrowser,代码行数:20,代码来源:test_ipc.py
示例7: run
def run(args):
"""Initialize everything and run the application."""
# pylint: disable=too-many-statements
if args.version:
print(version.version(short=True))
print()
print()
print(qutebrowser.__copyright__)
print()
print(version.GPL_BOILERPLATE.strip())
sys.exit(usertypes.Exit.ok)
if args.temp_basedir:
args.basedir = tempfile.mkdtemp(prefix='qutebrowser-basedir-')
quitter = Quitter(args)
objreg.register('quitter', quitter)
global qApp
qApp = Application(args)
qApp.setOrganizationName("qutebrowser")
qApp.setApplicationName("qutebrowser")
qApp.setApplicationVersion(qutebrowser.__version__)
qApp.lastWindowClosed.connect(quitter.on_last_window_closed)
crash_handler = crashsignal.CrashHandler(
app=qApp, quitter=quitter, args=args, parent=qApp)
crash_handler.activate()
objreg.register('crash-handler', crash_handler)
signal_handler = crashsignal.SignalHandler(app=qApp, quitter=quitter,
parent=qApp)
signal_handler.activate()
objreg.register('signal-handler', signal_handler)
try:
server = ipc.send_or_listen(args)
except ipc.Error:
# ipc.send_or_listen already displays the error message for us.
# We didn't really initialize much so far, so we just quit hard.
sys.exit(usertypes.Exit.err_ipc)
if server is None:
sys.exit(usertypes.Exit.ok)
else:
server.got_args.connect(lambda args, target_arg, cwd:
process_pos_args(args, cwd=cwd, via_ipc=True,
target_arg=target_arg))
init(args, crash_handler)
ret = qt_mainloop()
return ret
开发者ID:vyp,项目名称:qutebrowser,代码行数:52,代码来源:app.py
示例8: test_stale_legacy_server
def test_stale_legacy_server(self, caplog, qtbot, args, legacy_server,
ipc_server, py_proc):
legacy_name = ipc._get_socketname(args.basedir, legacy=True)
logging.debug('== Setting up the legacy server ==')
cmdline = py_proc("""
import sys
from PyQt5.QtCore import QCoreApplication
from PyQt5.QtNetwork import QLocalServer
app = QCoreApplication([])
QLocalServer.removeServer(sys.argv[1])
server = QLocalServer()
ok = server.listen(sys.argv[1])
assert ok
print(server.fullServerName())
""")
name = subprocess.check_output(
[cmdline[0]] + cmdline[1] + [legacy_name])
name = name.decode('utf-8').rstrip('\n')
# Closing the server should not remove the FIFO yet
assert os.path.exists(name)
## Setting up the new server
logging.debug('== Setting up new server ==')
ret_server = ipc.send_or_listen(args)
assert isinstance(ret_server, ipc.IPCServer)
logging.debug('== Connecting ==')
with qtbot.waitSignal(ret_server.got_args, raising=True):
ret_client = ipc.send_or_listen(args)
assert ret_client is None
开发者ID:axs221,项目名称:qutebrowser,代码行数:38,代码来源:test_ipc.py
示例9: test_correct_socket_name
def test_correct_socket_name(self, args):
server = ipc.send_or_listen(args)
expected_dir = ipc._get_socketname(args.basedir)
assert '/' in expected_dir
assert server._socketname == expected_dir
开发者ID:axs221,项目名称:qutebrowser,代码行数:5,代码来源:test_ipc.py
示例10: test_legacy_name
def test_legacy_name(self, caplog, qtbot, args, legacy_server):
with qtbot.waitSignal(legacy_server.got_args, raising=True):
ret = ipc.send_or_listen(args)
assert ret is None
msgs = [e.message for e in caplog.records]
assert "Connecting to {}".format(legacy_server._socketname) in msgs
开发者ID:axs221,项目名称:qutebrowser,代码行数:6,代码来源:test_ipc.py
注:本文中的qutebrowser.misc.ipc.send_or_listen函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论