Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/objects/util.py: 40%

3# This module is part of GitPython and is released under the

4# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/

6"""Utility functions for working with git objects."""

8__all__ = [

9 "get_object_type_by_name",

10 "parse_date",

11 "parse_actor_and_date",

12 "ProcessStreamAdapter",

13 "Traversable",

14 "altz_to_utctz_str",

15 "utctz_to_altz",

16 "verify_utctz",

17 "Actor",

18 "tzoffset",

19 "utc",

20]

22from abc import ABC, abstractmethod

23import calendar

24from collections import deque

25from datetime import datetime, timedelta, tzinfo

26import re

27from string import digits

28import time

29import warnings

31from git.util import Actor, IterableList, IterableObj

33# typing ------------------------------------------------------------

35from typing import (

36 Any,

37 Callable,

38 Deque,

39 Iterator,

40 NamedTuple,

41 Sequence,

42 TYPE_CHECKING,

43 Tuple,

44 Type,

45 TypeVar,

46 Union,

47 cast,

48 overload,

49)

51from git.types import Has_id_attribute, Literal

53if TYPE_CHECKING:

54 from io import BytesIO, StringIO

55 from subprocess import Popen

57 from git.types import Protocol, runtime_checkable

59 from .blob import Blob

60 from .commit import Commit

61 from .submodule.base import Submodule

62 from .tag import TagObject

63 from .tree import TraversedTreeTup, Tree

64else:

65 Protocol = ABC

67 def runtime_checkable(f):

68 return f

71class TraverseNT(NamedTuple):

72 depth: int

73 item: Union["Traversable", "Blob"]

74 src: Union["Traversable", None]

77T_TIobj = TypeVar("T_TIobj", bound="TraversableIterableObj") # For TraversableIterableObj.traverse()

79TraversedTup = Union[

80 Tuple[Union["Traversable", None], "Traversable"], # For Commit, Submodule.

81 "TraversedTreeTup", # For Tree.traverse().

82]

84# --------------------------------------------------------------------

86ZERO = timedelta(0)

88# { Functions

91def mode_str_to_int(modestr: Union[bytes, str]) -> int:

92 """Convert mode bits from an octal mode string to an integer mode for git.

94 :param modestr:

95 String like ``755`` or ``644`` or ``100644`` - only the last 6 chars will be

96 used.

98 :return:

99 String identifying a mode compatible to the mode methods ids of the :mod:`stat`

100 module regarding the rwx permissions for user, group and other, special flags

101 and file system flags, such as whether it is a symlink.

102 """

103 if isinstance(modestr, bytes):

104 # A byte is an int. int(b"4"[0]) is 52, so b"100644" became 0o6767524.

105 modestr = modestr.decode("ascii")

106 mode = 0

107 for iteration, char in enumerate(reversed(modestr[-6:])):

108 mode += int(char) << iteration * 3

109 # END for each char

110 return mode

111

112

113def get_object_type_by_name(

114 object_type_name: bytes,

115) -> Union[Type["Commit"], Type["TagObject"], Type["Tree"], Type["Blob"]]:

116 """Retrieve the Python class GitPython uses to represent a kind of Git object.

117

118 :return:

119 A type suitable to handle the given as `object_type_name`.

120 This type can be called create new instances.

121

122 :param object_type_name:

123 Member of :attr:`Object.TYPES <git.objects.base.Object.TYPES>`.

124

125 :raise ValueError:

126 If `object_type_name` is unknown.

127 """

128 if object_type_name == b"commit":

129 from . import commit

130

131 return commit.Commit

132 elif object_type_name == b"tag":

133 from . import tag

134

135 return tag.TagObject

136 elif object_type_name == b"blob":

137 from . import blob

138

139 return blob.Blob

140 elif object_type_name == b"tree":

141 from . import tree

142

143 return tree.Tree

144 else:

145 raise ValueError("Cannot handle unknown object type: %s" % object_type_name.decode())

146

147

148def utctz_to_altz(utctz: str) -> int:

149 """Convert a git timezone offset into a timezone offset west of UTC in seconds

150 (compatible with :attr:`time.altzone`).

151

152 :param utctz:

153 git utc timezone string, e.g. +0200

154 """

155 int_utctz = int(utctz)

156 seconds = (abs(int_utctz) // 100) * 3600 + (abs(int_utctz) % 100) * 60

157 return seconds if int_utctz < 0 else -seconds

158

159

160def altz_to_utctz_str(altz: float) -> str:

161 """Convert a timezone offset west of UTC in seconds into a Git timezone offset

162 string.

163

164 :param altz:

165 Timezone offset in seconds west of UTC.

166 """

167 hours = abs(altz) // 3600

168 minutes = (abs(altz) % 3600) // 60

169 sign = "-" if altz >= 60 else "+"

170 return "{}{:02}{:02}".format(sign, hours, minutes)

171

172

173def verify_utctz(offset: str) -> str:

174 """

175 :raise ValueError:

176 If `offset` is incorrect.

177

178 :return:

179 `offset`

180 """

181 fmt_exc = ValueError("Invalid timezone offset format: %s" % offset)

182 if len(offset) != 5:

183 raise fmt_exc

184 if offset[0] not in "+-":

185 raise fmt_exc

186 if offset[1] not in digits or offset[2] not in digits or offset[3] not in digits or offset[4] not in digits:

187 raise fmt_exc

188 # END for each char

189 return offset

190

191

192class tzoffset(tzinfo):

193 def __init__(self, secs_west_of_utc: float, name: Union[None, str] = None) -> None:

194 self._offset = timedelta(seconds=-secs_west_of_utc)

195 self._name = name or "fixed"

196

197 def __reduce__(self) -> Tuple[Type["tzoffset"], Tuple[float, str]]:

198 return tzoffset, (-self._offset.total_seconds(), self._name)

199

200 def utcoffset(self, dt: Union[datetime, None]) -> timedelta:

201 return self._offset

202

203 def tzname(self, dt: Union[datetime, None]) -> str:

204 return self._name

205

206 def dst(self, dt: Union[datetime, None]) -> timedelta:

207 return ZERO

208

209

210utc = tzoffset(0, "UTC")

211

212

213def from_timestamp(timestamp: float, tz_offset: float) -> datetime:

214 """Convert a `timestamp` + `tz_offset` into an aware :class:`~datetime.datetime`

215 instance."""

216 utc_dt = datetime.fromtimestamp(timestamp, utc)

217 try:

218 local_dt = utc_dt.astimezone(tzoffset(tz_offset))

219 return local_dt

220 except ValueError:

221 return utc_dt

222

223

224def parse_date(string_date: Union[str, datetime]) -> Tuple[int, int]:

225 """Parse the given date as one of the following:

226

227 * Aware datetime instance

228 * Git internal format: timestamp offset

229 * :rfc:`2822`: ``Thu, 07 Apr 2005 22:13:13 +0200``

230 * ISO 8601: ``2005-04-07T22:13:13`` - The ``T`` can be a space as well.

231

232 :return:

233 Tuple(int(timestamp_UTC), int(offset)), both in seconds since epoch

234

235 :raise ValueError:

236 If the format could not be understood.

237

238 :note:

239 Date can also be ``YYYY.MM.DD``, ``MM/DD/YYYY`` and ``DD.MM.YYYY``.

240 """

241 if isinstance(string_date, datetime):

242 if string_date.tzinfo:

243 utcoffset = cast(timedelta, string_date.utcoffset()) # typeguard, if tzinfoand is not None

244 offset = -int(utcoffset.total_seconds())

245 return int(string_date.astimezone(utc).timestamp()), offset

246 else:

247 raise ValueError(f"string_date datetime object without tzinfo, {string_date}")

248

249 # Git time

250 try:

251 if string_date.count(" ") == 1 and string_date.rfind(":") == -1:

252 timestamp, offset_str = string_date.split()

253 if timestamp.startswith("@"):

254 timestamp = timestamp[1:]

255 timestamp_int = int(timestamp)

256 return timestamp_int, utctz_to_altz(verify_utctz(offset_str))

257 else:

258 offset_str = "+0000" # Local time by default.

259 if string_date[-5] in "-+":

260 offset_str = verify_utctz(string_date[-5:])

261 string_date = string_date[:-6] # skip space as well

262 # END split timezone info

263 offset = utctz_to_altz(offset_str)

264

265 # Now figure out the date and time portion - split time.

266 date_formats = []

267 splitter = -1

268 if "," in string_date:

269 date_formats.append("%a, %d %b %Y")

270 splitter = string_date.rfind(" ")

271 else:

272 # ISO plus additional

273 date_formats.append("%Y-%m-%d")

274 date_formats.append("%Y.%m.%d")

275 date_formats.append("%m/%d/%Y")

276 date_formats.append("%d.%m.%Y")

277

278 splitter = string_date.rfind("T")

279 if splitter == -1:

280 splitter = string_date.rfind(" ")

281 # END handle 'T' and ' '

282 # END handle RFC or ISO

283

284 assert splitter > -1

285

286 # Split date and time.

287 time_part = string_date[splitter + 1 :] # Skip space.

288 date_part = string_date[:splitter]

289

290 # Parse time.

291 tstruct = time.strptime(time_part, "%H:%M:%S")

292

293 for fmt in date_formats:

294 try:

295 dtstruct = time.strptime(date_part, fmt)

296 utctime = calendar.timegm(

297 (

298 dtstruct.tm_year,

299 dtstruct.tm_mon,

300 dtstruct.tm_mday,

301 tstruct.tm_hour,

302 tstruct.tm_min,

303 tstruct.tm_sec,

304 dtstruct.tm_wday,

305 dtstruct.tm_yday,

306 tstruct.tm_isdst,

307 )

308 )

309 return int(utctime), offset

310 except ValueError:

311 continue

312 # END exception handling

313 # END for each fmt

314

315 # Still here ? fail.

316 raise ValueError("no format matched")

317 # END handle format

318 except Exception as e:

319 raise ValueError(f"Unsupported date format or type: {string_date}, type={type(string_date)}") from e

320 # END handle exceptions

321

322

323# Check the line ending once, before parsing fields, to avoid repeated backtracking.

324# Keep the field name from consuming spaces belonging to the actor.

325_re_actor_epoch = re.compile(r"^(?=[^\n]*$).[^ \n]* (.*) (\d+) ([+-]\d+)")

326_re_only_actor = re.compile(r"^.[^ \n]* (.*)$")

327

328

329def parse_actor_and_date(line: str) -> Tuple[Actor, int, int]:

330 """Parse out the actor (author or committer) info from a line like::

331

332 author Tom Preston-Werner <tom@mojombo.com> 1191999972 -0700

333

334 :return:

335 [Actor, int_seconds_since_epoch, int_timezone_offset]

336 """

337 actor, epoch, offset = "", "0", "0"

338 m = _re_actor_epoch.search(line)

339 if m:

340 actor, epoch, offset = m.groups()

341 else:

342 m = _re_only_actor.search(line)

343 actor = m.group(1) if m else line or ""

344 return (Actor._from_string(actor), int(epoch), utctz_to_altz(offset))

347# } END functions

350# { Classes

353class ProcessStreamAdapter:

354 """Class wiring all calls to the contained Process instance.

355

356 Use this type to hide the underlying process to provide access only to a specified

357 stream. The process is usually wrapped into an :class:`~git.cmd.Git.AutoInterrupt`

358 class to kill it if the instance goes out of scope.

359 """

360

361 __slots__ = ("_proc", "_stream")

362

363 def __init__(self, process: "Popen", stream_name: str) -> None:

364 self._proc = process

365 self._stream: StringIO = getattr(process, stream_name) # guessed type

366

367 def __getattr__(self, attr: str) -> Any:

368 return getattr(self._stream, attr)

369

370

371@runtime_checkable

372class Traversable(Protocol):

373 """Simple interface to perform depth-first or breadth-first traversals in one

374 direction.

375

376 Subclasses only need to implement one function.

377

378 Instances of the subclass must be hashable.

379

380 Defined subclasses:

381

382 * :class:`Commit <git.objects.Commit>`

383 * :class:`Tree <git.objects.tree.Tree>`

384 * :class:`Submodule <git.objects.submodule.base.Submodule>`

385 """

386

387 __slots__ = ()

388

389 @classmethod

390 @abstractmethod

391 def _get_intermediate_items(cls, item: Any) -> Sequence["Traversable"]:

392 """

393 :return:

394 Tuple of items connected to the given item.

395 Must be implemented in subclass.

396

397 class Commit:: (cls, Commit) -> Tuple[Commit, ...]

398 class Submodule:: (cls, Submodule) -> Iterablelist[Submodule]

399 class Tree:: (cls, Tree) -> Tuple[Tree, ...]

400 """

401 raise NotImplementedError("To be implemented in subclass")

402

403 @abstractmethod

404 def list_traverse(self, *args: Any, **kwargs: Any) -> Any:

405 """Traverse self and collect all items found.

406

407 Calling this directly on the abstract base class, including via a ``super()``

408 proxy, is deprecated. Only overridden implementations should be called.

409 """

410 warnings.warn(

411 "list_traverse() method should only be called from subclasses."

412 " Calling from Traversable abstract class will raise NotImplementedError in 4.0.0."

413 " The concrete subclasses in GitPython itself are 'Commit', 'RootModule', 'Submodule', and 'Tree'.",

414 DeprecationWarning,

415 stacklevel=2,

416 )

417 return self._list_traverse(*args, **kwargs)

418

419 def _list_traverse(

420 self, as_edge: bool = False, *args: Any, **kwargs: Any

421 ) -> IterableList[Union["Commit", "Submodule", "Tree", "Blob"]]:

422 """Traverse self and collect all items found.

423

424 :return:

425 :class:`~git.util.IterableList` with the results of the traversal as

426 produced by :meth:`traverse`::

427

428 Commit -> IterableList[Commit]

429 Submodule -> IterableList[Submodule]

430 Tree -> IterableList[Union[Submodule, Tree, Blob]]

431 """

432 # Commit and Submodule have id.__attribute__ as IterableObj.

433 # Tree has id.__attribute__ inherited from IndexObject.

434 if isinstance(self, Has_id_attribute):

435 id = self._id_attribute_

436 else:

437 # Shouldn't reach here, unless Traversable subclass created with no

438 # _id_attribute_.

439 id = ""

440 # Could add _id_attribute_ to Traversable, or make all Traversable also

441 # Iterable?

442

443 if not as_edge:

444 out: IterableList[Union["Commit", "Submodule", "Tree", "Blob"]] = IterableList(id)

445 out.extend(self.traverse(as_edge=as_edge, *args, **kwargs)) # noqa: B026

446 return out

447 # Overloads in subclasses (mypy doesn't allow typing self: subclass).

448 # Union[IterableList['Commit'], IterableList['Submodule'], IterableList[Union['Submodule', 'Tree', 'Blob']]]

449 else:

450 # Raise DeprecationWarning, it doesn't make sense to use this.

451 out_list: IterableList = IterableList(self.traverse(*args, **kwargs))

452 return out_list

453

454 @abstractmethod

455 def traverse(self, *args: Any, **kwargs: Any) -> Any:

456 """Iterator yielding items found when traversing self.

457

458 Calling this directly on the abstract base class, including via a ``super()``

459 proxy, is deprecated. Only overridden implementations should be called.

460 """

461 warnings.warn(

462 "traverse() method should only be called from subclasses."

463 " Calling from Traversable abstract class will raise NotImplementedError in 4.0.0."

464 " The concrete subclasses in GitPython itself are 'Commit', 'RootModule', 'Submodule', and 'Tree'.",

465 DeprecationWarning,

466 stacklevel=2,

467 )

468 return self._traverse(*args, **kwargs)

469

470 def _traverse(

471 self,

472 predicate: Callable[[Union["Traversable", "Blob", TraversedTup], int], bool] = lambda i, d: True,

473 prune: Callable[[Union["Traversable", "Blob", TraversedTup], int], bool] = lambda i, d: False,

474 depth: int = -1,

475 branch_first: bool = True,

476 visit_once: bool = True,

477 ignore_self: int = 1,

478 as_edge: bool = False,

479 ) -> Union[Iterator[Union["Traversable", "Blob"]], Iterator[TraversedTup]]:

480 """Iterator yielding items found when traversing `self`.

481

482 :param predicate:

483 A function ``f(i,d)`` that returns ``False`` if item i at depth ``d`` should

484 not be included in the result.

485

486 :param prune:

487 A function ``f(i,d)`` that returns ``True`` if the search should stop at

488 item ``i`` at depth ``d``. Item ``i`` will not be returned.

489

490 :param depth:

491 Defines at which level the iteration should not go deeper if -1. There is no

492 limit if 0, you would effectively only get `self`, the root of the

493 iteration. If 1, you would only get the first level of

494 predecessors/successors.

495

496 :param branch_first:

497 If ``True``, items will be returned branch first, otherwise depth first.

498

499 :param visit_once:

500 If ``True``, items will only be returned once, although they might be

501 encountered several times. Loops are prevented that way.

502

503 :param ignore_self:

504 If ``True``, `self` will be ignored and automatically pruned from the

505 result. Otherwise it will be the first item to be returned. If `as_edge` is

506 ``True``, the source of the first edge is ``None``.

507

508 :param as_edge:

509 If ``True``, return a pair of items, first being the source, second the

510 destination, i.e. tuple(src, dest) with the edge spanning from source to

511 destination.

512

513 :return:

514 Iterator yielding items found when traversing `self`::

515

516 Commit -> Iterator[Union[Commit, Tuple[Commit, Commit]] Submodule ->

517 Iterator[Submodule, Tuple[Submodule, Submodule]] Tree ->

518 Iterator[Union[Blob, Tree, Submodule,

519 Tuple[Union[Submodule, Tree], Union[Blob, Tree,

520 Submodule]]]

521

522 ignore_self=True is_edge=True -> Iterator[item] ignore_self=True

523 is_edge=False --> Iterator[item] ignore_self=False is_edge=True ->

524 Iterator[item] | Iterator[Tuple[src, item]] ignore_self=False

525 is_edge=False -> Iterator[Tuple[src, item]]

526 """

527

528 visited = set()

529 stack: Deque[TraverseNT] = deque()

530 stack.append(TraverseNT(0, self, None)) # self is always depth level 0.

531

532 def addToStack(

533 stack: Deque[TraverseNT],

534 src_item: "Traversable",

535 branch_first: bool,

536 depth: int,

537 ) -> None:

538 lst = self._get_intermediate_items(item)

539 if not lst: # Empty list

540 return

541 if branch_first:

542 stack.extendleft(TraverseNT(depth, i, src_item) for i in lst)

543 else:

544 reviter = (TraverseNT(depth, lst[i], src_item) for i in range(len(lst) - 1, -1, -1))

545 stack.extend(reviter)

546

547 # END addToStack local method

548

549 while stack:

550 d, item, src = stack.pop() # Depth of item, item, item_source

551

552 if visit_once and item in visited:

553 continue

554

555 if visit_once:

556 visited.add(item)

557

558 rval: Union[TraversedTup, "Traversable", "Blob"]

559 if as_edge:

560 # If as_edge return (src, item) unless rrc is None

561 # (e.g. for first item).

562 rval = (src, item)

563 else:

564 rval = item

565

566 if prune(rval, d):

567 continue

568

569 skipStartItem = ignore_self and (item is self)

570 if not skipStartItem and predicate(rval, d):

571 yield rval

572

573 # Only continue to next level if this is appropriate!

574 next_d = d + 1

575 if depth > -1 and next_d > depth:

576 continue

577

578 addToStack(stack, item, branch_first, next_d)

579 # END for each item on work stack

580

581

582@runtime_checkable

583class Serializable(Protocol):

584 """Defines methods to serialize and deserialize objects from and into a data

585 stream."""

586

587 __slots__ = ()

588

589 # @abstractmethod

590 def _serialize(self, stream: "BytesIO") -> "Serializable":

591 """Serialize the data of this object into the given data stream.

592

593 :note:

594 A serialized object would :meth:`_deserialize` into the same object.

595

596 :param stream:

597 A file-like object.

598

599 :return:

600 self

601 """

602 raise NotImplementedError("To be implemented in subclass")

603

604 # @abstractmethod

605 def _deserialize(self, stream: "BytesIO") -> "Serializable":

606 """Deserialize all information regarding this object from the stream.

607

608 :param stream:

609 A file-like object.

610

611 :return:

612 self

613 """

614 raise NotImplementedError("To be implemented in subclass")

615

616

617class TraversableIterableObj(IterableObj, Traversable):

618 __slots__ = ()

619

620 TIobj_tuple = Tuple[Union[T_TIobj, None], T_TIobj]

621

622 def list_traverse(self: T_TIobj, *args: Any, **kwargs: Any) -> IterableList[T_TIobj]:

623 return super()._list_traverse(*args, **kwargs)

624

625 @overload

626 def traverse(self: T_TIobj) -> Iterator[T_TIobj]: ...

627

628 @overload

629 def traverse(

630 self: T_TIobj,

631 predicate: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

632 prune: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

633 depth: int,

634 branch_first: bool,

635 visit_once: bool,

636 ignore_self: Literal[True],

637 as_edge: Literal[False],

638 ) -> Iterator[T_TIobj]: ...

639

640 @overload

641 def traverse(

642 self: T_TIobj,

643 predicate: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

644 prune: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

645 depth: int,

646 branch_first: bool,

647 visit_once: bool,

648 ignore_self: Literal[False],

649 as_edge: Literal[True],

650 ) -> Iterator[Tuple[Union[T_TIobj, None], T_TIobj]]: ...

651

652 @overload

653 def traverse(

654 self: T_TIobj,

655 predicate: Callable[[Union[T_TIobj, TIobj_tuple], int], bool],

656 prune: Callable[[Union[T_TIobj, TIobj_tuple], int], bool],

657 depth: int,

658 branch_first: bool,

659 visit_once: bool,

660 ignore_self: Literal[True],

661 as_edge: Literal[True],

662 ) -> Iterator[Tuple[T_TIobj, T_TIobj]]: ...

663

664 def traverse(

665 self: T_TIobj,

666 predicate: Callable[[Union[T_TIobj, TIobj_tuple], int], bool] = lambda i, d: True,

667 prune: Callable[[Union[T_TIobj, TIobj_tuple], int], bool] = lambda i, d: False,

668 depth: int = -1,

669 branch_first: bool = True,

670 visit_once: bool = True,

671 ignore_self: int = 1,

672 as_edge: bool = False,

673 ) -> Union[Iterator[T_TIobj], Iterator[Tuple[T_TIobj, T_TIobj]], Iterator[TIobj_tuple]]:

674 """For documentation, see :meth:`Traversable._traverse`."""

675

676 ## To typecheck instead of using cast:

677 #

678 # import itertools

679 # from git.types import TypeGuard

680 # def is_commit_traversed(inp: Tuple) -> TypeGuard[Tuple[Iterator[Tuple['Commit', 'Commit']]]]:

681 # for x in inp[1]:

682 # if not isinstance(x, tuple) and len(x) != 2:

683 # if all(isinstance(inner, Commit) for inner in x):

684 # continue

685 # return True

686 #

687 # ret = super(Commit, self).traverse(predicate, prune, depth, branch_first, visit_once, ignore_self, as_edge)

688 # ret_tup = itertools.tee(ret, 2)

689 # assert is_commit_traversed(ret_tup), f"{[type(x) for x in list(ret_tup[0])]}"

690 # return ret_tup[0]

691

692 return cast(

693 Union[Iterator[T_TIobj], Iterator[Tuple[Union[None, T_TIobj], T_TIobj]]],

694 super()._traverse(

695 predicate, # type: ignore[arg-type]

696 prune, # type: ignore[arg-type]

697 depth,

698 branch_first,

699 visit_once,

700 ignore_self,

701 as_edge,

702 ),

703 )