REST api and SUBPROC

added more logging and move more functions to REST api and SUBPROC

requirements.txt

@@ -23,3 +23,7 @@ smmap==5.0.0
 typing_extensions==4.5.0
 urllib3==1.26.15
 python-dotenv~=1.0.0
+
+setuptools~=67.6.1
+bs4~=0.0.1
+beautifulsoup4~=4.12.2

src/blob_constants.py

@@ -174,7 +174,7 @@
     "Monitoring-refine",
     "Blue-green-deploy",
     "GitOps-adoption",
-    "CI/CD-pipeline-opt",
+    "CICD-pipeline-opt",
     "Deployment-refactor",
     "Performance-refactor",
     "Error-handling-ref",

src/helpers/git_functions.py

@@ -1,5 +1,6 @@
 import subprocess
 from typing import Dict
+from bs4 import BeautifulSoup
 
 import requests
 
@@ -30,14 +31,17 @@ def get_github_headers(token: str) -> Dict:
     }
 
 
-def get_repo_data(url: str, token: str, data_key: str) -> str:
+def get_default_branch(url: str, token: str) -> str:
     headers = get_github_headers(token)
     response = requests.get(url, headers=headers)
 
     # Check if the response was successful
     if response.status_code == 200:
         handle_success(f"Repo {url} info was retrieved successfully")
-        repo_info = response.json()
-        return repo_info[data_key]
+        # Parse the HTML content
+        soup = BeautifulSoup(response.content, 'html.parser')
+        if not soup.find('span', {'class': 'css-truncate-target'}).text:
+            handle_error("not able to determine repo default branch - html parser could not find it")
+        return soup.find('span', {'class': 'css-truncate-target'}).text
     else:
-        handle_error(f"Repo {url} info was not retrieved successfully: HTTP {response.status_code}")
+        handle_error(f"default branch for {url} was not retrieved successfully: HTTP {response.status_code}")

src/helpers/logging_functions.py

@@ -1,5 +1,7 @@
 import logging
 # Set up logging
+from subprocess import CompletedProcess
+
 logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
 
 
@@ -12,3 +14,15 @@ def handle_error(error_message: str) -> None:
 def handle_success(message: str) -> None:
     print(message)
     logging.info(message)
+
+
+def handle_subprocess_output(output: CompletedProcess, task: str) -> None:
+    if output.returncode != 0:
+        if output.stderr:
+            handle_error(f'{task}: {output.stderr.decode("utf-8")}')
+        elif output.stdout:
+            handle_error(f'{task}: {output.stdout.decode("utf-8")}')
+        else:
+            handle_error(f'{task} failed: return code = {output.returncode}')
+    else:
+        handle_success(f"Successfully executed: {task}")

src/services/git_svc.py

@@ -1,4 +1,3 @@
-import json
 import os
 import random
 import requests
@@ -17,8 +16,8 @@
     PULL_REQUESTS_TITLES,
     SERVICE_COMMIT_MSGS,
 )
-from helpers.git_functions import validate_git_config_is_set, get_github_headers, get_repo_data
-from helpers.logging_functions import handle_error, handle_success
+from helpers.git_functions import validate_git_config_is_set, get_github_headers, get_default_branch
+from helpers.logging_functions import handle_error, handle_success, handle_subprocess_output
 from model.git_data import GitData
 from services.data_var_svc import get_random_app_service_name_and_id
 
@@ -41,6 +40,8 @@ def _set_gituser_cred(self) -> None:
         main_git_user_id = str(random.randint(1, self._git_data.git_users_count))
         self._username = os.environ[f"GIT_USER{main_git_user_id}"]
         self._password = os.environ[f"GIT_PAT{main_git_user_id}"]
+        subprocess.run(['git', 'config', '--global', 'user.name', self._username])
+        subprocess.run(['git', 'config', '--global', 'user.password', self._password])
 
     def produce_pull_request(self, jira_tickets: List[str]) -> None:
         number_of_commits = random.randint(int(os.environ['MIN_COMMITS']), int(os.environ['MAX_COMMITS']))
@@ -55,45 +56,41 @@ def produce_pull_request(self, jira_tickets: List[str]) -> None:
         self._create_pull_request(
             f"{random.choice(PULL_REQUESTS_TITLES)}",  # title
             f"General fixes/changes + taking care of the following tickets: {' '.join(jira_tickets)}",
-            get_repo_data(url=self._git_data.repo_url_short, token=self._password, data_key="default_branch"),  # head
+            get_default_branch(url=f"https://{self._git_data.repo_url_short}", token=self._password),
             new_branch_name  # base
         )
 
     def _clone_repo(self) -> None:
-        os.chdir(self._original_dir)
-        os.mkdir("repo")
-        os.chdir(f"{self._original_dir}{os.sep}repo")
+        self._cd_to_repo_dir()
 
-        extra_headers = f"Authorization: Bearer {self._password}"
-        clone_command = f'git clone {self._git_data.repo_url_short}.git . -c http.extraHeader={extra_headers}'
+        clone_command = f'git clone https://{self._username}:{self._password}@{self._git_data.repo_url_short}.git .'
         result = subprocess.run(clone_command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         if result.returncode == 0:
             handle_success(f'Repository {self._git_data.repo_url_short} cloned successfully.')
         else:
             handle_error(f'Failed to clone repository. Error: {result.stderr.decode("utf-8")}')
 
-    def _create_branch(self):
+    def _cd_to_repo_dir(self) -> None:
+        os.chdir(self._original_dir)
+        os.mkdir("repo")
+        os.chdir(f"{self._original_dir}{os.sep}repo")
+
+    def _create_branch(self) -> str:
         new_branch_name = f"{random.choice(BRANCHES_NAMES_PREFIXES)}{random.choice(BRANCHES_NAMES)}"
 
         GitService._create_local_branch(new_branch_name)
-        self._git_push()
+        self._git_push(new_branch_name)
         return new_branch_name
 
-    def _git_push(self) -> None:
-        push_result = subprocess.run(['git', 'push', '-u',
-                                      f'https://{self._username}:{self._password}@{self._git_data.repo_url_short}git'])
-        if push_result.returncode != 0:
-            handle_error(f'Error pushing changes: {push_result.stderr.decode("utf-8")}')
-        else:
-            handle_success("Successfully pushed changes")
+    def _git_push(self, branch: str) -> None:
+        command = f"git push https://{self._username}:{self._password}@{self._git_data.repo_url_short}.git {branch}"
+        push_result = subprocess.run(command)
+        handle_subprocess_output(push_result, "<Pushing Git Changes>")
 
     @staticmethod
     def _create_local_branch(new_branch_name: str) -> None:
         create_branch_result = subprocess.run(['git', 'checkout', '-b', new_branch_name])
-        if create_branch_result.returncode != 0:
-            handle_error(f'Error creating branch: {create_branch_result.stderr.decode("utf-8")}')
-        else:
-            handle_success(f"Successfully created branch locally :{new_branch_name}")
+        handle_subprocess_output(create_branch_result, "<Creating local Git Branch>")
 
     def _create_commit(self, app_service: str, app_service_id: int, jira_ticket_ref: str = "", ) -> None:
         self._set_gituser_cred()
@@ -102,8 +99,10 @@ def _create_commit(self, app_service: str, app_service_id: int, jira_ticket_ref:
             self._create_code_change(app_service)
             commit_msg = GitService.create_commit_msg(jira_ticket_ref, app_service_id)
             subprocess.run(['git', 'add', '.'])
-            subprocess.run(['git', 'commit', '-m', commit_msg, f'--author={self._username} <{self._password}>'])
-            self._git_push()
+            subprocess.run(['git', 'commit', '-m', commit_msg])
+            result = subprocess.run(['git', 'rev-parse', '--abbrev-ref', 'HEAD'], stdout=subprocess.PIPE, text=True)
+            current_branch = result.stdout.strip()
+            self._git_push(current_branch)
 
         except Exception as e:
             print(e)
@@ -135,8 +134,13 @@ def create_commit_msg(jira_ticket_ref, app_service_id) -> str:
     def _create_pull_request(self, title, description, source_branch, target_branch):
         """Creates the pull request for the head_branch against the base_branch"""
         headers = {"Accept": "application/vnd.github.v3+json", "Authorization": f"Bearer {self._password}"}
-        payload = self._create_pr_payload(description, source_branch, target_branch, title)
-        r = requests.post(self._git_data.git_pulls_api, headers=headers, data=json.dumps(payload))
+        payload = {
+            "title": title,
+            "body": description,
+            "head": f"{self._git_data.repo_org}:{target_branch}",
+            "base": f"{source_branch}",
+        }
+        r = requests.post(self._git_data.git_pulls_api, headers=headers, json=payload)
 
         if not r.ok:
             print("Request Failed: {0}".format(r.text))