mirror of
https://github.com/chatmail/core.git
synced 2026-05-23 00:36:32 +03:00
remove usage of py in favor of pathlib
This commit is contained in:
@@ -13,7 +13,7 @@ from typing import List, Callable
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
import requests
|
import requests
|
||||||
import py
|
import pathlib
|
||||||
|
|
||||||
from . import Account, const, account_hookimpl, get_core_info
|
from . import Account, const, account_hookimpl, get_core_info
|
||||||
from .events import FFIEventLogger, FFIEventTracker
|
from .events import FFIEventLogger, FFIEventTracker
|
||||||
@@ -180,7 +180,7 @@ class TestProcess:
|
|||||||
pytest.fail("more than {} live accounts requested.".format(MAX_LIVE_CREATED_ACCOUNTS))
|
pytest.fail("more than {} live accounts requested.".format(MAX_LIVE_CREATED_ACCOUNTS))
|
||||||
|
|
||||||
def cache_maybe_retrieve_configured_db_files(self, cache_addr, db_target_path):
|
def cache_maybe_retrieve_configured_db_files(self, cache_addr, db_target_path):
|
||||||
db_target_path = py.path.local(db_target_path)
|
db_target_path = pathlib.Path(db_target_path)
|
||||||
assert not db_target_path.exists()
|
assert not db_target_path.exists()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -190,7 +190,7 @@ class TestProcess:
|
|||||||
return False
|
return False
|
||||||
else:
|
else:
|
||||||
print("CACHE HIT for", cache_addr)
|
print("CACHE HIT for", cache_addr)
|
||||||
targetdir = db_target_path.dirpath()
|
targetdir = db_target_path.parent
|
||||||
write_dict_to_dir(filescache, targetdir)
|
write_dict_to_dir(filescache, targetdir)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@@ -200,27 +200,26 @@ class TestProcess:
|
|||||||
# don't overwrite existing entries
|
# don't overwrite existing entries
|
||||||
if addr not in self._addr2files:
|
if addr not in self._addr2files:
|
||||||
print("storing cache for", addr)
|
print("storing cache for", addr)
|
||||||
basedir = py.path.local(acc.get_blobdir()).dirpath()
|
basedir = pathlib.Path(acc.get_blobdir()).parent
|
||||||
self._addr2files[addr] = create_dict_from_files_in_path(basedir)
|
self._addr2files[addr] = create_dict_from_files_in_path(basedir)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
def create_dict_from_files_in_path(path):
|
def create_dict_from_files_in_path(base):
|
||||||
base = py.path.local(path)
|
|
||||||
cachedict = {}
|
cachedict = {}
|
||||||
for path in base.visit(fil=py.path.local.isfile):
|
for path in base.glob("**/*"):
|
||||||
cachedict[path.relto(base)] = path.read_binary()
|
if path.is_file():
|
||||||
|
cachedict[path.relative_to(base)] = path.read_bytes()
|
||||||
return cachedict
|
return cachedict
|
||||||
|
|
||||||
|
|
||||||
def write_dict_to_dir(dic, target_dir):
|
def write_dict_to_dir(dic, target_dir):
|
||||||
assert dic
|
assert dic
|
||||||
target_dir = py.path.local(target_dir)
|
|
||||||
for relpath, content in dic.items():
|
for relpath, content in dic.items():
|
||||||
path = target_dir.join(relpath)
|
path = target_dir.joinpath(relpath)
|
||||||
if not path.dirpath().exists():
|
if not path.parent.exists():
|
||||||
path.dirpath().ensure(dir=1)
|
os.makedirs(path.parent)
|
||||||
path.write_binary(content)
|
path.write_bytes(content)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
|
|||||||
@@ -7,11 +7,27 @@ from deltachat import register_global_plugin
|
|||||||
from deltachat.hookspec import global_hookimpl
|
from deltachat.hookspec import global_hookimpl
|
||||||
from deltachat.capi import ffi
|
from deltachat.capi import ffi
|
||||||
from deltachat.capi import lib
|
from deltachat.capi import lib
|
||||||
from deltachat.testplugin import ACSetup
|
from deltachat.testplugin import ACSetup, create_dict_from_files_in_path, write_dict_to_dir
|
||||||
# from deltachat.account import EventLogger
|
# from deltachat.account import EventLogger
|
||||||
|
|
||||||
|
|
||||||
class TestACSetup:
|
class TestACSetup:
|
||||||
|
|
||||||
|
def test_cache_writing(self, tmp_path):
|
||||||
|
base = tmp_path.joinpath("hello")
|
||||||
|
base.mkdir()
|
||||||
|
d1 = base.joinpath("dir1")
|
||||||
|
d1.mkdir()
|
||||||
|
d1.joinpath("file1").write_bytes(b'content1')
|
||||||
|
d2 = d1.joinpath("dir2")
|
||||||
|
d2.mkdir()
|
||||||
|
d2.joinpath("file2").write_bytes(b"123")
|
||||||
|
d = create_dict_from_files_in_path(base)
|
||||||
|
newbase = tmp_path.joinpath("other")
|
||||||
|
write_dict_to_dir(d, newbase)
|
||||||
|
assert newbase.joinpath("dir1", "dir2", "file2").exists()
|
||||||
|
assert newbase.joinpath("dir1", "file1").exists()
|
||||||
|
|
||||||
def test_basic_states(self, acfactory, monkeypatch, testprocess):
|
def test_basic_states(self, acfactory, monkeypatch, testprocess):
|
||||||
pc = ACSetup(init_time=0.0, testprocess=testprocess)
|
pc = ACSetup(init_time=0.0, testprocess=testprocess)
|
||||||
acc = acfactory.get_unconfigured_account()
|
acc = acfactory.get_unconfigured_account()
|
||||||
|
|||||||
Reference in New Issue
Block a user