Implement temporary workaround from unasync package

This commit is contained in:
Serhii Charykov 2022-05-01 18:15:50 +03:00
parent 490fcec1a7
commit 85540c02bd
5 changed files with 73 additions and 96 deletions

View file

@ -1,6 +1,5 @@
import os
from pathlib import Path
from typing import Iterable, Optional, Union
import unasync
@ -9,44 +8,9 @@ ADDITIONAL_REPLACEMENTS = {
"aioredis": "redis",
":tests.": ":tests_sync.",
"pytest_asyncio": "pytest",
"py_test_mark_asyncio": "py_test_mark_sync",
}
STRINGS_TO_REMOVE_FROM_SYNC_TESTS = {
"@pytest.mark.asyncio",
}
def remove_strings_from_files(
filepaths: Iterable[Union[bytes, str, os.PathLike]],
strings_to_remove: Iterable[str],
):
for filepath in filepaths:
tmp_filepath = f"{filepath}.tmp"
with open(filepath, "r") as read_file, open(tmp_filepath, "w") as write_file:
for line in read_file:
if line.strip() in strings_to_remove:
continue
print(line, end="", file=write_file)
os.replace(tmp_filepath, filepath)
def get_source_filepaths(directory: Optional[Union[bytes, str, os.PathLike]] = None):
walk_path = (
Path(__file__).absolute().parent
if directory is None
else os.path.join(Path(__file__).absolute().parent, directory)
)
filepaths = []
for root, _, filenames in os.walk(walk_path):
for filename in filenames:
if filename.rpartition(".")[-1] in (
"py",
"pyi",
):
filepaths.append(os.path.join(root, filename))
return filepaths
def main():
rules = [
@ -61,11 +25,15 @@ def main():
additional_replacements=ADDITIONAL_REPLACEMENTS,
),
]
filepaths = []
for root, _, filenames in os.walk(
Path(__file__).absolute().parent
):
for filename in filenames:
if filename.rpartition(".")[-1] in ("py", "pyi",):
filepaths.append(os.path.join(root, filename))
unasync.unasync_files(get_source_filepaths(), rules)
remove_strings_from_files(
get_source_filepaths("tests_sync"), STRINGS_TO_REMOVE_FROM_SYNC_TESTS
)
unasync.unasync_files(filepaths, rules)
if __name__ == "__main__":