Python pytest.wait_for函数代码示例

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

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

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

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

示例1: test_close

def test_close(Poller, ipv6):
    m = Manager() + Poller()
    server = Server() + UDPServer(0)
    server.register(m)
    m.start()

    try:
        assert pytest.wait_for(server, "ready")
        wait_host(server)

        host, port = server.host, server.port

        server.fire(close())
        assert pytest.wait_for(server, "disconnected")

        server.unregister()

        def test(obj, attr):
            return attr not in obj.components
        assert pytest.wait_for(m, server, value=test)

        server = Server() + UDPServer((host, port))
        server.register(m)

        assert pytest.wait_for(server, "ready", timeout=30.0)
    finally:
        m.stop()

开发者ID:ke4roh，项目名称:circuits，代码行数:27，代码来源:test_udp.py

示例2: test_basic

def test_basic(Poller, ipv6):
    m = Manager() + Poller()

    if ipv6:
        udp_server = UDP6Server(("::1", 0))
        udp_client = UDP6Client(("::1", 0), channel="client")
    else:
        udp_server = UDPServer(0)
        udp_client = UDPClient(0, channel="client")
    server = Server() + udp_server
    client = Client() + udp_client

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(server, "ready")
        assert pytest.wait_for(client, "ready")
        wait_host(server)

        client.fire(write((server.host, server.port), b"foo"))
        assert pytest.wait_for(server, "data", b"foo")

        client.fire(close())
        assert pytest.wait_for(client, "closed")

        server.fire(close())
        assert pytest.wait_for(server, "closed")
    finally:
        m.stop()

开发者ID:ke4roh，项目名称:circuits，代码行数:32，代码来源:test_udp.py

示例3: test_main

def test_main():
    id = "%s:%s" % (os.getpid(), current_thread().getName())

    m = Manager()
    assert repr(m) == "<Manager/ %s (queued=0) [S]>" % id

    app = App()
    app.register(m)
    s = repr(m)
    assert s == "<Manager/ %s (queued=1) [S]>" % id

    m.start()

    pytest.wait_for(m, "_running", True)
    sleep(0.1)

    s = repr(m)
    assert s == "<Manager/ %s (queued=0) [R]>" % id

    m.stop()

    pytest.wait_for(m, "_Manager__thread", None)

    s = repr(m)
    assert s == "<Manager/ %s (queued=0) [S]>" % id

开发者ID:AdricEpic，项目名称:circuits，代码行数:25，代码来源:test_manager_repr.py

示例4: test_tcp_lookup_failure

def test_tcp_lookup_failure(Poller, ipv6):
    m = Manager() + Poller()

    if ipv6:
        tcp_client = TCP6Client()
    else:
        tcp_client = TCPClient()
    client = Client() + tcp_client

    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(client, "ready")

        client.fire(connect("foo", 1234))
        assert pytest.wait_for(
            client, "error", lambda obj, attr: isinstance(getattr(obj, attr), SocketError))
        if pytest.PLATFORM == "win32":
            assert client.error.errno == 11004
        else:
            assert client.error.errno in (EAI_NODATA, EAI_NONAME,)
    finally:
        m.stop()

开发者ID:eriol，项目名称:circuits，代码行数:25，代码来源:test_tcp.py

示例5: test_unix

def test_unix(tmpdir, Poller):
    m = Manager() + Poller()

    sockpath = tmpdir.ensure("test.sock")
    filename = str(sockpath)

    server = Server() + UNIXServer(filename)
    client = Client() + UNIXClient()

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(server, "ready")
        assert pytest.wait_for(client, "ready")

        client.fire(connect(filename))
        assert pytest.wait_for(client, "connected")
        assert pytest.wait_for(server, "connected")
        assert pytest.wait_for(client, "data", b"Ready")

        client.fire(write(b"foo"))
        assert pytest.wait_for(server, "data", b"foo")

        client.fire(close())
        assert pytest.wait_for(client, "disconnected")
        assert pytest.wait_for(server, "disconnected")

        server.fire(close())
        assert pytest.wait_for(server, "closed")
    finally:
        m.stop()
        os.remove(filename)

开发者ID:billtsay，项目名称:win-demo-opcua，代码行数:35，代码来源:test_unix.py

示例6: test_unix

def test_unix(tmpfile, Poller):
    m = Manager() + Poller()

    server = Server() + UNIXServer(tmpfile)
    client = Client() + UNIXClient()

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(server, "ready")
        assert pytest.wait_for(client, "ready")

        client.fire(connect(tmpfile))
        assert pytest.wait_for(client, "connected")
        assert pytest.wait_for(server, "connected")
        assert pytest.wait_for(client, "data", b"Ready")

        client.fire(write(b"foo"))
        assert pytest.wait_for(server, "data", b"foo")

        client.fire(close())
        assert pytest.wait_for(client, "disconnected")
        assert pytest.wait_for(server, "disconnected")

        server.fire(close())
        assert pytest.wait_for(server, "closed")
    finally:
        m.stop()

开发者ID:spaceone，项目名称:circuits，代码行数:31，代码来源:test_unix.py

示例7: test

def test(worker):
    x = worker.fire(task(f))

    assert pytest.wait_for(x, "result")

    assert x.result
    assert x.value == 1000000

开发者ID:carriercomm，项目名称:circuits，代码行数:7，代码来源:test_worker_thread.py

示例8: test_args

def test_args(worker):
    x = worker.fire(task(add, 1, 2))

    assert pytest.wait_for(x, "result")

    assert x.result
    assert x.value == 3

开发者ID:carriercomm，项目名称:circuits，代码行数:7，代码来源:test_worker_thread.py

示例9: test_datetime

def test_datetime(app):
    now = datetime.now()
    d = now + timedelta(seconds=0.1)
    timer = Timer(d, test(), "timer")
    timer.register(app)
    assert pytest.wait_for(app, "flag")
    app.reset()

开发者ID:billtsay，项目名称:win-demo-opcua，代码行数:7，代码来源:test_timers.py

示例10: test_http_1_1_no_host_headers

def test_http_1_1_no_host_headers(webapp):
    transport = TCPClient()
    client = Client()
    client += transport
    client.start()

    host, port, resource, secure = parse_url(webapp.server.http.base)
    client.fire(connect(host, port))
    assert pytest.wait_for(transport, "connected")

    client.fire(write(b"GET / HTTP/1.1\r\n\r\n"))
    assert pytest.wait_for(client, "done")

    client.stop()

    s = client.buffer().decode('utf-8').split('\r\n')[0]
    assert s == "HTTP/1.1 400 Bad Request"

开发者ID:spaceone，项目名称:circuits，代码行数:17，代码来源:test_http.py

示例11: test_tcp_connect_closed_port

def test_tcp_connect_closed_port(Poller, ipv6):
    ### FIXME: This test is wrong.
    ### We need to figure out the sequence of events on Windows
    ### for this scenario. I think if you attempt to connect to
    ### a shutdown listening socket (tcp server) you should get
    ### an error event as response.

    if pytest.PLATFORM == "win32":
        pytest.skip("Broken on Windows")

    m = Manager() + Poller()
    if ipv6:
        tcp_server = TCP6Server(("::1", 0))
        tcp_client = TCP6Client()
    else:
        tcp_server = TCPServer(0)
        tcp_client = TCPClient()
    server = Server() + tcp_server
    client = Client() + tcp_client

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(client, "ready")
        assert pytest.wait_for(server, "ready")
        wait_host(server)

        host, port = server.host, server.port
        tcp_server._sock.close()

        # 1st connect
        client.fire(connect(host, port))
        assert pytest.wait_for(client, "connected")
        assert isinstance(client.error, SocketError)

        client.fire(write(b"foo"))
        assert pytest.wait_for(client, "disconnected")

        client.disconnected = False
        client.fire(write(b"foo"))
        assert pytest.wait_for(client, "disconnected", timeout=1.0) is None
    finally:
        m.stop()

开发者ID:billtsay，项目名称:win-demo-opcua，代码行数:46，代码来源:test_tcp.py

示例12: test

def test(webapp):
    transport = TCPClient()
    client = Client()
    client += transport
    client.start()

    host, port, resource, secure = parse_url(webapp.server.http.base)
    client.fire(connect(host, port))
    assert pytest.wait_for(transport, "connected")

    client.fire(write(b"GET / HTTP/1.1\r\n"))
    client.fire(write(b"Content-Type: text/plain\r\n\r\n"))
    assert pytest.wait_for(client, "done")

    client.stop()

    s = client.buffer().decode('utf-8').split('\r\n')[0]
    assert s == "HTTP/1.1 200 OK"

开发者ID:AdricEpic，项目名称:circuits，代码行数:18，代码来源:test_http.py

示例13: test_inheritence

def test_inheritence():
    app = App1()
    app.start()

    x = app.fire(test())
    assert pytest.wait_for(x, "result")
    v = x.value
    assert v == ["Hello World!", "Foobar"]

    app.stop()

开发者ID:ke4roh，项目名称:circuits，代码行数:10，代码来源:test_inheritence.py

示例14: test_override

def test_override():
    app = App2()
    app.start()

    x = app.fire(test())
    assert pytest.wait_for(x, "result")
    v = x.value
    assert v == "Foobar"

    app.stop()

开发者ID:ke4roh，项目名称:circuits，代码行数:10，代码来源:test_inheritence.py

示例15: test_tcp_bind

def test_tcp_bind(Poller, ipv6):
    m = Manager() + Poller()

    if ipv6:
        sock = socket(AF_INET6, SOCK_STREAM)
        sock.bind(("::1", 0))
        sock.listen(5)
        _, bind_port, _, _ = sock.getsockname()
        sock.close()
        server = Server() + TCP6Server(("::1", 0))
        client = Client() + TCP6Client()
    else:
        sock = socket(AF_INET, SOCK_STREAM)
        sock.bind(("", 0))
        sock.listen(5)
        _, bind_port = sock.getsockname()
        sock.close()
        server = Server() + TCPServer(0)
        client = Client() + TCPClient()

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(client, "ready")
        assert pytest.wait_for(server, "ready")
        wait_host(server)

        client.fire(connect(server.host, server.port))
        assert pytest.wait_for(client, "connected")
        assert pytest.wait_for(server, "connected")
        assert pytest.wait_for(client, "data", b"Ready")

        # assert server.client[1] == bind_port

        client.fire(write(b"foo"))
        assert pytest.wait_for(server, "data", b"foo")

        client.fire(close())
        assert pytest.wait_for(client, "disconnected")
        assert pytest.wait_for(server, "disconnected")

        server.fire(close())
        assert pytest.wait_for(server, "closed")
    finally:
        m.stop()

开发者ID:ke4roh，项目名称:circuits，代码行数:48，代码来源:test_tcp.py

示例16: test_tcps_basic

def test_tcps_basic(Poller, ipv6):
    from circuits import Debugger
    m = Manager() + Debugger() + Poller()

    if ipv6:
        tcp_server = TCP6Server(("::1", 0), secure=True, certfile=CERT_FILE)
        tcp_client = TCP6Client()
    else:
        tcp_server = TCPServer(0, secure=True, certfile=CERT_FILE)
        tcp_client = TCPClient()
    server = Server() + tcp_server
    client = Client() + tcp_client

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(client, "ready")
        assert pytest.wait_for(server, "ready")
        wait_host(server)

        client.fire(connect(server.host, server.port, secure=True))
        assert pytest.wait_for(client, "connected")
        assert pytest.wait_for(server, "connected")
        assert pytest.wait_for(client, "data", b"Ready")

        client.fire(write(b"foo"))
        assert pytest.wait_for(server, "data", b"foo")
        assert pytest.wait_for(client, "data", b"foo")

        client.fire(close())
        assert pytest.wait_for(client, "disconnected")
        assert pytest.wait_for(server, "disconnected")

        server.fire(close())
        assert pytest.wait_for(server, "closed")
    finally:
        m.stop()

开发者ID:giuse88，项目名称:circuits，代码行数:40，代码来源:test_tcp.py

示例17: test_pipe

def test_pipe(Poller):
    m = Manager() + Poller()

    a, b = Pipe("a", "b")
    a.register(m)
    b.register(m)

    a = Client(channel=a.channel).register(m)
    b = Client(channel=b.channel).register(m)

    m.start()

    try:
        assert pytest.wait_for(a, "ready")
        assert pytest.wait_for(b, "ready")

        a.fire(write(b"foo"))
        assert pytest.wait_for(b, "data", b"foo")

        b.fire(write(b"foo"))
        assert pytest.wait_for(a, "data", b"foo")

        a.fire(close())
        assert pytest.wait_for(a, "disconnected")

        b.fire(close())
        assert pytest.wait_for(b, "disconnected")
    finally:
        m.stop()

开发者ID:spaceone，项目名称:circuits，代码行数:29，代码来源:test_pipe.py

示例18: test_tcp_basic

def test_tcp_basic(Poller, ipv6):
    m = Manager() + Poller()

    if ipv6:
        tcp_server = TCP6Server(("::1", 0))
        tcp_client = TCP6Client()
    else:
        tcp_server = TCPServer(0)
        tcp_client = TCPClient()
    server = Server() + tcp_server
    client = Client() + tcp_client

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(client, "ready")
        assert pytest.wait_for(server, "ready")
        wait_host(server)

        client.fire(connect(server.host, server.port))
        assert pytest.wait_for(client, "connected")
        assert pytest.wait_for(server, "connected")
        assert pytest.wait_for(client, "data", b"Ready")

        client.fire(write(b"foo"))
        assert pytest.wait_for(server, "data", b"foo")
        assert pytest.wait_for(client, "data", b"foo")

        client.fire(close())
        assert pytest.wait_for(client, "disconnected")
        assert pytest.wait_for(server, "disconnected")

        server.fire(close())
        assert pytest.wait_for(server, "closed")
    finally:
        m.stop()

开发者ID:ke4roh，项目名称:circuits，代码行数:39，代码来源:test_tcp.py

示例19: test_tcp_connect_closed_port

def test_tcp_connect_closed_port(Poller, ipv6):

    if pytest.PLATFORM == "win32":
        pytest.skip("Broken on Windows")

    m = Manager() + Poller() + Debugger()

    if ipv6:
        tcp_server = TCP6Server(("::1", 0))
        tcp_client = TCP6Client(connect_timeout=1)
    else:
        tcp_server = TCPServer(0)
        tcp_client = TCPClient(connect_timeout=1)
    server = Server() + tcp_server
    client = Client() + tcp_client

    server.register(m)
    client.register(m)

    m.start()

    try:
        assert pytest.wait_for(client, "ready")
        assert pytest.wait_for(server, "ready")
        wait_host(server)

        host, port = server.host, server.port
        tcp_server._sock.close()

        # 1st connect
        client.fire(connect(host, port))
        waiter = WaitEvent(m, "unreachable", channel='client')
        assert waiter.wait()
    finally:
        server.unregister()
        client.unregister()
        m.stop()

开发者ID:ke4roh，项目名称:circuits，代码行数:37，代码来源:test_tcp.py

示例20: test

def test(manager, watcher):
    app = App()
    process, bridge = app.start(process=True, link=manager)
    assert watcher.wait("ready", timeout=30)

    x = manager.fire(hello())

    assert pytest.wait_for(x, "result")

    assert x.value == "Hello from {0:d}".format(app.pid)

    app.stop()
    app.join()

    bridge.unregister()
    watcher.wait("unregistered")

开发者ID:eriol，项目名称:circuits，代码行数:16，代码来源:test_bridge.py

注：本文中的pytest.wait_for函数示例由纯净天空整理自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 pytest.warns函数代码示例发布时间：2022-05-27

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

Python util.grid_equal函数代码示例

1 Python 入门教程

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

阅读：13812|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10205|2022-05-24

3 Python 简介

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

阅读：4092|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4045|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3845|2022-05-27

6 Python 操练实例98

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

阅读：3515|2022-01-22

7 Python 环境搭建

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

阅读：3032|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2655|2022-05-25

9 Python 基础语法

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

阅读：2651|2022-01-22

10 Python 中文编码

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

阅读：2303|2022-01-22

客服电话

电子邮件

Python pytest.wait_for函数代码示例

示例1: test_close

示例2: test_basic

示例3: test_main

示例4: test_tcp_lookup_failure

示例5: test_unix

示例6: test_unix

示例7: test

示例8: test_args

示例9: test_datetime

示例10: test_http_1_1_no_host_headers

示例11: test_tcp_connect_closed_port

示例12: test

示例13: test_inheritence

示例14: test_override

示例15: test_tcp_bind

示例16: test_tcps_basic

示例17: test_pipe

示例18: test_tcp_basic

示例19: test_tcp_connect_closed_port

示例20: test

请发表评论

全部评论

上一篇：

下一篇：

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