RepoSetup.py

#!/usr/bin/env python3
 
# SPDX-FileCopyrightText: © 2023,2025 Siemens AG
# SPDX-FileContributor: Gaurav Mishra <mishra.gaurav@siemens.com>
 
# SPDX-License-Identifier: GPL-2.0-only
 
import fnmatch
import json
import os
import ssl
import urllib.request
from tempfile import TemporaryDirectory
 
from .ApiConfig import ApiConfig, Runner
from .CliOptions import CliOptions
 
 
class RepoSetup:
  """
  Setup temp_dir using the diff or current MR
 
  :ivar temp_dir: Temporary directory location for storing MR changes.
  :ivar allowlist: Allow list from JSON
  :ivar api_config: ApiConfig
  :ivar cli_options: CliOptions object
  """
 
  def __init__(self, cli_options: CliOptions, api_config: ApiConfig):
    """
    Create a temp dir
 
    :param cli_options: CliOptions object to get allow list from
    :param api_config: API configuration for the CI
    """
    self.temp_dir: TemporaryDirectory[str] | TemporaryDirectory[bytes] \
      = TemporaryDirectory()
    self.allowlist: dict[str, list[str]] = cli_options.allowlist
    self.api_config: ApiConfig = api_config
    self.cli_options: CliOptions = cli_options
 
  def __del__(self):
    """
    Clean the created temp dir
    """
    self.temp_dir.cleanup()
 
  def __is_excluded_path(self, path: str) -> bool:
    """
    Check if the path is allow listed
 
    The function used fnmatch to check if the path is in allow list or not.
 
    :param path: path to check
    :return: True if the path is in allow list, False otherwise
    """
    for pattern in self.allowlist.get('exclude', []):
      if fnmatch.fnmatchcase(path, pattern):
        return True
    return False
 
  def get_diff_dir(self) -> str:
    """
    Populate temp dir using the gitlab API `merge_requests`
 
    :return: temp dir path
    """
 
    context = ssl.create_default_context()
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE
 
    if self.api_config.running_on == Runner.GITLAB:
      headers = {'Private-Token': self.api_config.api_token}
      path_key = "new_path"
      change_key = "diff"
      api_req_url = (
        f"{self.api_config.api_url}/projects/"
        f"{self.api_config.project_id}/merge_requests/"
        f"{self.api_config.mr_iid}/changes"
      )
      if self.cli_options.differential and self.cli_options.tags != ('',''):
        tags = self.cli_options.tags
        api_req_url = (
          f"{self.api_config.api_url}/projects/"
          f"{self.api_config.project_id}/repository/compare?"
          f"from={tags[0]}&to={tags[1]}"
        )
 
    elif self.api_config.running_on == Runner.GITHUB:
      headers = {
        "Authorization": f"Bearer {self.api_config.api_token}",
        "X-GitHub-Api-Version": "2022-11-28",
        "Accept": "application/vnd.github+json"
      }
      path_key = "filename"
      change_key = "patch"
      api_req_url = (
        f"{self.api_config.api_url}/repos/"
        f"{self.api_config.github_repo_slug}/pulls/"
        f"{self.api_config.github_pull_request}/files"
      )
      if self.cli_options.differential and self.cli_options.tags != ('',''):
        tags = self.cli_options.tags
        api_req_url = (
          f"{self.api_config.api_url}/repos/"
          f"{self.api_config.github_repo_slug}/compare/"
          f"{tags[0]}...{tags[1]}"
        )
 
    else: # Default to Travis/GitHub behavior if runner is not explicitly GitLab or GitHub
      api_req_url = (
        "https://api.github.com/repos/"
        f"{self.api_config.travis_repo_slug}/pulls/"
        f"{self.api_config.travis_pull_request}/files"
      )
      headers = {}
      path_key = "filename"
      change_key = "patch"
 
    if not api_req_url:
      raise ValueError("API request URL could not be constructed. Check ApiConfig and runner type.")
 
    req = urllib.request.Request(api_req_url, headers=headers)
    try:
      with urllib.request.urlopen(req, context=context) as response:
        change_response = response.read()
    except Exception as e:
      print(f"Unable to get URL {api_req_url}")
      raise e
 
    change_response = json.loads(change_response)
    if self.api_config.running_on == Runner.GITLAB:
      if self.cli_options.differential and self.cli_options.tags != ('',''):
        changes = change_response.get('diffs', [])
      else:
        changes = change_response.get('changes', [])
    elif self.api_config.running_on == Runner.GITHUB:
      if self.cli_options.differential and self.cli_options.tags != ('',''):
        changes = change_response.get('files', [])
      else:
        changes = change_response
    else: # TRAVIS (or other unspecified runners)
      changes = change_response
 
    for change in changes:
      if path_key in change and change_key in change:
        file_path_in_repo = change[path_key]
        if not self.__is_excluded_path__is_excluded_path(file_path_in_repo):
          full_temp_file_path = os.path.join(self.temp_dir.name, file_path_in_repo)
          target_dir = os.path.dirname(full_temp_file_path)
          if target_dir and not os.path.exists(target_dir):
            os.makedirs(name=target_dir, exist_ok=True)
 
          try:
            with open(file=full_temp_file_path, mode='w+', encoding='UTF-8') as f:
              f.write(change[change_key])
          except IOError as e:
            print(f"Error writing file {full_temp_file_path}: {e}")
            raise e
 
    return self.temp_dir.name