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 | |||||