diff --git a/python/tests/stress_test_db.py b/python/tests/stress_test_db.py index 43a04116a..9a793a816 100644 --- a/python/tests/stress_test_db.py +++ b/python/tests/stress_test_db.py @@ -8,7 +8,7 @@ import pytest import deltachat -def test_db_busy_error(acfactory, tmpdir): +def test_db_busy_error(acfactory): starttime = time.time() log_lock = threading.RLock() diff --git a/python/tests/test_0_complex_or_slow.py b/python/tests/test_0_complex_or_slow.py index 736c0d1dd..32d9b82a0 100644 --- a/python/tests/test_0_complex_or_slow.py +++ b/python/tests/test_0_complex_or_slow.py @@ -494,7 +494,7 @@ def test_multidevice_sync_seen(acfactory, lp): assert "Expires: " in ac1_clone_message.get_message_info() -def test_see_new_verified_member_after_going_online(acfactory, tmpdir, lp): +def test_see_new_verified_member_after_going_online(acfactory, tmp_path, lp): """The test for the bug #3836: - Alice has two devices, the second is offline. - Alice creates a verified group and sends a QR invitation to Bob. @@ -507,9 +507,10 @@ def test_see_new_verified_member_after_going_online(acfactory, tmpdir, lp): for ac in [ac1, ac1_offl]: ac.set_config("bcc_self", "1") acfactory.bring_accounts_online() - dir = tmpdir.mkdir("exportdir") - ac1.export_self_keys(dir.strpath) - ac1_offl.import_self_keys(dir.strpath) + dir = tmp_path / "exportdir" + dir.mkdir() + ac1.export_self_keys(str(dir)) + ac1_offl.import_self_keys(str(dir)) ac1_offl.stop_io() lp.sec("ac1: create verified-group QR, ac2 scans and joins") @@ -541,7 +542,7 @@ def test_see_new_verified_member_after_going_online(acfactory, tmpdir, lp): ac1.set_config("bcc_self", "0") -def test_use_new_verified_group_after_going_online(acfactory, tmpdir, lp): +def test_use_new_verified_group_after_going_online(acfactory, tmp_path, lp): """Another test for the bug #3836: - Bob has two devices, the second is offline. - Alice creates a verified group and sends a QR invitation to Bob. @@ -556,9 +557,10 @@ def test_use_new_verified_group_after_going_online(acfactory, tmpdir, lp): for ac in [ac2, ac2_offl]: ac.set_config("bcc_self", "1") acfactory.bring_accounts_online() - dir = tmpdir.mkdir("exportdir") - ac2.export_self_keys(dir.strpath) - ac2_offl.import_self_keys(dir.strpath) + dir = tmp_path / "exportdir" + dir.mkdir() + ac2.export_self_keys(str(dir)) + ac2_offl.import_self_keys(str(dir)) ac2_offl.stop_io() lp.sec("ac1: create verified-group QR, ac2 scans and joins") diff --git a/python/tests/test_1_online.py b/python/tests/test_1_online.py index 84b85014d..b412b1d45 100644 --- a/python/tests/test_1_online.py +++ b/python/tests/test_1_online.py @@ -13,7 +13,7 @@ from deltachat.message import Message from deltachat.tracker import ImexTracker -def test_basic_imap_api(acfactory, tmpdir): +def test_basic_imap_api(acfactory, tmp_path): ac1, ac2 = acfactory.get_online_accounts(2) chat12 = acfactory.get_accepted_chat(ac1, ac2) @@ -28,7 +28,7 @@ def test_basic_imap_api(acfactory, tmpdir): imap2.mark_all_read() assert imap2.get_unread_cnt() == 0 - imap2.dump_imap_structures(tmpdir, logfile=sys.stdout) + imap2.dump_imap_structures(tmp_path, logfile=sys.stdout) imap2.shutdown() @@ -72,35 +72,37 @@ def test_configure_canceled(acfactory): pass -def test_configure_unref(tmpdir): +def test_configure_unref(tmp_path): """Test that removing the last reference to the context during ongoing configuration does not result in use-after-free.""" from deltachat.capi import ffi, lib - path = tmpdir.mkdir("test_configure_unref").join("dc.db").strpath - dc_context = lib.dc_context_new(ffi.NULL, path.encode("utf8"), ffi.NULL) + path = tmp_path / "test_configure_unref" + path.mkdir() + dc_context = lib.dc_context_new(ffi.NULL, str(path / "dc.db").encode("utf8"), ffi.NULL) lib.dc_set_config(dc_context, "addr".encode("utf8"), "foo@x.testrun.org".encode("utf8")) lib.dc_set_config(dc_context, "mail_pw".encode("utf8"), "abc".encode("utf8")) lib.dc_configure(dc_context) lib.dc_context_unref(dc_context) -def test_export_import_self_keys(acfactory, tmpdir, lp): +def test_export_import_self_keys(acfactory, tmp_path, lp): ac1, ac2 = acfactory.get_online_accounts(2) - dir = tmpdir.mkdir("exportdir") - export_files = ac1.export_self_keys(dir.strpath) + dir = tmp_path / "exportdir" + dir.mkdir() + export_files = ac1.export_self_keys(str(dir)) assert len(export_files) == 2 for x in export_files: - assert x.startswith(dir.strpath) + assert x.startswith(str(dir)) (key_id,) = ac1._evtracker.get_info_regex_groups(r".*xporting.*KeyId\((.*)\).*") ac1._evtracker.consume_events() lp.sec("exported keys (private and public)") - for name in os.listdir(dir.strpath): - lp.indent(dir.strpath + os.sep + name) + for name in dir.iterdir(): + lp.indent(str(dir / name)) lp.sec("importing into existing account") - ac2.import_self_keys(dir.strpath) + ac2.import_self_keys(str(dir)) (key_id2,) = ac2._evtracker.get_info_regex_groups(r".*stored.*KeyId\((.*)\).*") assert key_id2 == key_id @@ -156,20 +158,19 @@ def test_one_account_send_bcc_setting(acfactory, lp): assert len(list(ac1.direct_imap.conn.fetch(AND(seen=True)))) == 1 -def test_send_file_twice_unicode_filename_mangling(tmpdir, acfactory, lp): +def test_send_file_twice_unicode_filename_mangling(tmp_path, acfactory, lp): ac1, ac2 = acfactory.get_online_accounts(2) chat = acfactory.get_accepted_chat(ac1, ac2) basename = "somedäüta.html.zip" - p = os.path.join(tmpdir.strpath, basename) - with open(p, "w") as f: - f.write("some data") + p = tmp_path / basename + p.write_text("some data") def send_and_receive_message(): lp.sec("ac1: prepare and send attachment + text to ac2") msg1 = Message.new_empty(ac1, "file") msg1.set_text("withfile") - msg1.set_file(p) + msg1.set_file(str(p)) chat.send_msg(msg1) lp.sec("ac2: receive message") @@ -189,21 +190,20 @@ def test_send_file_twice_unicode_filename_mangling(tmpdir, acfactory, lp): assert msg.filename != msg2.filename -def test_send_file_html_attachment(tmpdir, acfactory, lp): +def test_send_file_html_attachment(tmp_path, acfactory, lp): ac1, ac2 = acfactory.get_online_accounts(2) chat = acfactory.get_accepted_chat(ac1, ac2) basename = "test.html" content = "textdata" - p = os.path.join(tmpdir.strpath, basename) - with open(p, "w") as f: - # write wrong html to see if core tries to parse it - # (it shouldn't as it's a file attachment) - f.write(content) + p = tmp_path / basename + # write wrong html to see if core tries to parse it + # (it shouldn't as it's a file attachment) + p.write_text(content) lp.sec("ac1: prepare and send attachment + text to ac2") - chat.send_file(p, mime_type="text/html") + chat.send_file(str(p), mime_type="text/html") lp.sec("ac2: receive message") ev = ac2._evtracker.get_matching("DC_EVENT_INCOMING_MSG") @@ -1207,7 +1207,7 @@ def test_quote_encrypted(acfactory, lp): assert msg_in.is_encrypted() == quoted_msg.is_encrypted() -def test_quote_attachment(tmpdir, acfactory, lp): +def test_quote_attachment(tmp_path, acfactory, lp): """Test that replies with an attachment and a quote are received correctly.""" ac1, ac2 = acfactory.get_online_accounts(2) @@ -1222,15 +1222,14 @@ def test_quote_attachment(tmpdir, acfactory, lp): assert received_message.text == "hi" basename = "attachment.txt" - p = os.path.join(tmpdir.strpath, basename) - with open(p, "w") as f: - f.write("data to send") + p = tmp_path / basename + p.write_text("data to send") lp.sec("ac2 sends a reply to ac1") chat2 = received_message.create_chat() reply = Message.new_empty(ac2, "file") reply.set_text("message reply") - reply.set_file(p) + reply.set_file(str(p)) reply.quote = received_message chat2.send_msg(reply) @@ -1337,7 +1336,7 @@ def test_send_and_receive_image(acfactory, lp, data): assert m == msg_in -def test_reaction_to_partially_fetched_msg(acfactory, lp, tmpdir): +def test_reaction_to_partially_fetched_msg(acfactory, lp, tmp_path): """See https://github.com/deltachat/deltachat-core-rust/issues/3688 "Partially downloaded messages are received out of order". @@ -1372,10 +1371,9 @@ def test_reaction_to_partially_fetched_msg(acfactory, lp, tmpdir): lp.sec("sending small+large messages from ac1 to ac2") msgs = [] msgs.append(chat.send_text("hi")) - path = tmpdir.join("large") - with open(path, "wb") as fout: - fout.write(os.urandom(download_limit + 1)) - msgs.append(chat.send_file(path.strpath)) + path = tmp_path / "large" + path.write_bytes(os.urandom(download_limit + 1)) + msgs.append(chat.send_file(str(path))) lp.sec("sending a reaction to the large message from ac1 to ac2") react_str = "\N{THUMBS UP SIGN}" @@ -1434,7 +1432,7 @@ def test_reactions_for_a_reordering_move(acfactory, lp): assert reactions.get_by_contact(contacts[0]) == react_str -def test_import_export_online_all(acfactory, tmpdir, data, lp): +def test_import_export_online_all(acfactory, tmp_path, data, lp): (ac1,) = acfactory.get_online_accounts(1) lp.sec("create some chat content") @@ -1446,10 +1444,10 @@ def test_import_export_online_all(acfactory, tmpdir, data, lp): chat1.send_image(original_image_path) # Add another 100KB file that ensures that the progress is smooth enough - path = tmpdir.join("attachment.txt") - with open(path, "w") as file: + path = tmp_path / "attachment.txt" + with path.open("w") as file: file.truncate(100000) - chat1.send_file(path.strpath) + chat1.send_file(str(path)) def assert_account_is_proper(ac): contacts = ac.get_contacts(query="some1") @@ -1467,12 +1465,13 @@ def test_import_export_online_all(acfactory, tmpdir, data, lp): assert_account_is_proper(ac1) - backupdir = tmpdir.mkdir("backup") + backupdir = tmp_path / "backup" + backupdir.mkdir() lp.sec(f"export all to {backupdir}") with ac1.temp_plugin(ImexTracker()) as imex_tracker: ac1.stop_io() - ac1.imex(backupdir.strpath, const.DC_IMEX_EXPORT_BACKUP) + ac1.imex(str(backupdir), const.DC_IMEX_EXPORT_BACKUP) # check progress events for export assert imex_tracker.wait_progress(1, progress_upper_limit=249) @@ -1490,7 +1489,7 @@ def test_import_export_online_all(acfactory, tmpdir, data, lp): ac2 = acfactory.get_unconfigured_account() lp.sec("get latest backup file") - path2 = ac2.get_latest_backupfile(backupdir.strpath) + path2 = ac2.get_latest_backupfile(str(backupdir)) assert path2 == path lp.sec("import backup and check it's proper") @@ -1508,10 +1507,10 @@ def test_import_export_online_all(acfactory, tmpdir, data, lp): lp.sec(f"Second-time export all to {backupdir}") ac1.stop_io() - path2 = ac1.export_all(backupdir.strpath) + path2 = ac1.export_all(str(backupdir)) assert os.path.exists(path2) assert path2 != path - assert ac2.get_latest_backupfile(backupdir.strpath) == path2 + assert ac2.get_latest_backupfile(str(backupdir)) == path2 def test_ac_setup_message(acfactory, lp): diff --git a/python/tests/test_2_increation.py b/python/tests/test_2_increation.py index c7e35119a..761b3eb86 100644 --- a/python/tests/test_2_increation.py +++ b/python/tests/test_2_increation.py @@ -30,25 +30,26 @@ def wait_msgs_changed(account, msgs_list): class TestOnlineInCreation: - def test_increation_not_blobdir(self, tmpdir, acfactory, lp): + def test_increation_not_blobdir(self, tmp_path, acfactory, lp): ac1, ac2 = acfactory.get_online_accounts(2) chat = ac1.create_chat(ac2) lp.sec("Creating in-creation file outside of blobdir") - assert tmpdir.strpath != ac1.get_blobdir() - src = tmpdir.join("file.txt").ensure(file=1) + assert str(tmp_path) != ac1.get_blobdir() + src = tmp_path / "file.txt" + src.touch() with pytest.raises(Exception): - chat.prepare_message_file(src.strpath) + chat.prepare_message_file(str(src)) - def test_no_increation_copies_to_blobdir(self, tmpdir, acfactory, lp): + def test_no_increation_copies_to_blobdir(self, tmp_path, acfactory, lp): ac1, ac2 = acfactory.get_online_accounts(2) chat = ac1.create_chat(ac2) lp.sec("Creating file outside of blobdir") - assert tmpdir.strpath != ac1.get_blobdir() - src = tmpdir.join("file.txt") - src.write("hello there\n") - chat.send_file(src.strpath) + assert str(tmp_path) != ac1.get_blobdir() + src = tmp_path / "file.txt" + src.write_text("hello there\n") + chat.send_file(str(src)) blob_src = os.path.join(ac1.get_blobdir(), "file.txt") assert os.path.exists(blob_src), "file.txt not copied to blobdir" diff --git a/python/tests/test_3_offline.py b/python/tests/test_3_offline.py index e9a297681..50a95ed61 100644 --- a/python/tests/test_3_offline.py +++ b/python/tests/test_3_offline.py @@ -52,18 +52,18 @@ def test_parse_system_add_remove(msgtext, res): class TestOfflineAccountBasic: - def test_wrong_db(self, tmpdir): - p = tmpdir.join("hello.db") - p.write("123") - account = Account(p.strpath) + def test_wrong_db(self, tmp_path): + p = tmp_path / "hello.db" + p.write_text("123") + account = Account(str(p)) assert not account.is_open() - def test_os_name(self, tmpdir): - p = tmpdir.join("hello.db") + def test_os_name(self, tmp_path): + p = tmp_path / "hello.db" # we can't easily test if os_name is used in X-Mailer # outgoing messages without a full Online test # but we at least check Account accepts the arg - ac1 = Account(p.strpath, os_name="solarpunk") + ac1 = Account(str(p), os_name="solarpunk") ac1.get_info() def test_preconfigure_keypair(self, acfactory, data): @@ -496,22 +496,22 @@ class TestOfflineChat: contact = msg.get_sender_contact() assert contact == ac1.get_self_contact() - def test_import_export_on_unencrypted_acct(self, acfactory, tmpdir): - backupdir = tmpdir.mkdir("backup") + def test_import_export_on_unencrypted_acct(self, acfactory, tmp_path): + backupdir = tmp_path / "backup" + backupdir.mkdir() ac1 = acfactory.get_pseudo_configured_account() chat = ac1.create_contact("some1 ").create_chat() # send a text message msg = chat.send_text("msg1") # send a binary file - bin = tmpdir.join("some.bin") - with bin.open("w") as f: - f.write("\00123" * 10000) - msg = chat.send_file(bin.strpath) + bin = tmp_path / "some.bin" + bin.write_bytes(b"\00123" * 10000) + msg = chat.send_file(str(bin)) contact = msg.get_sender_contact() assert contact == ac1.get_self_contact() - assert not backupdir.listdir() + assert not list(backupdir.iterdir()) ac1.stop_io() - path = ac1.export_all(backupdir.strpath) + path = ac1.export_all(str(backupdir)) assert os.path.exists(path) ac2 = acfactory.get_unconfigured_account() ac2.import_all(path) @@ -525,27 +525,27 @@ class TestOfflineChat: assert messages[0].text == "msg1" assert os.path.exists(messages[1].filename) - def test_import_export_on_encrypted_acct(self, acfactory, tmpdir): + def test_import_export_on_encrypted_acct(self, acfactory, tmp_path): passphrase1 = "passphrase1" passphrase2 = "passphrase2" - backupdir = tmpdir.mkdir("backup") + backupdir = tmp_path / "backup" + backupdir.mkdir() ac1 = acfactory.get_pseudo_configured_account(passphrase=passphrase1) chat = ac1.create_contact("some1 ").create_chat() # send a text message msg = chat.send_text("msg1") # send a binary file - bin = tmpdir.join("some.bin") - with bin.open("w") as f: - f.write("\00123" * 10000) - msg = chat.send_file(bin.strpath) + bin = tmp_path / "some.bin" + bin.write_bytes(b"\00123" * 10000) + msg = chat.send_file(str(bin)) contact = msg.get_sender_contact() assert contact == ac1.get_self_contact() - assert not backupdir.listdir() + assert not list(backupdir.iterdir()) ac1.stop_io() - path = ac1.export_all(backupdir.strpath) + path = ac1.export_all(str(backupdir)) assert os.path.exists(path) ac2 = acfactory.get_unconfigured_account(closed=True) @@ -580,27 +580,27 @@ class TestOfflineChat: assert messages[0].text == "msg1" assert os.path.exists(messages[1].filename) - def test_import_export_with_passphrase(self, acfactory, tmpdir): + def test_import_export_with_passphrase(self, acfactory, tmp_path): passphrase = "test_passphrase" wrong_passphrase = "wrong_passprase" - backupdir = tmpdir.mkdir("backup") + backupdir = tmp_path / "backup" + backupdir.mkdir() ac1 = acfactory.get_pseudo_configured_account() chat = ac1.create_contact("some1 ").create_chat() # send a text message msg = chat.send_text("msg1") # send a binary file - bin = tmpdir.join("some.bin") - with bin.open("w") as f: - f.write("\00123" * 10000) - msg = chat.send_file(bin.strpath) + bin = tmp_path / "some.bin" + bin.write_bytes(b"\00123" * 10000) + msg = chat.send_file(str(bin)) contact = msg.get_sender_contact() assert contact == ac1.get_self_contact() - assert not backupdir.listdir() + assert not list(backupdir.iterdir()) ac1.stop_io() - path = ac1.export_all(backupdir.strpath, passphrase) + path = ac1.export_all(str(backupdir), passphrase) assert os.path.exists(path) ac2 = acfactory.get_unconfigured_account() @@ -619,7 +619,7 @@ class TestOfflineChat: assert messages[0].text == "msg1" assert os.path.exists(messages[1].filename) - def test_import_encrypted_bak_into_encrypted_acct(self, acfactory, tmpdir): + def test_import_encrypted_bak_into_encrypted_acct(self, acfactory, tmp_path): """ Test that account passphrase isn't lost if backup failed to be imported. See https://github.com/deltachat/deltachat-core-rust/issues/3379 @@ -627,24 +627,24 @@ class TestOfflineChat: acct_passphrase = "passphrase1" bak_passphrase = "passphrase2" wrong_passphrase = "wrong_passprase" - backupdir = tmpdir.mkdir("backup") + backupdir = tmp_path / "backup" + backupdir.mkdir() ac1 = acfactory.get_pseudo_configured_account() chat = ac1.create_contact("some1 ").create_chat() # send a text message msg = chat.send_text("msg1") # send a binary file - bin = tmpdir.join("some.bin") - with bin.open("w") as f: - f.write("\00123" * 10000) - msg = chat.send_file(bin.strpath) + bin = tmp_path / "some.bin" + bin.write_bytes(b"\00123" * 10000) + msg = chat.send_file(str(bin)) contact = msg.get_sender_contact() assert contact == ac1.get_self_contact() - assert not backupdir.listdir() + assert not list(backupdir.iterdir()) ac1.stop_io() - path = ac1.export_all(backupdir.strpath, bak_passphrase) + path = ac1.export_all(str(backupdir), bak_passphrase) assert os.path.exists(path) ac2 = acfactory.get_unconfigured_account(closed=True) diff --git a/python/tests/test_4_lowlevel.py b/python/tests/test_4_lowlevel.py index 1c3f15bea..ac7ce1f12 100644 --- a/python/tests/test_4_lowlevel.py +++ b/python/tests/test_4_lowlevel.py @@ -1,4 +1,3 @@ -import os from queue import Queue from deltachat import capi, const, cutil, register_global_plugin @@ -65,16 +64,17 @@ class TestACSetup: assert pc._account2state[ac1] == pc.IDLEREADY assert pc._account2state[ac2] == pc.IDLEREADY - def test_store_and_retrieve_configured_account_cache(self, acfactory, tmpdir): + def test_store_and_retrieve_configured_account_cache(self, acfactory, tmp_path): ac1 = acfactory.get_pseudo_configured_account() holder = acfactory._acsetup.testprocess assert holder.cache_maybe_store_configured_db_files(ac1) assert not holder.cache_maybe_store_configured_db_files(ac1) - acdir = tmpdir.mkdir("newaccount") + acdir = tmp_path / "newaccount" + acdir.mkdir() addr = ac1.get_config("addr") - target_db_path = acdir.join("db").strpath - assert holder.cache_maybe_retrieve_configured_db_files(addr, target_db_path) - assert len(os.listdir(acdir)) >= 2 + target_db_path = acdir / "db" + assert holder.cache_maybe_retrieve_configured_db_files(addr, str(target_db_path)) + assert sum(1 for _ in acdir.iterdir()) >= 2 def test_liveconfig_caching(acfactory, monkeypatch): @@ -112,20 +112,20 @@ def test_dc_close_events(acfactory): shutdowns.get(timeout=2) -def test_wrong_db(tmpdir): - p = tmpdir.join("hello.db") +def test_wrong_db(tmp_path): + p = tmp_path / "hello.db" # write an invalid database file - p.write("x123" * 10) + p.write_bytes(b"x123" * 10) - context = lib.dc_context_new(ffi.NULL, p.strpath.encode("ascii"), ffi.NULL) + context = lib.dc_context_new(ffi.NULL, str(p).encode("ascii"), ffi.NULL) assert not lib.dc_context_is_open(context) -def test_empty_blobdir(tmpdir): - db_fname = tmpdir.join("hello.db") +def test_empty_blobdir(tmp_path): + db_fname = tmp_path / "hello.db" # Apparently some client code expects this to be the same as passing NULL. ctx = ffi.gc( - lib.dc_context_new(ffi.NULL, db_fname.strpath.encode("ascii"), b""), + lib.dc_context_new(ffi.NULL, str(db_fname).encode("ascii"), b""), lib.dc_context_unref, ) assert ctx != ffi.NULL @@ -174,10 +174,10 @@ def test_provider_info_none(): assert lib.dc_provider_new_from_email(ctx, cutil.as_dc_charpointer("email@unexistent.no")) == ffi.NULL -def test_get_info_open(tmpdir): - db_fname = tmpdir.join("test.db") +def test_get_info_open(tmp_path): + db_fname = tmp_path / "test.db" ctx = ffi.gc( - lib.dc_context_new(ffi.NULL, db_fname.strpath.encode("ascii"), ffi.NULL), + lib.dc_context_new(ffi.NULL, str(db_fname).encode("ascii"), ffi.NULL), lib.dc_context_unref, ) info = cutil.from_dc_charpointer(lib.dc_get_info(ctx)) @@ -218,10 +218,10 @@ def test_logged_ac_process_ffi_failure(acfactory): assert "Traceback" in res -def test_jsonrpc_blocking_call(tmpdir): - accounts_fname = tmpdir.join("accounts") +def test_jsonrpc_blocking_call(tmp_path): + accounts_fname = tmp_path / "accounts" accounts = ffi.gc( - lib.dc_accounts_new(ffi.NULL, accounts_fname.strpath.encode("ascii")), + lib.dc_accounts_new(ffi.NULL, str(accounts_fname).encode("ascii")), lib.dc_accounts_unref, ) jsonrpc = ffi.gc(lib.dc_jsonrpc_init(accounts), lib.dc_jsonrpc_unref)