diff --git a/domain/API.py b/domain/API.py index 6c471d6..d3fb222 100644 --- a/domain/API.py +++ b/domain/API.py @@ -7,9 +7,10 @@ class API: __DEFAULT_API_PATH = "/api/v1" - def __init__(self): + def __init__(self, verify_ssl, ca_bundle): - pass + self.__verify_ssl = verify_ssl + self.__ca_bundle = ca_bundle @staticmethod def _make_api_base_url(hostname, port): @@ -20,17 +21,23 @@ class API: return base - @staticmethod - def factory(hostname, port, token) -> gitea.Gitea: + def get(self, hostname, port, token) -> gitea.Gitea: url = API._make_api_base_url( hostname=hostname, port=port ) + ssl_verify_arg = True + if self.__verify_ssl is not None: + ssl_verify_arg = self.__verify_ssl + if self.__ca_bundle is not None: + ssl_verify_arg = self.__ca_bundle + g = gitea.Gitea( gitea_url=url, - token_text=token + token_text=token, + verify=ssl_verify_arg ) return g diff --git a/domain/Migrator.py b/domain/Migrator.py index 3b7d034..e6b7667 100644 --- a/domain/Migrator.py +++ b/domain/Migrator.py @@ -18,14 +18,13 @@ class Migrator: self, source_host, source_port, source_token, destination_host, destination_port, destination_token, + verify_ssl: bool = True, ca_bundle: str = None ): # noinspection PyTypeChecker self.__logger: logging.Logger = None self._init_logger() - self.__verify_ssl = True - self.__source_host = source_host self.__source_port = source_port self.__source_token = source_token @@ -34,15 +33,24 @@ class Migrator: self.__destination_port = destination_port self.__destination_token = destination_token - self.__source_api = API.factory( + self.__verify_ssl = verify_ssl + self.__ca_bundle = ca_bundle + + api = API( + verify_ssl=self.__verify_ssl, + ca_bundle=self.__ca_bundle, + ) + + self.__source_api = api.get( hostname=self.__source_host, port=self.__source_port, - token=self.__source_token + token=self.__source_token, + ) - self.__destination_api = API.factory( + self.__destination_api = api.get( hostname=self.__destination_host, port=self.__destination_port, - token=self.__destination_token + token=self.__destination_token, ) def _init_logger(self): @@ -111,10 +119,7 @@ class Migrator: return repo_name - def set_verify_ssl(self, b: bool): - - self.__verify_ssl = b - + """ def set_ca_bundle(self, bundle_path: str): self.__logger.info("Setting certificate bundle path") @@ -123,6 +128,10 @@ class Migrator: self.__logger.info(f"Old path: {certifi.where()}") certifi.core._CACERT_PATH = bundle_path self.__logger.info(f"New path: {certifi.where()}") + + # TODO: JUST TESTING + self.__verify_ssl = bundle_path + """ def migrate_entire_org( self, @@ -165,19 +174,21 @@ class Migrator: for repo in source_repos: repo: gitea.Repository self.__logger.info(f"- {repo.get_full_name()}") - assert False, "Poop" repos_migrate = [] repos_ignore = [] go_right_now = False for repo in source_repos: - repo: giteapy.Repository + repo: gitea.Repository while True: if interactive: - response = input(f"Migrate repo #{repo.id} \"{repo.full_name}\" ? (Y)es, (N)o, (G)o right now, (Q)uit ==> ") + response = input( + f"Migrate repo #{repo.id} \"{repo.full_name}\" ?" + " (Y)es, (N)o, (G)o right now, (Q)uit ==> " + ) response = response.lower() else: response = "y" @@ -205,28 +216,32 @@ class Migrator: if go_right_now: break - # + # Announce repo destination names self.__logger.info("") if len(repos_migrate): self.__logger.info("Repos to migrate:") for repo in repos_migrate: - repo: giteapy.Repository - destination_name = self._make_destination_repo_name(pattern=destination_repo_name, repo=repo) + repo: gitea.Repository + destination_name = self._make_destination_repo_name( + pattern=destination_repo_name, repo=repo + ) self.__logger.info( f"#{repo.id} \"{repo.name}\"\n> \"{destination_name}\"" ) else: self.__logger.info("No repos marked to migrate") + # Announce manually ignored repos self.__logger.info("") if len(repos_ignore): self.__logger.info("Repos to ignore:") for repo in repos_ignore: - repo: giteapy.Repository - self.__logger.info(f"#{repo.id} \"{repo.name}\"") + repo: gitea.Repository + self.__logger.info(f"#{repo.id} \"{repo.get_full_name()}\"") else: self.__logger.info("No repos marked to ignore") + # Migrate if len(repos_migrate): confirmation = input("Do you confirm the above selections? Enter MIGRATE ==> ") @@ -242,7 +257,8 @@ class Migrator: repos=repos_migrate ) self.__logger.info( - f"{len(source_repos_successful)} of {len(repos_migrate)} repos successfully migrated." + f"{len(source_repos_successful)} of {len(repos_migrate)}" + " repos successfully migrated." ) if len(source_repos_failed) > 0: self.__logger.error(f"Failed to migrate {len(source_repos_failed)} repos:") @@ -256,7 +272,10 @@ class Migrator: f"Failed to migrate repo: {repo.name}\n> {exception}" ) - self._delete_migrated_repos(source_org_name=source_org, repos=source_repos_successful) + self._delete_migrated_repos( + source_org_name=source_org, + repos=source_repos_successful + ) else: self.__logger.info("Confirmation not received; Won't do anything.") @@ -296,9 +315,7 @@ class Migrator: repo_key = repo.get_full_name() topics_present = repo.get_topics() - print(topics_present) - repo_topics[repo_key] = topics_present["topics"] - print(repo_topics[repo_key]) + repo_topics[repo_key] = topics_present if self._check_required_topics( topics_present=repo_topics[repo_key], @@ -334,9 +351,6 @@ class Migrator: @staticmethod def _check_required_topics(topics_present: list[str], topics_required: list[str]) -> bool: - print("Required topics:", topics_required) - print("Present topics:", topics_present) - for topic in topics_required: if topic not in topics_present: return False @@ -352,44 +366,75 @@ class Migrator: repos: list ): - api_source, api_destination = self._get_org_apis() + # api_source, api_destination = self._get_org_apis() + # destination_org = api_destination.org_get(org=destination_org_name) + # destination_org: giteapy.Organization - destination_org = api_destination.org_get(org=destination_org_name) - destination_org: giteapy.Organization - self.__logger.info(f"Destination organization: {destination_org.full_name}") - - api_source_repos = self._get_repo_api( - hostname=self.__source_host, port=self.__source_port, - token=self.__source_token + api_dest_org = gitea.Organization.request( + gitea=self.__destination_api, + name=destination_org_name ) + self.__logger.info(f"Destination organization: {api_dest_org.full_name}") + source_repos_successful = [] source_repos_failed = [] for source_repo in repos: - source_repo: giteapy.Repository + source_repo: gitea.Repository - this_destination_repo_name = destination_repo_name.replace("%N%", source_repo.name) - - self.__logger.info(f"Migrating: {source_repo.name} ==> {this_destination_repo_name}") - - source_repo_topics = api_source_repos.repo_list_topics(owner=source_repo.owner.login, repo=source_repo.name) - source_repo_topics = source_repo_topics.topics - - migrate_body = giteapy.MigrateRepoForm( - mirror=False, - clone_addr=source_repo.clone_url, - uid=destination_org.id, - private=source_repo.private, - repo_name=this_destination_repo_name, - description=source_repo.description, - labels=True, issues=True, pull_requests=True, releases=True, milestones=True, wiki=True + this_destination_repo_name = destination_repo_name.replace( + "%N%", + source_repo.name ) + + self.__logger.info( + f"Migrating: {source_repo.name} ==> {this_destination_repo_name}" + ) + + # source_repo_topics = api_source_repos.repo_list_topics( + # owner=source_repo.owner.login, repo=source_repo.name + # ) + # source_repo_topics = source_repo_topics.topics + topics = source_repo.get_topics() + + # migrate_body = giteapy.MigrateRepoForm( + # mirror=False, + # clone_addr=source_repo.clone_url, + # uid=destination_org.id, + # private=source_repo.private, + # repo_name=this_destination_repo_name, + # description=source_repo.description, + # labels=True, issues=True, pull_requests=True, releases=True, milestones=True, wiki=True + # ) # TODO: These three lines represent feature request to giteapy authors #migrate_body.auth_token = self.__source_token #migrate_body.swagger_types["auth_token"] = "str" #migrate_body.attribute_map["auth_token"] = "auth_token" + destination_repo = gitea.Repository.migrate_repo( + gitea=self.__destination_api, + service: str, + clone_addr: str, + repo_name: str, + description: str = "", + private: bool = False, + auth_token: str = None, + auth_username: str = None, + auth_password: str = None, + mirror: bool = False, + mirror_interval: str = None, + lfs: bool = False, + lfs_endpoint: str = "", + wiki: bool = False, + labels: bool = False, + issues: bool = False, + pull_requests: bool = False, + releases: bool = False, + milestones: bool = False, + repo_owner: str = None, + ) + self.__logger.debug("Migrate body:") self.__logger.debug(migrate_body) diff --git a/main.py b/main.py index fd4257a..9eac14f 100644 --- a/main.py +++ b/main.py @@ -138,13 +138,11 @@ def main(): source_token=args.source_token, destination_host=args.destination_hostname, destination_port=args.destination_port, - destination_token=args.destination_token + destination_token=args.destination_token, + verify_ssl=args.verify_ssl, + ca_bundle=args.ca_bundle ) - mig.set_verify_ssl(args.verify_ssl) - if args.ca_bundle: - mig.set_ca_bundle(args.ca_bundle) - mig.migrate_entire_org( interactive=args.interactive, source_org=args.source_org,