(acfactory, lp)
| 947 | |
| 948 | |
| 949 | def test_connectivity(acfactory, lp): |
| 950 | ac1, ac2 = acfactory.get_online_accounts(2) |
| 951 | |
| 952 | ac1._evtracker.wait_for_connectivity(dc.const.DC_CONNECTIVITY_CONNECTED) |
| 953 | |
| 954 | lp.sec("Test stop_io() and start_io()") |
| 955 | ac1.stop_io() |
| 956 | ac1._evtracker.wait_for_connectivity(dc.const.DC_CONNECTIVITY_NOT_CONNECTED) |
| 957 | |
| 958 | ac1.start_io() |
| 959 | ac1._evtracker.wait_for_connectivity(dc.const.DC_CONNECTIVITY_CONNECTING) |
| 960 | ac1._evtracker.wait_for_connectivity_change(dc.const.DC_CONNECTIVITY_CONNECTING, dc.const.DC_CONNECTIVITY_WORKING) |
| 961 | ac1._evtracker.wait_for_connectivity_change(dc.const.DC_CONNECTIVITY_WORKING, dc.const.DC_CONNECTIVITY_CONNECTED) |
| 962 | |
| 963 | lp.sec( |
| 964 | "Test that after calling start_io(), maybe_network() and waiting for `DC_CONNECTIVITY_CONNECTED`, " |
| 965 | "all messages are fetched", |
| 966 | ) |
| 967 | |
| 968 | ac1.direct_imap.select_config_folder("inbox") |
| 969 | with ac1.direct_imap.idle() as idle1: |
| 970 | ac2.create_chat(ac1).send_text("Hi") |
| 971 | idle1.wait_for_new_message() |
| 972 | ac1.maybe_network() |
| 973 | ac1._evtracker.wait_for_connectivity(dc.const.DC_CONNECTIVITY_CONNECTED) |
| 974 | msgs = ac1.create_chat(ac2).get_messages() |
| 975 | assert len(msgs) == 1 + E2EE_INFO_MSGS |
| 976 | assert msgs[0 + E2EE_INFO_MSGS].text == "Hi" |
| 977 | |
| 978 | lp.sec("Test that the connectivity changes to WORKING while new messages are fetched") |
| 979 | |
| 980 | ac2.create_chat(ac1).send_text("Hi 2") |
| 981 | |
| 982 | ac1._evtracker.wait_for_connectivity_change(dc.const.DC_CONNECTIVITY_CONNECTED, dc.const.DC_CONNECTIVITY_WORKING) |
| 983 | ac1._evtracker.wait_for_connectivity_change(dc.const.DC_CONNECTIVITY_WORKING, dc.const.DC_CONNECTIVITY_CONNECTED) |
| 984 | |
| 985 | msgs = ac1.create_chat(ac2).get_messages() |
| 986 | assert len(msgs) == 2 + E2EE_INFO_MSGS |
| 987 | assert msgs[1 + E2EE_INFO_MSGS].text == "Hi 2" |
| 988 | |
| 989 | |
| 990 | def test_fetch_deleted_msg(acfactory, lp): |
nothing calls this directly
no test coverage detected