Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/svn/tests/utils.py
Show All 17 Lines | class CommitChangeType(Enum): | ||||
Delete = 2 | Delete = 2 | ||||
class CommitChange(TypedDict, total=False): | class CommitChange(TypedDict, total=False): | ||||
change_type: CommitChangeType | change_type: CommitChangeType | ||||
path: str | path: str | ||||
properties: Dict[str, str] | properties: Dict[str, str] | ||||
data: bytes | data: bytes | ||||
copyfrom_path: str | |||||
copyfrom_rev: int | |||||
def add_commit(repo_url: str, message: str, changes: List[CommitChange]) -> None: | def add_commit(repo_url: str, message: str, changes: List[CommitChange]) -> None: | ||||
conn = RemoteAccess(repo_url, auth=Auth([get_username_provider()])) | conn = RemoteAccess(repo_url, auth=Auth([get_username_provider()])) | ||||
editor = conn.get_commit_editor({"svn:log": message}) | editor = conn.get_commit_editor({"svn:log": message}) | ||||
root = editor.open_root() | root = editor.open_root() | ||||
for change in changes: | for change in changes: | ||||
if change["change_type"] == CommitChangeType.Delete: | if change["change_type"] == CommitChangeType.Delete: | ||||
root.delete_entry(change["path"].rstrip("/")) | root.delete_entry(change["path"].rstrip("/")) | ||||
else: | else: | ||||
dir_change = change["path"].endswith("/") | dir_change = change["path"].endswith("/") | ||||
split_path = change["path"].rstrip("/").split("/") | split_path = change["path"].rstrip("/").split("/") | ||||
copyfrom_path = change.get("copyfrom_path") | |||||
copyfrom_rev = change.get("copyfrom_rev", -1) | |||||
for i in range(len(split_path)): | for i in range(len(split_path)): | ||||
path = "/".join(split_path[0 : i + 1]) | path = "/".join(split_path[0 : i + 1]) | ||||
if i < len(split_path) - 1: | if i < len(split_path) - 1: | ||||
try: | try: | ||||
root.add_directory(path).close() | root.add_directory(path, copyfrom_path, copyfrom_rev).close() | ||||
except SubversionException: | except SubversionException: | ||||
pass | pass | ||||
else: | else: | ||||
if dir_change: | if dir_change: | ||||
try: | try: | ||||
dir = root.add_directory(path) | dir = root.add_directory(path, copyfrom_path, copyfrom_rev) | ||||
except SubversionException: | except SubversionException: | ||||
dir = root.open_directory(path) | dir = root.open_directory(path) | ||||
if "properties" in change: | if "properties" in change: | ||||
for prop, value in change["properties"].items(): | for prop, value in change["properties"].items(): | ||||
dir.change_prop(prop, value) | dir.change_prop(prop, value) | ||||
dir.close() | dir.close() | ||||
else: | else: | ||||
try: | try: | ||||
file = root.add_file(path) | file = root.add_file(path, copyfrom_path, copyfrom_rev) | ||||
except SubversionException: | except SubversionException: | ||||
file = root.open_file(path) | file = root.open_file(path) | ||||
if "properties" in change: | if "properties" in change: | ||||
for prop, value in change["properties"].items(): | for prop, value in change["properties"].items(): | ||||
file.change_prop(prop, value) | file.change_prop(prop, value) | ||||
if "data" in change: | if "data" in change: | ||||
txdelta = file.apply_textdelta() | txdelta = file.apply_textdelta() | ||||
delta.send_stream(BytesIO(change["data"]), txdelta) | delta.send_stream(BytesIO(change["data"]), txdelta) | ||||
Show All 9 Lines |