remove unused functions

Author: truib

scripts/automated_ingestion/eessi_task.py

@@ -219,178 +219,6 @@ def _determine_sequence_numbers_including_task_file(self, repo: str, pr: str) ->
                 return {}
         return sequence_numbers
 
-    @log_function_entry_exit()
-    def _find_highest_number(self, str_list: List[str]) -> int:
-        """
-        Find the highest number in a list of strings.
-        """
-        # Convert all strings to integers
-        int_list = [int(num) for num in str_list]
-        return max(int_list)
-
-    @log_function_entry_exit()
-    def _get_sequence_number_for_task_file(self) -> int:
-        """
-        Get the sequence number this task is assigned to at the moment.
-        NOTE, should only be called if the task is actually assigned to a sequence number.
-        """
-        repo_name = self.description.get_repo_name()
-        pr_number = self.description.get_pr_number()
-        sequence_numbers = self._determine_sequence_numbers_including_task_file(repo_name, pr_number)
-        if len(sequence_numbers) == 0:
-            raise ValueError("Found no sequence numbers at all")
-        else:
-            # get all entries with value True, there should be only one, so we return the first one
-            sequence_numbers_true = [key for key, value in sequence_numbers.items() if value is True]
-            if len(sequence_numbers_true) == 0:
-                raise ValueError("Found no sequence numbers that include the task file for task %s",
-                                 self.description)
-            else:
-                return sequence_numbers_true[0]
-
-    @log_function_entry_exit()
-    def _get_current_sequence_number(self, sequence_numbers: Dict[int, bool] = None) -> int:
-        """
-        Get the current sequence number based on the sequence numbers.
-        If sequence_numbers is not provided, we determine the sequence numbers from the task description.
-        """
-        if sequence_numbers is None:
-            repo_name = self.description.get_repo_name()
-            pr_number = self.description.get_pr_number()
-            sequence_numbers = self._determine_sequence_numbers_including_task_file(repo_name, pr_number)
-        if len(sequence_numbers) == 0:
-            return 0
-        return self._find_highest_number(sequence_numbers.keys())
-
-    @log_function_entry_exit()
-    def _get_fixed_sequence_number(self) -> int:
-        """
-        Get a fixed sequence number.
-        """
-        return 11
-
-    @log_function_entry_exit()
-    def _find_staging_pr(self) -> Tuple[Optional[PullRequest], Optional[str], Optional[int]]:
-        """
-        Find the staging PR for the task.
-        TODO: arg sequence number --> make function simpler
-        """
-        repo_name = self.description.get_repo_name()
-        pr_number = self.description.get_pr_number()
-        try:
-            sequence_number = self._get_sequence_number_for_task_file()
-        except ValueError:
-            # no sequence number found, so we return None
-            log_message(LoggingScope.ERROR, "ERROR", "no sequence number found for task '%s'", self.description)
-            return None, None, None
-        except Exception as err:
-            # some other error
-            log_message(LoggingScope.ERROR, "ERROR", "error finding staging PR for task '%s': '%s'",
-                        self.description, err)
-            return None, None, None
-        branch_name = f"{repo_name.replace('/', '-')}-PR-{pr_number}-SEQ-{sequence_number}"
-        if branch_name in [branch.name for branch in self.git_repo.get_branches()]:
-            find_pr = [pr for pr in self.git_repo.get_pulls(head=branch_name, state="all")]
-            if find_pr:
-                pr = find_pr.pop(0)
-                return pr, branch_name, sequence_number
-            else:
-                return None, branch_name, sequence_number
-        else:
-            return None, None, None
-
-    @log_function_entry_exit()
-    def _create_staging_pr(self, sequence_number: int) -> Tuple[PullRequest, str]:
-        """
-        Create a staging PR for the task.
-        NOTE, SHALL only be called if no staging PR for the task exists yet.
-        """
-        repo_name = self.description.get_repo_name()
-        pr_number = self.description.get_pr_number()
-        branch_name = f"{repo_name.replace('/', '-')}-PR-{pr_number}-SEQ-{sequence_number}"
-        default_branch_name = self.git_repo.default_branch
-        pr = self.git_repo.create_pull(title=f"Add task for {repo_name} PR {pr_number} seq {sequence_number}",
-                                       body=f"Add task for {repo_name} PR {pr_number} seq {sequence_number}",
-                                       head=branch_name, base=default_branch_name)
-        return pr, branch_name
-
-    @log_function_entry_exit()
-    def _find_state(self) -> EESSITaskState:
-        """
-        Determine the state of the task based on the task description metadata.
-
-        Returns:
-            The state of the task.
-        """
-        # obtain repo and pr from metadata
-        log_message(LoggingScope.TASK_OPS_DETAILS, "INFO", "finding state of task '%s'", self.description.task_object)
-        repo = self.description.get_repo_name()
-        pr = self.description.get_pr_number()
-        log_message(LoggingScope.TASK_OPS_DETAILS, "INFO", "repo: '%s', pr: '%s'", repo, pr)
-
-        # obtain all sequence numbers in repo/pr dir which include a state file for this task
-        sequence_numbers = self._determine_sequence_numbers_including_task_file(repo, pr)
-        if len(sequence_numbers) == 0:
-            # no sequence numbers found, so we return NEW_TASK
-            log_message(LoggingScope.TASK_OPS_DETAILS, "INFO", "no sequence numbers found, state: NEW_TASK")
-            return EESSITaskState.NEW_TASK
-        # we got at least one sequence number
-        # if one value for a sequence number is True, we can determine the state from the file in the directory
-        sequence_including_task = [key for key, value in sequence_numbers.items() if value is True]
-        if len(sequence_including_task) == 0:
-            # no sequence number includes the task file, so we return NEW_TASK
-            log_message(LoggingScope.TASK_OPS_DETAILS, "INFO",
-                        "no sequence number includes the task file, state: NEW_TASK")
-            return EESSITaskState.NEW_TASK
-        # we got at least one sequence number which includes the task file
-        # we can determine the state from the filename in the directory
-        # NOTE, we use the first element in sequence_including_task (there should be only one)
-        #     we ignore other elements in sequence_including_task
-        sequence_number = sequence_including_task[0]
-        task_file_name = self.description.get_task_file_name()
-        metadata_file_state_path_prefix = f"{repo}/{pr}/{sequence_number}/{task_file_name}."
-        state = self._get_state_for_metadata_file_prefix(metadata_file_state_path_prefix, sequence_number)
-        log_message(LoggingScope.TASK_OPS_DETAILS, "INFO", "state: '%s'", state)
-        return state
-
-    @log_function_entry_exit()
-    def _get_state_for_metadata_file_prefix(self, metadata_file_state_path_prefix: str,
-                                            sequence_number: int) -> EESSITaskState:
-        """
-        Get the state from the file in the metadata_file_state_path_prefix.
-        """
-        # depending on the state of the deployment (NEW_TASK, PAYLOAD_STAGED, PULL_REQUEST, APPROVED, REJECTED,
-        # INGESTED, DONE)
-        # we need to check the task file in the default branch or in the branch corresponding to the sequence number
-        directory_part = os.path.dirname(metadata_file_state_path_prefix)
-        repo_name = self.description.get_repo_name()
-        pr_number = self.description.get_pr_number()
-        default_branch_name = self.git_repo.default_branch
-        branch_name = f"{repo_name.replace('/', '-')}-PR-{pr_number}-SEQ-{sequence_number}"
-        all_branch_names = [branch.name for branch in self.git_repo.get_branches()]
-        states = []
-        for branch in [default_branch_name, branch_name]:
-            if branch in all_branch_names:
-                # first get all files in directory part of metadata_file_state_path_prefix
-                files = self._list_directory_contents(directory_part, branch)
-                # check if any of the files has metadata_file_state_path_prefix as prefix
-                for file in files:
-                    if file.path.startswith(metadata_file_state_path_prefix):
-                        # get state from file name taking only the suffix
-                        state = EESSITaskState.from_string(file.name.split(".")[-1])
-                        log_message(LoggingScope.TASK_OPS_DETAILS, "INFO", "state: '%s'", state)
-                        states.append(state)
-        if len(states) == 0:
-            # did not find any file with metadata_file_state_path_prefix as prefix
-            log_message(LoggingScope.TASK_OPS, "INFO", "did not find any file with prefix '%s'",
-                        metadata_file_state_path_prefix)
-            return EESSITaskState.NEW_TASK
-        # sort the states and return the last one
-        states.sort()
-        state = states[-1]
-        log_message(LoggingScope.TASK_OPS_DETAILS, "INFO", "state: '%s'", state)
-        return state
-
     @log_function_entry_exit()
     def _list_directory_contents(self, directory_path: str, branch_name: str = None) -> List[Any]:
         """
@@ -920,39 +748,6 @@ def _determine_feature_branch_name(self) -> str:
         org, repo, pr, seq, _ = pull_request_dir.split("/")
         return f"{org}-{repo}-PR-{pr}-SEQ-{seq}"
 
-    @log_function_entry_exit()
-    def _sync_task_state_file(self, source_branch: str, target_branch: str):
-        """Update task state file from source to target branch"""
-        task_pointer_file = self.description.task_object.remote_file_path
-        pull_request_dir = self._read_pull_request_dir_from_file(task_pointer_file, self.git_repo.default_branch)
-        task_state_file_path = f"{pull_request_dir}/TaskState"
-
-        try:
-            # get content from source branch
-            source_content = self.git_repo.get_contents(task_state_file_path, ref=source_branch)
-
-            # get current file in target branch
-            target_file = self.git_repo.get_contents(task_state_file_path, ref=target_branch)
-
-            # update if content is different
-            if source_content.sha != target_file.sha:
-                result = self.git_repo.update_file(
-                    path=task_state_file_path,
-                    message=f"Sync {task_state_file_path} from {source_branch} to {target_branch}",
-                    content=source_content.decoded_content,
-                    sha=target_file.sha,
-                    branch=target_branch
-                )
-                log_message(LoggingScope.TASK_OPS, "INFO", "Updated '%s'", task_state_file_path)
-                return result
-            else:
-                log_message(LoggingScope.TASK_OPS, "INFO", "No changes needed for '%s'", task_state_file_path)
-                return None
-
-        except Exception as err:
-            log_message(LoggingScope.TASK_OPS, "ERROR", "Error syncing task state file: '%s'", err)
-            return None
-
     @log_function_entry_exit()
     def _update_task_states(self, next_state: EESSITaskState, default_branch_name: str,
                             approved_state: EESSITaskState, feature_branch_name: str):