accounting/bench
2
0
Fork 0
mirror of https://github.com/frappe/bench.git synced 2025-01-06 23:44:03 +00:00
Code Issues Releases Activity

fix: add useforsecurity with fallback for <3.9

Browse Source
This commit is contained in:
18alantom 2024-05-08 10:15:39 +05:30
parent 4c3827207d
commit 319809d2df
1 changed files with 24 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
bench/utils/__init__.py
View File

@ -59,6 +59,7 @@ def is_frappe_app(directory: str) -> bool:
return bool(is_frappe_app) return bool(is_frappe_app)
def get_bench_cache_path(sub_dir: Optional[str]) -> Path: def get_bench_cache_path(sub_dir: Optional[str]) -> Path:
relative_path = "~/.cache/bench" relative_path = "~/.cache/bench"
if sub_dir and not sub_dir.startswith("/"): if sub_dir and not sub_dir.startswith("/"):
@ -69,6 +70,7 @@ def get_bench_cache_path(sub_dir: Optional[str]) -> Path:
cache_path.mkdir(parents=True, exist_ok=True) cache_path.mkdir(parents=True, exist_ok=True)
return cache_path return cache_path
@lru_cache(maxsize=None) @lru_cache(maxsize=None)
def is_valid_frappe_branch(frappe_path: str, frappe_branch: str): def is_valid_frappe_branch(frappe_path: str, frappe_branch: str):
"""Check if a branch exists in a repo. Throws InvalidRemoteException if branch is not found """Check if a branch exists in a repo. Throws InvalidRemoteException if branch is not found
@ -424,7 +426,7 @@ def get_env_frappe_commands(bench_path=".") -> List:
return [] return []
def find_org(org_repo, using_cached: bool=False): def find_org(org_repo, using_cached: bool = False):
import requests import requests
org_repo = org_repo[0] org_repo = org_repo[0]
@ -439,10 +441,14 @@ def find_org(org_repo, using_cached: bool=False):
if using_cached: if using_cached:
return "", org_repo return "", org_repo
raise InvalidRemoteException(f"{org_repo} not found under frappe or erpnext GitHub accounts") raise InvalidRemoteException(
f"{org_repo} not found under frappe or erpnext GitHub accounts"
)
def fetch_details_from_tag(_tag: str, using_cached: bool=False) -> Tuple[str, str, str]: def fetch_details_from_tag(
_tag: str, using_cached: bool = False
) -> Tuple[str, str, str]:
if not _tag: if not _tag:
raise Exception("Tag is not provided") raise Exception("Tag is not provided")
@ -585,14 +591,17 @@ def get_cmd_from_sysargv():
def get_app_cache_extract_filter( def get_app_cache_extract_filter(
count_threshold: int = 10_000, count_threshold: int = 10_000,
size_threshold: int = 1_000_000_000, size_threshold: int = 1_000_000_000,
): # -> Callable[[TarInfo, str], TarInfo | None] ): # -> Callable[[TarInfo, str], TarInfo | None]
state = dict(count=0, size=0) state = dict(count=0, size=0)
AbsoluteLinkError = Exception AbsoluteLinkError = Exception
def data_filter(m: TarInfo, _:str) -> TarInfo:
def data_filter(m: TarInfo, _: str) -> TarInfo:
return m return m
if (sys.version_info.major == 3 and sys.version_info.minor > 7) or sys.version_info.major > 3: if (
sys.version_info.major == 3 and sys.version_info.minor > 7
) or sys.version_info.major > 3:
from tarfile import data_filter, AbsoluteLinkError from tarfile import data_filter, AbsoluteLinkError
def filter_function(member: TarInfo, dest_path: str) -> Optional[TarInfo]: def filter_function(member: TarInfo, dest_path: str) -> Optional[TarInfo]:
@ -613,9 +622,17 @@ def get_app_cache_extract_filter(
return filter_function return filter_function
def get_file_md5(p: Path) -> "str": def get_file_md5(p: Path) -> "str":
with open(p.as_posix(), "rb") as f: with open(p.as_posix(), "rb") as f:
file_md5 = hashlib.md5() try:
file_md5 = hashlib.md5(usedforsecurity=False)
# Will throw if < 3.9, can be removed once support
# is dropped
except TypeError:
file_md5 = hashlib.md5()
while chunk := f.read(2**16): while chunk := f.read(2**16):
file_md5.update(chunk) file_md5.update(chunk)
return file_md5.hexdigest() return file_md5.hexdigest()