"""Utility class which adds more opcodes to

REMOVE, PATHCHANGE, BRANCHCHANGE, URLCHANGE = [

1 << x for x in range(UpdateProgress._num_op_codes, UpdateProgress._num_op_codes + 4)

]

_num_op_codes = UpdateProgress._num_op_codes + 4

"""A (virtual) root of all submodules in the given repository.

__slots__ = ()

k_root_name = "__ROOT__"

def __init__(self, repo: "Repo") -> None:

# repo, binsha, mode=None, path=None, name = None, parent_commit=None, url=None, ref=None)

super().__init__(

repo,

binsha=self.NULL_BIN_SHA,

mode=self.k_default_mode,

path="",

name=self.k_root_name,

parent_commit=repo.head.commit,

url="",

branch_path=git.Head.to_full_path(self.k_head_default),

)

def _clear_cache(self) -> None:

"""May not do anything."""

pass

# { Interface

def update( # type: ignore[override]

self,

previous_commit: Union[Commit_ish, str, None] = None,

recursive: bool = True,

force_remove: bool = False,

init: bool = True,

to_latest_revision: bool = False,

progress: Union[None, "RootUpdateProgress"] = None,

dry_run: bool = False,

force_reset: bool = False,

keep_going: bool = False,

no_fetch: bool = False,

) -> "RootModule":

"""Update the submodules of this repository to the current HEAD commit.

if self.repo.bare:

raise InvalidGitRepositoryError("Cannot update submodules in bare repositories")

# END handle bare

if progress is None:

progress = RootUpdateProgress()

# END ensure progress is set

prefix = ""

if dry_run:

prefix = "DRY-RUN: "

repo = self.repo

sms: "IterableList[Submodule]" = IterableList("name")

try:

# SETUP BASE COMMIT

###################

cur_commit = repo.head.commit

if previous_commit is None:

try:

previous_commit = repo.commit(repo.head.log_entry(-1).oldhexsha)

if previous_commit.binsha == previous_commit.NULL_BIN_SHA:

raise IndexError

# END handle initial commit

except IndexError:

# In new repositories, there is no previous commit.

previous_commit = cur_commit

# END exception handling

else:

previous_commit = repo.commit(previous_commit) # Obtain commit object.

# END handle previous commit

psms: "IterableList[Submodule]" = self.list_items(repo, parent_commit=previous_commit)

sms = self.list_items(repo)

spsms = set(psms)

ssms = set(sms)

# HANDLE REMOVALS

###################

rrsm = spsms - ssms

len_rrsm = len(rrsm)

for i, rsm in enumerate(rrsm):

op = REMOVE

if i == 0:

op |= BEGIN

# END handle begin

# Fake it into thinking its at the current commit to allow deletion

# of previous module. Trigger the cache to be updated before that.

progress.update(

op,

i,

len_rrsm,

prefix + "Removing submodule %r at %s" % (rsm.name, rsm.abspath),

)

rsm._parent_commit = repo.head.commit

rsm.remove(

configuration=False,

module=True,

force=force_remove,

dry_run=dry_run,

)

if i == len_rrsm - 1:

op |= END

# END handle end

progress.update(op, i, len_rrsm, prefix + "Done removing submodule %r" % rsm.name)

# END for each removed submodule

# HANDLE PATH RENAMES

#####################

# URL changes + branch changes.

csms = spsms & ssms

len_csms = len(csms)

for i, csm in enumerate(csms):

psm: "Submodule" = psms[csm.name]

sm: "Submodule" = sms[csm.name]

# PATH CHANGES

##############

if sm.path != psm.path and psm.module_exists():

progress.update(

BEGIN | PATHCHANGE,

i,

len_csms,

prefix + "Moving repository of submodule %r from %s to %s" % (sm.name, psm.abspath, sm.abspath),

)

# Move the module to the new path.

if not dry_run:

psm.move(sm.path, module=True, configuration=False)

# END handle dry_run

progress.update(

END | PATHCHANGE,

i,

len_csms,

prefix + "Done moving repository of submodule %r" % sm.name,

)

# END handle path changes

if sm.module_exists():

# HANDLE URL CHANGE

###################

if sm.url != psm.url:

# When fetching, add the new remote and remove the old one.

# This way, if the url just changes, the commits will not have

# to be re-retrieved.

nn = "__new_origin__"

smm = sm.module()

rmts = smm.remotes

# Don't do anything if we already have the url we search in

# place.

if len([r for r in rmts if r.url == sm.url]) == 0:

progress.update(

BEGIN | URLCHANGE,

i,

len_csms,

prefix + "Changing url of submodule %r from %s to %s" % (sm.name, psm.url, sm.url),

)

if not dry_run:

previous_remote = None

for remote in rmts:

if remote.url == psm.url:

previous_remote = remote

break

# END if urls match

# END for each remote

# If we didn't find a matching remote, but have exactly

# one, we can safely use this one.

if previous_remote is None:

if len(rmts) == 1:

previous_remote = rmts[0]

else:

# If we have not found any remote with the

# original URL we may not have a name. This is a

# special case, and its okay to fail here.

# Alternatively we could just generate a unique

# name and leave all existing ones in place.

raise InvalidGitRepositoryError(

"Couldn't find original remote-repo at url %r" % psm.url

)

# END handle one single remote

# END handle check we found a remote

if no_fetch:

# A new remote would have no cached refs. Preserve

# the existing refs and tracking configuration for

# offline updates instead of replacing the remote.

previous_remote.set_url(git.Git.polish_url(sm.url, expand_vars=False))

else:

assert nn not in [r.name for r in rmts]

smr = smm.create_remote(nn, sm.url)

smr.fetch(progress=progress)

# If we have a tracking branch, it should be available

# in the new remote as well.

if len([r for r in smr.refs if r.remote_head == sm.branch_name]) == 0:

raise ValueError(

"Submodule branch named %r was not available in new submodule remote at %r"

% (sm.branch_name, sm.url)

)

# END head is not detached

orig_name = previous_remote.name

smm.delete_remote(previous_remote)

# NOTE: Currently we leave tags from the deleted remotes

# as well as separate tracking branches in the possibly

# totally changed repository (someone could have changed

# the url to another project). At some point, one might

# want to clean it up, but the danger is high to remove

# stuff the user has added explicitly.

# Rename the new remote back to what it was.

smr.rename(orig_name)

# Early on, we verified that the our current tracking

# branch exists in the remote. Now we have to ensure

# that the sha we point to is still contained in the new

# remote tracking branch.

smsha = sm.binsha

found = False

rref = smr.refs[self.branch_name]

for c in rref.commit.traverse():

if c.binsha == smsha:

found = True

break

# END traverse all commits in search for sha

# END for each commit

if not found:

# Adjust our internal binsha to use the one of the

# remote this way, it will be checked out in the

# next step. This will change the submodule relative

# to us, so the user will be able to commit the

# change easily.

_logger.warning(

"Current sha %s was not contained in the tracking\

branch at the new remote, setting it the the remote's tracking branch",

sm.hexsha,

)

sm.binsha = rref.commit.binsha

# END reset binsha

# NOTE: All checkout is performed by the base

# implementation of update.

# END handle dry_run

progress.update(

END | URLCHANGE,

i,

len_csms,

prefix + "Done adjusting url of submodule %r" % (sm.name),

)

# END skip remote handling if new url already exists in module

# END handle url

# HANDLE PATH CHANGES

#####################

if sm.branch_path != psm.branch_path:

# Finally, create a new tracking branch which tracks the new

# remote branch.

progress.update(

BEGIN | BRANCHCHANGE,

i,

len_csms,

prefix

+ "Changing branch of submodule %r from %s to %s"

% (sm.name, psm.branch_path, sm.branch_path),

)

if not dry_run:

smm = sm.module()

smmr = smm.remotes

# As the branch might not exist yet, fetch all remotes

# unless restricted to locally cached refs.

if not no_fetch:

for remote in smmr:

remote.fetch(progress=progress)

# END for each remote

try:

tbr = git.Head.create(

smm,

sm.branch_name,

logmsg="branch: Created from HEAD",

)

except OSError:

# ...or reuse the existing one.

tbr = git.Head(smm, sm.branch_path)

# END ensure tracking branch exists

tbr.set_tracking_branch(find_first_remote_branch(smmr, sm.branch_name))

# NOTE: All head-resetting is done in the base

# implementation of update but we will have to checkout the

# new branch here. As it still points to the currently

# checked out commit, we don't do any harm.

# As we don't want to update working-tree or index, changing

# the ref is all there is to do.

smm.head.reference = tbr

# END handle dry_run

progress.update(

END | BRANCHCHANGE,

i,

len_csms,

prefix + "Done changing branch of submodule %r" % sm.name,

)

# END handle branch

# END handle

# END for each common submodule

except Exception as err:

if not keep_going:

raise

_logger.error(str(err))

# END handle keep_going

# FINALLY UPDATE ALL ACTUAL SUBMODULES

######################################

for sm in sms:

# Update the submodule using the default method.

sm.update(

recursive=False,

init=init,

to_latest_revision=to_latest_revision,

progress=progress,

dry_run=dry_run,

force=force_reset,

keep_going=keep_going,

no_fetch=no_fetch,

)

# Update recursively depth first - question is which inconsistent state will

# be better in case it fails somewhere. Defective branch or defective depth.

# The RootSubmodule type will never process itself, which was done in the

# previous expression.

if recursive:

# The module would exist by now if we are not in dry_run mode.

if sm.module_exists():

type(self)(sm.module()).update(

recursive=True,

force_remove=force_remove,

init=init,

to_latest_revision=to_latest_revision,

progress=progress,

dry_run=dry_run,

force_reset=force_reset,

keep_going=keep_going,

no_fetch=no_fetch,

)

# END handle dry_run

# END handle recursive

# END for each submodule to update

return self

def module(self) -> "Repo":

""":return: The actual repository containing the submodules"""

return self.repo