"""Named tuple allowing easy access to the revlog data fields"""

_re_hexsha_only = re.compile('^[0-9A-Fa-f]{40}$')

__slots__ = ()

def __repr__(self) -> str:

"""Representation of ourselves in git reflog format"""

return self.format()

def format(self) -> str:

""":return: a string suitable to be placed in a reflog file"""

act = self.actor

time = self.time

return "{} {} {} <{}> {!s} {}\t{}\n".format(self.oldhexsha,

self.newhexsha,

act.name,

act.email,

time[0],

altz_to_utctz_str(time[1]),

self.message)

@property

def oldhexsha(self) -> str:

"""The hexsha to the commit the ref pointed to before the change"""

return self[0]

@property

def newhexsha(self) -> str:

"""The hexsha to the commit the ref now points to, after the change"""

return self[1]

@property

def actor(self) -> Actor:

"""Actor instance, providing access"""

return self[2]

@property

def time(self) -> Tuple[int, int]:

"""time as tuple:

return self[3]

@property

def message(self) -> str:

"""Message describing the operation that acted on the reference"""

return self[4]

@classmethod

def new(cls, oldhexsha: str, newhexsha: str, actor: Actor, time: int, tz_offset: int, message: str

) -> 'RefLogEntry': # skipcq: PYL-W0621

""":return: New instance of a RefLogEntry"""

if not isinstance(actor, Actor):

raise ValueError("Need actor instance, got %s" % actor)

# END check types

return RefLogEntry((oldhexsha, newhexsha, actor, (time, tz_offset), message))

@classmethod

def from_line(cls, line: bytes) -> 'RefLogEntry':

""":return: New RefLogEntry instance from the given revlog line.

line_str = line.decode(defenc)

fields = line_str.split('\t', 1)

if len(fields) == 1:

info, msg = fields[0], None

elif len(fields) == 2:

info, msg = fields

else:

raise ValueError("Line must have up to two TAB-separated fields."

" Got %s" % repr(line_str))

# END handle first split

oldhexsha = info[:40]

newhexsha = info[41:81]

for hexsha in (oldhexsha, newhexsha):

if not cls._re_hexsha_only.match(hexsha):

raise ValueError("Invalid hexsha: %r" % (hexsha,))

# END if hexsha re doesn't match

# END for each hexsha

email_end = info.find('>', 82)

if email_end == -1:

raise ValueError("Missing token: >")

# END handle missing end brace

actor = Actor._from_string(info[82:email_end + 1])

time, tz_offset = parse_date(

info[email_end + 2:]) # skipcq: PYL-W0621

"""A reflog contains RefLogEntrys, each of which defines a certain state

__slots__ = ('_path', )

def __new__(cls, filepath: Union[PathLike, None] = None) -> 'RefLog':

inst = super(RefLog, cls).__new__(cls)

return inst

def __init__(self, filepath: Union[PathLike, None] = None):

"""Initialize this instance with an optional filepath, from which we will

self._path = filepath

if filepath is not None:

self._read_from_file()

# END handle filepath

def _read_from_file(self) -> None:

try:

fmap = file_contents_ro_filepath(

self._path, stream=True, allow_mmap=True)

except OSError:

# it is possible and allowed that the file doesn't exist !

return

# END handle invalid log

try:

self._deserialize(fmap)

finally:

fmap.close()

# END handle closing of handle

# { Interface

@classmethod

def from_file(cls, filepath: PathLike) -> 'RefLog':

"""

return cls(filepath)

@classmethod

def path(cls, ref: 'SymbolicReference') -> str:

"""

return osp.join(ref.repo.git_dir, "logs", to_native_path(ref.path))

@classmethod

def iter_entries(cls, stream: Union[str, 'BytesIO', mmap]) -> Iterator[RefLogEntry]:

"""

new_entry = RefLogEntry.from_line

if isinstance(stream, str):

# default args return mmap on py>3

_stream = file_contents_ro_filepath(stream)

assert isinstance(_stream, mmap)

else:

_stream = stream

# END handle stream type

while True:

line = _stream.readline()

if not line:

return

yield new_entry(line.strip())

# END endless loop

@classmethod

def entry_at(cls, filepath: PathLike, index: int) -> 'RefLogEntry':

"""

with open(filepath, 'rb') as fp:

if index < 0:

return RefLogEntry.from_line(fp.readlines()[index].strip())

# read until index is reached

for i in range(index + 1):

line = fp.readline()

if not line:

raise IndexError(

f"Index file ended at line {i+1}, before given index was reached")

# END abort on eof

# END handle runup

return RefLogEntry.from_line(line.strip())

# END handle index

def to_file(self, filepath: PathLike) -> None:

"""Write the contents of the reflog instance to a file at the given filepath.

lfd = LockedFD(filepath)

assure_directory_exists(filepath, is_file=True)

fp = lfd.open(write=True, stream=True)

try:

self._serialize(fp)

lfd.commit()

except Exception:

# on failure it rolls back automatically, but we make it clear

lfd.rollback()

raise

# END handle change

@classmethod

def append_entry(cls, config_reader: Union[Actor, 'GitConfigParser', 'SectionConstraint', None],

filepath: PathLike, oldbinsha: bytes, newbinsha: bytes, message: str,

write: bool = True) -> 'RefLogEntry':

"""Append a new log entry to the revlog at filepath.

if len(oldbinsha) != 20 or len(newbinsha) != 20:

raise ValueError("Shas need to be given in binary format")

# END handle sha type

assure_directory_exists(filepath, is_file=True)

first_line = message.split('\n')[0]

if isinstance(config_reader, Actor):

committer = config_reader # mypy thinks this is Actor | Gitconfigparser, but why?

else:

committer = Actor.committer(config_reader)

entry = RefLogEntry((

bin_to_hex(oldbinsha).decode('ascii'),

bin_to_hex(newbinsha).decode('ascii'),

committer, (int(_time.time()), _time.altzone), first_line

))

if write:

lf = LockFile(filepath)

lf._obtain_lock_or_raise()

fd = open(filepath, 'ab')

try:

fd.write(entry.format().encode(defenc))

finally:

fd.close()

lf._release_lock()

# END handle write operation

return entry

def write(self) -> 'RefLog':

"""Write this instance's data to the file we are originating from

if self._path is None:

raise ValueError(

"Instance was not initialized with a path, use to_file(...) instead")

# END assert path

self.to_file(self._path)

return self

# } END interface

# { Serializable Interface

def _serialize(self, stream: 'BytesIO') -> 'RefLog':

write = stream.write

# write all entries

for e in self:

write(e.format().encode(defenc))

# END for each entry

return self

def _deserialize(self, stream: 'BytesIO') -> 'RefLog':

self.extend(self.iter_entries(stream))

# } END serializable interface