Changeset View
Standalone View
swh/loader/svn/ra.py
# Copyright (C) 2016-2021 The Software Heritage developers | # Copyright (C) 2016-2022 The Software Heritage developers | |||||||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | |||||||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | |||||||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | |||||||||
"""Remote Access client to svn server. | """Remote Access client to svn server. | |||||||||
""" | """ | |||||||||
from __future__ import annotations | from __future__ import annotations | |||||||||
import codecs | import codecs | |||||||||
import dataclasses | from collections import defaultdict | |||||||||
from dataclasses import dataclass, field | ||||||||||
from itertools import chain | ||||||||||
import logging | import logging | |||||||||
import os | import os | |||||||||
import shutil | import shutil | |||||||||
import tempfile | import tempfile | |||||||||
from typing import ( | from typing import ( | |||||||||
TYPE_CHECKING, | TYPE_CHECKING, | |||||||||
Any, | Any, | |||||||||
BinaryIO, | BinaryIO, | |||||||||
Callable, | Callable, | |||||||||
Dict, | Dict, | |||||||||
List, | List, | |||||||||
Optional, | Optional, | |||||||||
Set, | ||||||||||
Tuple, | Tuple, | |||||||||
Union, | Union, | |||||||||
cast, | cast, | |||||||||
) | ) | |||||||||
import click | import click | |||||||||
from subvertpy import delta, properties | from subvertpy import SubversionException, delta, properties | |||||||||
from subvertpy.ra import Auth, RemoteAccess, get_username_provider | from subvertpy.ra import Auth, RemoteAccess, get_username_provider | |||||||||
from swh.model import from_disk, hashutil | from swh.model import from_disk, hashutil | |||||||||
from swh.model.model import Content, Directory, SkippedContent | from swh.model.model import Content, Directory, SkippedContent | |||||||||
if TYPE_CHECKING: | if TYPE_CHECKING: | |||||||||
from swh.loader.svn.svn import SvnRepo | from swh.loader.svn.svn import SvnRepo | |||||||||
from swh.loader.svn.utils import parse_external_definition, svn_urljoin | ||||||||||
_eol_style = {"native": b"\n", "CRLF": b"\r\n", "LF": b"\n", "CR": b"\r"} | _eol_style = {"native": b"\n", "CRLF": b"\r\n", "LF": b"\n", "CR": b"\r"} | |||||||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | |||||||||
def _normalize_line_endings(lines: bytes, eol_style: str = "native") -> bytes: | def _normalize_line_endings(lines: bytes, eol_style: str = "native") -> bytes: | |||||||||
r"""Normalize line endings to unix (\\n), windows (\\r\\n) or mac (\\r). | r"""Normalize line endings to unix (\\n), windows (\\r\\n) or mac (\\r). | |||||||||
▲ Show 20 Lines • Show All 96 Lines • ▼ Show 20 Lines | ||||||||||
DEFAULT_FLAG = 0 | DEFAULT_FLAG = 0 | |||||||||
EXEC_FLAG = 1 | EXEC_FLAG = 1 | |||||||||
NOEXEC_FLAG = 2 | NOEXEC_FLAG = 2 | |||||||||
SVN_PROPERTY_EOL = "svn:eol-style" | SVN_PROPERTY_EOL = "svn:eol-style" | |||||||||
@dataclasses.dataclass | @dataclass | |||||||||
class FileState: | class FileState: | |||||||||
"""Persists some file states (eg. end of lines style) across revisions while | """Persists some file states (eg. end of lines style) across revisions while | |||||||||
replaying them.""" | replaying them.""" | |||||||||
eol_style: Optional[str] = None | eol_style: Optional[str] = None | |||||||||
"""EOL state check mess""" | """EOL state check mess""" | |||||||||
svn_special_path_non_link_data: Optional[bytes] = None | svn_special_path_non_link_data: Optional[bytes] = None | |||||||||
Show All 15 Lines | class FileEditor: | |||||||||
__slots__ = [ | __slots__ = [ | |||||||||
"directory", | "directory", | |||||||||
"path", | "path", | |||||||||
"fullpath", | "fullpath", | |||||||||
"executable", | "executable", | |||||||||
"link", | "link", | |||||||||
"state", | "state", | |||||||||
"svnrepo", | "svnrepo", | |||||||||
"editor", | ||||||||||
] | ] | |||||||||
def __init__( | def __init__( | |||||||||
self, | self, | |||||||||
directory: from_disk.Directory, | directory: from_disk.Directory, | |||||||||
rootpath: bytes, | rootpath: bytes, | |||||||||
path: bytes, | path: bytes, | |||||||||
state: FileState, | state: FileState, | |||||||||
svnrepo: SvnRepo, | svnrepo: SvnRepo, | |||||||||
): | ): | |||||||||
self.directory = directory | self.directory = directory | |||||||||
self.path = path | self.path = path | |||||||||
self.fullpath = os.path.join(rootpath, path) | self.fullpath = os.path.join(rootpath, path) | |||||||||
self.state = state | self.state = state | |||||||||
self.svnrepo = svnrepo | self.svnrepo = svnrepo | |||||||||
self.editor = svnrepo.swhreplay.editor | ||||||||||
def change_prop(self, key: str, value: str) -> None: | def change_prop(self, key: str, value: str) -> None: | |||||||||
if key == properties.PROP_EXECUTABLE: | if key == properties.PROP_EXECUTABLE: | |||||||||
if value is None: # bit flip off | if value is None: # bit flip off | |||||||||
self.state.executable = NOEXEC_FLAG | self.state.executable = NOEXEC_FLAG | |||||||||
else: | else: | |||||||||
self.state.executable = EXEC_FLAG | self.state.executable = EXEC_FLAG | |||||||||
elif key == properties.PROP_SPECIAL: | elif key == properties.PROP_SPECIAL: | |||||||||
Show All 34 Lines | def __make_svnlink(self) -> bytes: | |||||||||
src = os.readlink(self.fullpath) | src = os.readlink(self.fullpath) | |||||||||
os.remove(self.fullpath) | os.remove(self.fullpath) | |||||||||
sbuf = b"link " + src | sbuf = b"link " + src | |||||||||
with open(self.fullpath, "wb") as f: | with open(self.fullpath, "wb") as f: | |||||||||
f.write(sbuf) | f.write(sbuf) | |||||||||
return sbuf | return sbuf | |||||||||
def apply_textdelta(self, base_checksum) -> Callable[[Any, bytes, BinaryIO], None]: | def apply_textdelta(self, base_checksum) -> Callable[[Any, bytes, BinaryIO], None]: | |||||||||
# if the filepath matches an external, do not apply local patch | ||||||||||
if self.path in self.editor.external_paths: | ||||||||||
return lambda *args: None | ||||||||||
if os.path.lexists(self.fullpath): | if os.path.lexists(self.fullpath): | |||||||||
if os.path.islink(self.fullpath): | if os.path.islink(self.fullpath): | |||||||||
# svn does not deal with symlink so we transform into | # svn does not deal with symlink so we transform into | |||||||||
# real svn symlink for potential patching in later | # real svn symlink for potential patching in later | |||||||||
# commits | # commits | |||||||||
sbuf = self.__make_svnlink() | sbuf = self.__make_svnlink() | |||||||||
self.state.link = True | self.state.link = True | |||||||||
else: | else: | |||||||||
Show All 28 Lines | def close(self) -> None: | |||||||||
# the svn export operation might extract a truncated version of it | # the svn export operation might extract a truncated version of it | |||||||||
# if it is a binary file, so ensure to produce the same file as the | # if it is a binary file, so ensure to produce the same file as the | |||||||||
# export operation. | # export operation. | |||||||||
with open(self.fullpath, "rb") as f: | with open(self.fullpath, "rb") as f: | |||||||||
content = f.read() | content = f.read() | |||||||||
self.svnrepo.client.export( | self.svnrepo.client.export( | |||||||||
os.path.join(self.svnrepo.remote_url.encode(), self.path), | os.path.join(self.svnrepo.remote_url.encode(), self.path), | |||||||||
to=self.fullpath, | to=self.fullpath, | |||||||||
rev=self.svnrepo.swhreplay.editor.revnum, | rev=self.editor.revnum, | |||||||||
ignore_keywords=True, | ignore_keywords=True, | |||||||||
overwrite=True, | overwrite=True, | |||||||||
) | ) | |||||||||
with open(self.fullpath, "rb") as f: | with open(self.fullpath, "rb") as f: | |||||||||
exported_data = f.read() | exported_data = f.read() | |||||||||
if exported_data != content: | if exported_data != content: | |||||||||
# keep track of original file content in order to restore | # keep track of original file content in order to restore | |||||||||
# it if the svn:special property gets unset in another revision | # it if the svn:special property gets unset in another revision | |||||||||
Show All 27 Lines | def close(self) -> None: | |||||||||
mode = os.lstat(self.fullpath).st_mode | mode = os.lstat(self.fullpath).st_mode | |||||||||
self.directory[self.path] = from_disk.Content.from_bytes( | self.directory[self.path] = from_disk.Content.from_bytes( | |||||||||
mode=mode, data=data | mode=mode, data=data | |||||||||
) | ) | |||||||||
else: | else: | |||||||||
self.directory[self.path] = from_disk.Content.from_file(path=self.fullpath) | self.directory[self.path] = from_disk.Content.from_file(path=self.fullpath) | |||||||||
@dataclass | ||||||||||
class DirState: | ||||||||||
"""Persists some directory states (eg. externals) across revisions while | ||||||||||
replaying them.""" | ||||||||||
externals: Dict[str, Tuple[str, Optional[int]]] = field(default_factory=dict) | ||||||||||
class DirEditor: | class DirEditor: | |||||||||
"""Directory Editor in charge of updating directory hashes computation. | """Directory Editor in charge of updating directory hashes computation. | |||||||||
This implementation includes empty folder in the hash computation. | This implementation includes empty folder in the hash computation. | |||||||||
""" | """ | |||||||||
__slots__ = ["directory", "rootpath", "path", "file_states", "svnrepo"] | __slots__ = [ | |||||||||
"directory", | ||||||||||
"rootpath", | ||||||||||
"path", | ||||||||||
"file_states", | ||||||||||
"dir_states", | ||||||||||
"svnrepo", | ||||||||||
"editor", | ||||||||||
"externals", | ||||||||||
] | ||||||||||
def __init__( | def __init__( | |||||||||
self, | self, | |||||||||
directory: from_disk.Directory, | directory: from_disk.Directory, | |||||||||
rootpath: bytes, | rootpath: bytes, | |||||||||
path: bytes, | path: bytes, | |||||||||
file_states: Dict[bytes, FileState], | file_states: Dict[bytes, FileState], | |||||||||
dir_states: Dict[bytes, DirState], | ||||||||||
svnrepo: SvnRepo, | svnrepo: SvnRepo, | |||||||||
): | ): | |||||||||
self.directory = directory | self.directory = directory | |||||||||
self.rootpath = rootpath | self.rootpath = rootpath | |||||||||
self.path = path | self.path = path | |||||||||
# build directory on init | # build directory on init | |||||||||
os.makedirs(rootpath, exist_ok=True) | os.makedirs(rootpath, exist_ok=True) | |||||||||
self.file_states = file_states | self.file_states = file_states | |||||||||
self.dir_states = dir_states | ||||||||||
self.svnrepo = svnrepo | self.svnrepo = svnrepo | |||||||||
self.editor = svnrepo.swhreplay.editor | ||||||||||
self.externals: Dict[str, Tuple[str, Optional[int], bool]] = {} | ||||||||||
def remove_child(self, path: bytes) -> None: | def remove_child(self, path: bytes) -> None: | |||||||||
"""Remove a path from the current objects. | """Remove a path from the current objects. | |||||||||
The path can be resolved as link, file or directory. | The path can be resolved as link, file or directory. | |||||||||
This function takes also care of removing the link between the | This function takes also care of removing the link between the | |||||||||
child and the parent. | child and the parent. | |||||||||
Show All 26 Lines | def open_directory(self, path: str, *args) -> DirEditor: | |||||||||
"""Updating existing directory. | """Updating existing directory. | |||||||||
""" | """ | |||||||||
return DirEditor( | return DirEditor( | |||||||||
self.directory, | self.directory, | |||||||||
rootpath=self.rootpath, | rootpath=self.rootpath, | |||||||||
path=os.fsencode(path), | path=os.fsencode(path), | |||||||||
file_states=self.file_states, | file_states=self.file_states, | |||||||||
dir_states=self.dir_states, | ||||||||||
svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | |||||||||
) | ) | |||||||||
def add_directory(self, path: str, *args) -> DirEditor: | def add_directory(self, path: str, *args) -> DirEditor: | |||||||||
"""Adding a new directory. | """Adding a new directory. | |||||||||
""" | """ | |||||||||
path_bytes = os.fsencode(path) | path_bytes = os.fsencode(path) | |||||||||
os.makedirs(os.path.join(self.rootpath, path_bytes), exist_ok=True) | os.makedirs(os.path.join(self.rootpath, path_bytes), exist_ok=True) | |||||||||
if path_bytes and path_bytes not in self.directory: | ||||||||||
self.dir_states[path_bytes] = DirState() | ||||||||||
self.directory[path_bytes] = from_disk.Directory() | self.directory[path_bytes] = from_disk.Directory() | |||||||||
return DirEditor( | return DirEditor( | |||||||||
self.directory, | self.directory, | |||||||||
rootpath=self.rootpath, | self.rootpath, | |||||||||
path=path_bytes, | path_bytes, | |||||||||
file_states=self.file_states, | self.file_states, | |||||||||
self.dir_states, | ||||||||||
svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | |||||||||
) | ) | |||||||||
def open_file(self, *args) -> FileEditor: | def open_file(self, *args) -> FileEditor: | |||||||||
"""Updating existing file. | """Updating existing file. | |||||||||
""" | """ | |||||||||
path = os.fsencode(args[0]) | path = os.fsencode(args[0]) | |||||||||
Show All 29 Lines | def change_prop(self, key: str, value: str) -> None: | |||||||||
""" | """ | |||||||||
if key == properties.PROP_EXTERNALS: | if key == properties.PROP_EXTERNALS: | |||||||||
logger.debug( | logger.debug( | |||||||||
"Setting '%s' property with value '%s' on path %s", | "Setting '%s' property with value '%s' on path %s", | |||||||||
key, | key, | |||||||||
value, | value, | |||||||||
self.path, | self.path, | |||||||||
) | ) | |||||||||
raise ValueError("Property '%s' detected. Not implemented yet." % key) | self.externals = {} | |||||||||
if value is not None: | ||||||||||
# externals are set on that directory path, parse and store them | ||||||||||
# for later processing in the close method | ||||||||||
for external in value.split("\n"): | ||||||||||
external = external.rstrip("\r") | ||||||||||
# skip empty line or comment | ||||||||||
if not external or external.startswith("#"): | ||||||||||
continue | ||||||||||
( | ||||||||||
path, | ||||||||||
external_url, | ||||||||||
revision, | ||||||||||
relative_url, | ||||||||||
) = parse_external_definition( | ||||||||||
external, os.fsdecode(self.path), self.svnrepo.origin_url | ||||||||||
) | ||||||||||
self.externals[path] = (external_url, revision, relative_url) | ||||||||||
if not self.externals: | ||||||||||
# externals might have been unset on that directory path, | ||||||||||
# remove associated paths from the reconstructed filesystem | ||||||||||
externals = self.dir_states[self.path].externals | ||||||||||
for path in externals.keys(): | ||||||||||
self.remove_external_path(os.fsencode(path)) | ||||||||||
self.dir_states[self.path].externals = {} | ||||||||||
def delete_entry(self, path: str, revision: int) -> None: | def delete_entry(self, path: str, revision: int) -> None: | |||||||||
"""Remove a path. | """Remove a path. | |||||||||
""" | """ | |||||||||
fullpath = os.path.join(self.rootpath, path.encode("utf-8")) | path_bytes = os.fsencode(path) | |||||||||
if path_bytes not in self.editor.external_paths: | ||||||||||
fullpath = os.path.join(self.rootpath, path_bytes) | ||||||||||
self.file_states.pop(fullpath, None) | self.file_states.pop(fullpath, None) | |||||||||
self.remove_child(path.encode("utf-8")) | self.remove_child(path_bytes) | |||||||||
def close(self): | def close(self): | |||||||||
"""Function called when we finish walking a repository. | """Function called when we finish processing a repository. | |||||||||
vlorentz: Why are they processed by a function called `close`? | ||||||||||
Done Inline ActionsOfficial subversion client processes externals after the export of all the files local to the repository. The code in that diff does something similar by processing them when all local modifications to anlambert: Official subversion client processes externals after the export of all the files local to the… | ||||||||||
SVN external definitions are processed by it. | ||||||||||
""" | """ | |||||||||
prev_externals = self.dir_states[self.path].externals | ||||||||||
if self.externals: | ||||||||||
# externals definition list might have changed in the current processed | ||||||||||
# revision, we need to determine if some were removed and delete the | ||||||||||
# associated paths | ||||||||||
old_externals = set(prev_externals) - set(self.externals) | ||||||||||
for old_external in old_externals: | ||||||||||
self.remove_external_path(os.fsencode(old_external)) | ||||||||||
# For each external, try to export it in reconstructed filesystem | ||||||||||
for path, (external_url, revision, relative_url) in self.externals.items(): | ||||||||||
external = (external_url, revision) | ||||||||||
dest_path = os.fsencode(path) | ||||||||||
dest_fullpath = os.path.join(self.path, dest_path) | ||||||||||
if ( | ||||||||||
path in prev_externals | ||||||||||
and prev_externals[path] == external | ||||||||||
and dest_fullpath in self.directory | ||||||||||
): | ||||||||||
# external already exported, nothing to do | ||||||||||
continue | ||||||||||
try: | ||||||||||
# try to export external in a temporary path, destination path could | ||||||||||
# be versioned and must be overridden only if the external URL is | ||||||||||
# still valid | ||||||||||
temp_dir = os.fsencode(tempfile.mkdtemp()) | ||||||||||
Not Done Inline Actionsare these directories removed? even if the loader crashes in this function. You could use the dir argument of tempfile.mkdtemp() to create them in a workdir you can clean at the end without keeping track of all paths vlorentz: are these directories removed? even if the loader crashes in this function.
You could use the… | ||||||||||
Done Inline ActionsI also noticed that yesterday, I have another diff coming that adds an external cache to avoid exporting them again and again when a same external URL is set on different directories. It will ensure that temp directories are properly removed at the end of the loading. anlambert: I also noticed that yesterday, I have another diff coming that adds an external cache to avoid… | ||||||||||
Done Inline Actionsanlambert: D6925 | ||||||||||
temp_path = os.path.join(temp_dir, dest_path) | ||||||||||
os.makedirs(b"/".join(temp_path.split(b"/")[:-1]), exist_ok=True) | ||||||||||
if external_url not in self.editor.dead_externals: | ||||||||||
logger.debug("Exporting external %s to path %s", external_url, path) | ||||||||||
self.svnrepo.client.export( | ||||||||||
external_url.rstrip("/"), | ||||||||||
to=temp_path, | ||||||||||
rev=revision, | ||||||||||
ignore_keywords=True, | ||||||||||
) | ||||||||||
self.editor.valid_externals[dest_fullpath] = ( | ||||||||||
external_url, | ||||||||||
relative_url, | ||||||||||
) | ||||||||||
except SubversionException as se: | ||||||||||
# external no longer available (404) | ||||||||||
logger.debug(se) | ||||||||||
self.editor.dead_externals.add(external_url) | ||||||||||
Not Done Inline Actionspass can be dropped as there are other instructions. ardumont: pass can be dropped as there are other instructions. | ||||||||||
Done Inline Actionsright anlambert: right | ||||||||||
# subversion export will always create the subdirectories of the external | ||||||||||
# path regardless the validity of the remote URL | ||||||||||
dest_path_split = dest_path.split(b"/") | ||||||||||
current_path = self.path | ||||||||||
self.add_directory(current_path) | ||||||||||
for subpath in dest_path_split[:-1]: | ||||||||||
current_path = os.path.join(current_path, subpath) | ||||||||||
self.add_directory(current_path) | ||||||||||
if os.path.exists(temp_path): | ||||||||||
# external successfully exported | ||||||||||
# remove previous path in from_disk model | ||||||||||
self.remove_child(dest_fullpath) | ||||||||||
# move exported path to reconstructed filesystem | ||||||||||
fullpath = os.path.join(self.rootpath, dest_fullpath) | ||||||||||
shutil.move(temp_path, fullpath) | ||||||||||
# update from_disk model and store external paths | ||||||||||
self.editor.external_paths.add(dest_fullpath) | ||||||||||
if os.path.isfile(fullpath): | ||||||||||
self.directory[dest_fullpath] = from_disk.Content.from_file( | ||||||||||
path=fullpath | ||||||||||
) | ||||||||||
else: | ||||||||||
self.directory[dest_fullpath] = from_disk.Directory.from_disk( | ||||||||||
path=fullpath | ||||||||||
) | ||||||||||
for root, dirs, files in os.walk(fullpath): | ||||||||||
self.editor.external_paths.update( | ||||||||||
[ | ||||||||||
os.path.join(root.replace(self.rootpath + b"/", b""), p) | ||||||||||
for p in chain(dirs, files) | ||||||||||
] | ||||||||||
) | ||||||||||
# ensure hash update for the directory with externals set | ||||||||||
self.directory[self.path].update_hash(force=True) | ||||||||||
# backup externals in directory state | ||||||||||
if self.externals: | ||||||||||
self.dir_states[self.path].externals = self.externals | ||||||||||
self.svnrepo.has_relative_externals = any( | ||||||||||
[relative_url for (_, relative_url) in self.editor.valid_externals.values()] | ||||||||||
) | ||||||||||
def remove_external_path(self, external_path: bytes) -> None: | ||||||||||
"""Remove a previously exported SVN external path from | ||||||||||
Not Done Inline Actions
ardumont: | ||||||||||
the reconstruted filesystem. | ||||||||||
""" | ||||||||||
fullpath = os.path.join(self.path, external_path) | ||||||||||
self.remove_child(fullpath) | ||||||||||
self.editor.external_paths.discard(fullpath) | ||||||||||
self.editor.valid_externals.pop(fullpath, None) | ||||||||||
for path in list(self.editor.external_paths): | ||||||||||
if path.startswith(fullpath + b"/"): | ||||||||||
self.editor.external_paths.remove(path) | ||||||||||
subpath_split = external_path.split(b"/")[:-1] | ||||||||||
for i in reversed(range(1, len(subpath_split) + 1)): | ||||||||||
# delete external sub-directory only if it is empty | ||||||||||
subpath = os.path.join(self.path, b"/".join(subpath_split[0:i])) | ||||||||||
if not os.listdir(os.path.join(self.rootpath, subpath)): | ||||||||||
self.remove_child(subpath) | ||||||||||
else: | ||||||||||
break | ||||||||||
try: | ||||||||||
# externals can overlap with versioned files so we must restore | ||||||||||
# them after removing the path above | ||||||||||
dest_path = os.path.join(self.rootpath, fullpath) | ||||||||||
self.svnrepo.client.export( | ||||||||||
svn_urljoin(self.svnrepo.remote_url, os.fsdecode(fullpath)), | ||||||||||
to=dest_path, | ||||||||||
rev=self.editor.revnum, | ||||||||||
ignore_keywords=True, | ||||||||||
) | ||||||||||
if os.path.isfile(dest_path): | ||||||||||
self.directory[fullpath] = from_disk.Content.from_file(path=dest_path) | ||||||||||
else: | ||||||||||
self.directory[fullpath] = from_disk.Directory.from_disk(path=dest_path) | ||||||||||
except SubversionException: | ||||||||||
Not Done Inline Actionswhat kind of error can arise here? ardumont: what kind of error can arise here?
Don't we want to log it? | ||||||||||
Done Inline ActionsThere is few chances that a versioned path overlaps an external but cases exist. We do not log the error here as it means there is no overlap between versioned path and external. anlambert: There is few chances that a versioned path overlaps an external but cases exist.
We do not log… | ||||||||||
pass | pass | |||||||||
class Editor: | class Editor: | |||||||||
"""Editor in charge of replaying svn events and computing objects | """Editor in charge of replaying svn events and computing objects | |||||||||
along. | along. | |||||||||
This implementation accounts for empty folder during hash | This implementation accounts for empty folder during hash | |||||||||
computations. | computations. | |||||||||
""" | """ | |||||||||
def __init__( | def __init__( | |||||||||
self, rootpath: bytes, directory: from_disk.Directory, svnrepo: SvnRepo | self, rootpath: bytes, directory: from_disk.Directory, svnrepo: SvnRepo | |||||||||
): | ): | |||||||||
self.rootpath = rootpath | self.rootpath = rootpath | |||||||||
self.directory = directory | self.directory = directory | |||||||||
self.file_states: Dict[bytes, FileState] = {} | self.file_states: Dict[bytes, FileState] = defaultdict(FileState) | |||||||||
self.dir_states: Dict[bytes, DirState] = defaultdict(DirState) | ||||||||||
self.external_paths: Set[bytes] = set() | ||||||||||
self.valid_externals: Dict[bytes, Tuple[str, bool]] = {} | ||||||||||
self.dead_externals: Set[str] = set() | ||||||||||
self.svnrepo = svnrepo | self.svnrepo = svnrepo | |||||||||
self.revnum = None | self.revnum = None | |||||||||
def set_target_revision(self, revnum) -> None: | def set_target_revision(self, revnum) -> None: | |||||||||
self.revnum = revnum | self.revnum = revnum | |||||||||
def abort(self) -> None: | def abort(self) -> None: | |||||||||
pass | pass | |||||||||
def close(self) -> None: | def close(self) -> None: | |||||||||
pass | pass | |||||||||
def open_root(self, base_revnum: int) -> DirEditor: | def open_root(self, base_revnum: int) -> DirEditor: | |||||||||
return DirEditor( | return DirEditor( | |||||||||
self.directory, | self.directory, | |||||||||
rootpath=self.rootpath, | rootpath=self.rootpath, | |||||||||
path=b"", | path=b"", | |||||||||
file_states=self.file_states, | file_states=self.file_states, | |||||||||
dir_states=self.dir_states, | ||||||||||
svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | |||||||||
) | ) | |||||||||
class Replay: | class Replay: | |||||||||
"""Replay class. | """Replay class. | |||||||||
""" | """ | |||||||||
▲ Show 20 Lines • Show All 122 Lines • Show Last 20 Lines |
Why are they processed by a function called close?