diff --git a/README.md b/README.md index c8fef2e..685018f 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,32 @@ Run `source ~/.zshrc` or restart terminal. ## Usage āš” +## Project structure + +The CLI entry point remains `gitHappens.py` so existing aliases keep working, but +the implementation is split into focused modules: + +``` +gitHappens.py # Compatibility wrapper +githappens/ + main.py # Argument parsing and command routing + config.py # config.ini/templates.json loading + gitlab_api.py # GitLab and glab API interactions + git_utils.py # Local git helpers + interactive.py # Inquirer prompts and selections + templates.py # Template lookup rules + commands/ + create_issue.py # Issue, branch, and MR creation workflow + open_mr.py # Open active MR in the browser + review.py # Reviewers, time tracking, and AI review + summary.py # Git summary and AI summary commands + report.py # Incident report workflow + deploy.py # Last production deployment lookup +``` + +This split keeps configuration, API calls, git shelling, prompts, and command +logic separate so each part can be tested and extended independently. + ### Project selection - Project selection is made automatically if you run script in same path as your project is located. @@ -226,3 +252,11 @@ I suggest checking Gitlab's official API documentation: https://docs.gitlab.com/ Make sure to check this project on [OpenPledge](https://app.openpledge.io/repositories/zigcBenx/gitHappens). +## Development + +Run the unit tests with: + +```bash +python -m unittest discover +``` + diff --git a/gitHappens.py b/gitHappens.py index 27d47f3..b77e8da 100755 --- a/gitHappens.py +++ b/gitHappens.py @@ -1,845 +1,8 @@ #!/usr/bin/env python3 -import subprocess -import json -import argparse -import configparser -import inquirer -import datetime -import re -import os -import requests -import sys -import webbrowser +"""Backward-compatible entry point for the GitHappens CLI.""" -# Setup config parser and read settings -config = configparser.ConfigParser() -absolute_config_path = os.path.dirname(os.path.abspath(__file__)) -config_path = os.path.join(absolute_config_path, 'configs/config.ini') -config.read(config_path) +from githappens.main import main -BASE_URL = config.get('DEFAULT', 'base_url') -API_URL = BASE_URL + '/api/v4' -GROUP_ID = config.get('DEFAULT', 'group_id') -CUSTOM_TEMPLATE = config.get('DEFAULT', 'custom_template') -GITLAB_TOKEN = config.get('DEFAULT', 'GITLAB_TOKEN').strip('\"\'') -DELETE_BRANCH = config.get('DEFAULT', 'delete_branch_after_merge').lower() == 'true' -DEVELOPER_EMAIL = config.get('DEFAULT', 'developer_email', fallback=None) -SQUASH_COMMITS = config.get('DEFAULT', 'squash_commits').lower() == 'true' -PRODUCTION_PIPELINE_NAME = config.get('DEFAULT', 'production_pipeline_name', fallback='deploy') -PRODUCTION_JOB_NAME = config.get('DEFAULT', 'production_job_name', fallback=None) -PRODUCTION_REF = config.get('DEFAULT', 'production_ref', fallback=None) -MAIN_BRANCH = 'master' -# Read templates from json config -with open(os.path.join(absolute_config_path,'configs/templates.json'), 'r') as f: - jsonConfig = json.load(f) -TEMPLATES = jsonConfig['templates'] -REVIEWERS = jsonConfig['reviewers'] -PRODUCTION_MAPPINGS = jsonConfig.get('productionMappings', {}) - -def get_project_id(): - project_link = getProjectLinkFromCurrentDir() - if (project_link == -1): - return enterProjectId() - - allProjects = get_all_projects(project_link) - # Find projects id by project ssh link gathered from repo - matching_id = None - for project in allProjects: - if project.get("ssh_url_to_repo") == project_link: - matching_id = project.get("id") - break - return matching_id - -def get_all_projects(project_link): - url = API_URL + "/projects?membership=true&search=" + project_link.split('/')[-1].split('.')[0] - - headers = { - "PRIVATE-TOKEN": GITLAB_TOKEN - } - - response = requests.get(url, headers=headers) - - if response.status_code == 200: - return response.json() - elif response.status_code == 401: - print("Error: Unauthorized (401). Your GitLab token is probably expired, invalid, or missing required permissions.") - print("Please generate a new token and update your configs/config.ini.") - exit(1) - else: - print(f"Request failed with status code {response.status_code}") - return None - -def getProjectLinkFromCurrentDir(): - try: - cmd = 'git remote get-url origin' - result = subprocess.run(cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE) - if result.returncode == 0: - output = result.stdout.decode('utf-8').strip() - return output - else: - return -1 - except FileNotFoundError: - return -1 - -def enterProjectId(): - while True: - project_id = input('Please enter the ID of your GitLab project: ') - if project_id: - return project_id - exit('Invalid project ID.') - -def list_milestones(current=False): - cmd = f'glab api /groups/{GROUP_ID}/milestones?state=active' - result = subprocess.run(cmd.split(), stdout=subprocess.PIPE) - milestones = json.loads(result.stdout) - if current: - today = datetime.date.today().strftime('%Y-%m-%d') - active_milestones = [] - for milestone in milestones: - start_date = milestone['start_date'] - due_date = milestone['due_date'] - if start_date and due_date and start_date <= today and due_date >= today: - active_milestones.append(milestone) - active_milestones.sort(key=lambda x: x['due_date']) - return active_milestones[0] - return milestones - -def select_template(): - template_names = [t['name'] for t in TEMPLATES] - template_names.append(CUSTOM_TEMPLATE) - questions = [ - inquirer.List('template', - message="Select template:", - choices=template_names, - ), - ] - answer = inquirer.prompt(questions) - return answer['template'] - -def getIssueSettings(template_name): - if template_name == CUSTOM_TEMPLATE: - return {} - return next((t for t in TEMPLATES if t['name'] == template_name), None) - -def createIssue(title, project_id, milestoneId, epic, iteration, settings): - if settings: - issueType = settings.get('type') or 'issue' - return executeIssueCreate(project_id, title, settings.get('labels'), milestoneId, epic, iteration, settings.get('weight'), settings.get('estimated_time'), issueType) - print("No settings in template") - exit(2) - pass - -def executeIssueCreate(project_id, title, labels, milestoneId, epic, iteration, weight, estimated_time, issue_type='issue'): - labels = ",".join(labels) if type(labels) == list else labels - assignee_id = getAuthorizedUser()['id'] - issue_command = [ - "glab", "api", - f"/projects/{str(project_id)}/issues", - "-f", f'title={title}', - "-f", f'assignee_ids={assignee_id}', - "-f", f'issue_type={issue_type}' - ] - if labels: - issue_command.append("-f") - issue_command.append(f'labels={labels}') - - if weight: - issue_command.append("-f") - issue_command.append(f'weight={str(weight)}') - - if milestoneId: - issue_command.append("-f") - issue_command.append(f'milestone_id={str(milestoneId)}') - - if epic: - epicId = epic['id'] - issue_command.append("-f") - issue_command.append(f'epic_id={str(epicId)}') - - # Set the description, including iteration, estimated time, and other info - description = "" - if iteration: - iterationId = iteration['id'] - description += f"/iteration *iteration:{str(iterationId)} " - - if estimated_time: - description += f"\n/estimate {estimated_time}m " - - issue_command.extend(["-f", f'description={description}']) - - issue_output = subprocess.check_output(issue_command) - return json.loads(issue_output.decode()) - -def select_milestone(milestones): - milestones = [t['title'] for t in milestones] - questions = [ - inquirer.List('milestones', - message="Select milestone:", - choices=milestones, - ), - ] - answer = inquirer.prompt(questions) - return answer['milestones'] - -def getSelectedMilestone(milestone, milestones): - return next((t for t in milestones if t['title'] == milestone), None) - -def get_milestone(manual): - if manual: - milestones = list_milestones() - return getSelectedMilestone(select_milestone(milestones), milestones) - milestone = list_milestones(True) # select active for today - return milestone - -def get_iteration(manual): - if manual: - iterations = list_iterations() - return getSelectedIteration(select_iteration(iterations), iterations) - return getActiveIteration() - -def getSelectedIteration(iteration, iterations): - return next((t for t in iterations if t['start_date'] + ' - ' + t['due_date'] == iteration), None) - -def select_iteration(iterations): - iterations = [t['start_date'] + ' - ' + t['due_date'] for t in iterations] - questions = [ - inquirer.List('iterations', - message="Select iteration:", - choices=iterations, - ), - ] - answer = inquirer.prompt(questions) - return answer['iterations'] - -def list_iterations(): - cmd = f'glab api /groups/{GROUP_ID}/iterations?state=opened' - result = subprocess.run(cmd.split(), stdout=subprocess.PIPE) - iterations = json.loads(result.stdout) - return iterations - -def getActiveIteration(): - iterations = list_iterations() - today = datetime.date.today().strftime('%Y-%m-%d') - active_iterations = [] - for iteration in iterations: - start_date = iteration['start_date'] - due_date = iteration['due_date'] - if start_date and due_date and start_date <= today and due_date >= today: - active_iterations.append(iteration) - active_iterations.sort(key=lambda x: x['due_date']) - return active_iterations[0] - -def getAuthorizedUser(): - output = subprocess.check_output(["glab", "api", "/user"]) - return json.loads(output) - -def list_epics(): - cmd = f'glab api /groups/{GROUP_ID}/epics?per_page=1000&state=opened' - result = subprocess.run(cmd.split(), stdout=subprocess.PIPE) - return json.loads(result.stdout) - -def select_epic(epics): - epics = [t['title'] for t in epics] - search_query = inquirer.prompt([ - inquirer.Text('search_query', message='Search epic:'), - ])['search_query'] - - # Filter choices based on search query - filtered_epics = [c for c in epics if search_query.lower() in c.lower()] - questions = [ - inquirer.List('epics', - message="Select epic:", - choices=filtered_epics, - ), - ] - answer = inquirer.prompt(questions) - return answer['epics'] - -def getSelectedEpic(epic, epics): - return next((t for t in epics if t['title'] == epic), None) - -def get_epic(): - epics = list_epics() - return getSelectedEpic(select_epic(epics), epics) - -def create_branch(project_id, issue): - issueId = str(issue['iid']) - title = re.sub('\\s+', '-', issue['title']).lower() - title = issueId + '-' + title.replace(':','').replace('(',' ').replace(')', '').replace(' ','-') - branch_output = subprocess.check_output(["glab", "api", f"/projects/{str(project_id)}/repository/branches", "-f", f'branch={title}', "-f", f'ref={MAIN_BRANCH}', "-f", f'issue_iid={issueId}']) - return json.loads(branch_output.decode()) - -def create_merge_request(project_id, branch, issue, labels, milestoneId): - issueId = str(issue['iid']) - branch = branch['name'] - title = issue['title'] - assignee_id = getAuthorizedUser()['id'] - labels = ",".join(labels) if type(labels) == list else labels - merge_request_command = [ - "glab", "api", - f"/projects/{str(project_id)}/merge_requests", - "-f", f'title={title}', - "-f", f'description="Closes #{issueId}"', - "-f", f'source_branch={branch}', - "-f", f'target_branch={MAIN_BRANCH}', - "-f", f'issue_iid={issueId}', - "-f", f'assignee_ids={assignee_id}' - ] - - if SQUASH_COMMITS: - merge_request_command.append("-f") - merge_request_command.append("squash=true") - - if DELETE_BRANCH: - merge_request_command.append("-f") - merge_request_command.append("remove_source_branch=true") - - if labels: - merge_request_command.append("-f") - merge_request_command.append(f'labels={labels}') - - if milestoneId: - merge_request_command.append("-f") - merge_request_command.append(f'milestone_id={str(milestoneId)}') - - mr_output = subprocess.check_output(merge_request_command) - return json.loads(mr_output.decode()) - -def startIssueCreation(project_id, title, milestone, epic, iteration, selectedSettings, onlyIssue): - # Prompt for estimated time - estimated_time = inquirer.prompt([ - inquirer.Text('estimated_time', - message='Estimated time to complete this issue (in minutes, optional)', - validate=lambda _, x: x == '' or x.isdigit()) - ])['estimated_time'] - - # If multiple project IDs, split the estimated time - if isinstance(project_id, list): - estimated_time_per_project = int(estimated_time) / len(project_id) if estimated_time else None - else: - estimated_time_per_project = estimated_time - - # Modify settings to include estimated time - if estimated_time_per_project: - selectedSettings = selectedSettings.copy() if selectedSettings else {} - selectedSettings['estimated_time'] = int(estimated_time_per_project) - - createdIssue = createIssue(title, project_id, milestone, epic, iteration, selectedSettings) - print(f"Issue #{createdIssue['iid']}: {createdIssue['title']} created.") - - if onlyIssue: - return createdIssue - - createdBranch = create_branch(project_id, createdIssue) - - createdMergeRequest = create_merge_request(project_id, createdBranch, createdIssue, selectedSettings.get('labels'), milestone) - print(f"Merge request #{createdMergeRequest['iid']}: {createdMergeRequest['title']} created.") - - print("Run:") - print(" git fetch origin") - print(f" git checkout -b '{createdMergeRequest['source_branch']}' 'origin/{createdMergeRequest['source_branch']}'") - print("to switch to new branch.") - - return createdIssue - -def getCurrentBranch(): - return subprocess.check_output(['git', 'rev-parse', '--abbrev-ref', 'HEAD'], text=True).strip() - -def openMergeRequestInBrowser(): - try: - merge_request_id = getActiveMergeRequestId() - remote_url = subprocess.check_output(["git", "config", "--get", "remote.origin.url"], text=True).strip() - url = BASE_URL + '/' + remote_url.split(':')[1][:-4] - webbrowser.open(f"{url}/-/merge_requests/{merge_request_id}") - except subprocess.CalledProcessError: - return None - -def getActiveMergeRequestId(): - branch_to_find = getCurrentBranch() - return find_merge_request_id_by_branch(branch_to_find) - -def find_merge_request_id_by_branch(branch_name): - return getMergeRequestForBranch(branch_name)['iid'] - -def getMergeRequestForBranch(branchName): - project_id = get_project_id() - api_url = f"{API_URL}/projects/{project_id}/merge_requests" - headers = {"Private-Token": GITLAB_TOKEN} - - params = { - "source_branch": branchName, - } - - response = requests.get(api_url, headers=headers, params=params) - if response.status_code == 200: - merge_requests = response.json() - for mr in merge_requests: - if mr["source_branch"] == branchName: - return mr - else: - print(f"Failed to fetch Merge Requests: {response.status_code} - {response.text}") - return None - -def chooseReviewersManually(): - """Prompt the user to select reviewers manually from the available list, showing names.""" - # Fetch user details for each reviewer ID - reviewer_choices = [] - for reviewer_id in REVIEWERS: - api_url = f"{API_URL}/users/{reviewer_id}" - headers = {"Private-Token": GITLAB_TOKEN} - try: - response = requests.get(api_url, headers=headers) - if response.status_code == 200: - user = response.json() - display_name = f"{user.get('name')} ({user.get('username')})" - reviewer_choices.append((display_name, reviewer_id)) - else: - reviewer_choices.append((str(reviewer_id), reviewer_id)) - except Exception: - reviewer_choices.append((str(reviewer_id), reviewer_id)) - - questions = [ - inquirer.Checkbox( - "selected_reviewers", - message="Select reviewers", - choices=[(name, str(rid)) for name, rid in reviewer_choices], - ) - ] - answers = inquirer.prompt(questions) - if answers and "selected_reviewers" in answers: - return [int(r) for r in answers["selected_reviewers"]] - else: - return [] - -def addReviewersToMergeRequest(reviewers=None): - project_id = get_project_id() - mr_id = getActiveMergeRequestId() - api_url = f"{API_URL}/projects/{project_id}/merge_requests/{mr_id}" - headers = {"Private-Token": GITLAB_TOKEN} - - data = { - "reviewer_ids": reviewers if reviewers is not None else REVIEWERS - } - - requests.put(api_url, headers=headers, json=data) - -def setMergeRequestToAutoMerge(): - project_id = get_project_id() - mr_id = getActiveMergeRequestId() - api_url = f"{API_URL}/projects/{project_id}/merge_requests/{mr_id}/merge" - headers = {"Private-Token": GITLAB_TOKEN} - - data = { - "id": project_id, - "merge_request_iid": mr_id, - "should_remove_source_branch": True, - "merge_when_pipeline_succeeds": True, - "auto_merge_strategy": "merge_when_pipeline_succeeds", - } - - requests.put(api_url, headers=headers, json=data) - -def getMainBranch(): - command = "git symbolic-ref refs/remotes/origin/HEAD | sed 's@^refs/remotes/origin/@@'" - output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, universal_newlines=True) - return output.strip() - - -def get_two_weeks_commits(return_output=False): - two_weeks_ago = (datetime.datetime.now() - datetime.timedelta(weeks=2)).strftime('%Y-%m-%d') - - cmd = f'git log --since={two_weeks_ago} --format="%ad - %ae - %s" --date=short | grep -v "Merge branch"' - if (DEVELOPER_EMAIL): - cmd = f'{cmd} | grep {DEVELOPER_EMAIL}' - try: - output = subprocess.check_output(cmd, shell=True, text=True, stderr=subprocess.DEVNULL, universal_newlines=True).strip() - if output: - if return_output: - return output - print(output) - else: - print("No commits found.") - return "" if return_output else None - except subprocess.CalledProcessError as e: - print(f"No commits were found or an error occurred. (exit status {e.returncode})") - return "" if return_output else None - except FileNotFoundError: - print("Git is not installed or not found in PATH.") - return "" if return_output else None - -def generate_smart_summary(): - commits = get_two_weeks_commits(return_output=True) - if not commits: - return - - # Check if OpenAI API key is set - openai_api_key = config.get('DEFAULT', 'OPENAI_API_KEY', fallback=None) - if not openai_api_key: - print("OpenAI API key not set. Skipping AI summary generation.") - return - - # Dynamically import openai only if API key is present - try: - import openai - except ImportError: - print("OpenAI package not installed. Please install it using: pip install openai") - return - - openai.api_key = openai_api_key - - try: - response = openai.chat.completions.create( - model="gpt-3.5-turbo", - messages=[ - {"role": "system", "content": "You are a helpful assistant that summarizes git commits. Provide a concise, well-organized summary of the main changes and themes."}, - {"role": "user", "content": f"Please summarize these git commits in a clear, bulleted format:\n\n{commits}"} - ] - ) - - print("\nšŸ“‹ AI-Generated Summary of Recent Changes:\n") - print(response.choices[0].message.content) - except Exception as e: - print(f"Error generating AI summary: {e}") - -def process_report(text, minutes): - # Get the incident project ID from config - try: - incident_project_id = config.get('DEFAULT', 'incident_project_id') - except (configparser.NoOptionError, configparser.NoSectionError): - print("Error: incident_project_id not found in config.ini") - print("Please add your incident project ID to configs/config.ini under [DEFAULT] section:") - print("incident_project_id = your_project_id_here") - return - - issue_title = f"Incident Report: {text}" - - selected_label = selectLabels('Department') - - incident_settings = { - 'labels': ['incident', 'report'], - 'onlyIssue': True, - 'type': 'incident' - } - - if selected_label: - incident_settings['labels'].append(selected_label) - - try: - # Create the incident issue - iteration = getActiveIteration() - created_issue = createIssue(issue_title, incident_project_id, False, False, iteration, incident_settings) - issue_iid = created_issue['iid'] - - closeOpenedIssue(issue_iid, incident_project_id) - print(f"Incident issue #{issue_iid} created successfully.") - print(f"Title: {issue_title}") - - # Add time tracking to the issue - time_tracking_command = [ - "glab", "api", - f"/projects/{incident_project_id}/issues/{issue_iid}/add_spent_time", - "-f", f"duration={minutes}m" - ] - - try: - subprocess.run(time_tracking_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - print(f"Added {minutes} minutes to issue time tracking.") - except subprocess.CalledProcessError as e: - print(f"Error adding time tracking: {str(e)}") - - except Exception as e: - print(f"Error creating incident issue: {str(e)}") - -def closeOpenedIssue(issue_iid, project_id): - issue_command = [ - "glab", "api", - f"/projects/{project_id}/issues/{issue_iid}", - '-X', 'PUT', - '-f', 'state_event=close' - ] - try: - subprocess.run(issue_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - except subprocess.CalledProcessError as e: - print(f"Error closing issue: {str(e)}") - -def selectLabels(search, multiple = False): - labels = getLabelsOfGroup(search) - labels = sorted([t['name'] for t in labels]) - - question_type = inquirer.Checkbox if multiple else inquirer.List - questions = [ - question_type( - 'labels', - message="Select one or more department labels:", - choices=labels, - ), - ] - answer = inquirer.prompt(questions) - return answer['labels'] - -def getLabelsOfGroup(search=''): - cmd = f'glab api /groups/{GROUP_ID}/labels?search={search}' - try: - result = subprocess.run(cmd.split(), stdout=subprocess.PIPE, check=True) - return json.loads(result.stdout) - except subprocess.CalledProcessError as e: - print(f"Error getting labels: {str(e)}") - return [] - -def getCurrentIssueId(): - mr = getMergeRequestForBranch(getCurrentBranch()) - return mr['description'].replace('"','').replace('#','').split()[1] - -def track_issue_time(): - # Get the current merge request - try: - project_id = get_project_id() - issue_id = getCurrentIssueId() - except Exception as e: - print(f"Error getting issue details: {str(e)}") - return - - # Prompt for actual time spent - spent_time = inquirer.prompt([ - inquirer.Text('spent_time', - message='How many minutes did you actually spend on this issue?', - validate=lambda _, x: x.isdigit()) - ])['spent_time'] - - # Add spent time to the issue description - time_tracking_command = [ - "glab", "api", - f"/projects/{project_id}/issues/{issue_id}/notes", - "-f", f"body=/spend {spent_time}m" - ] - - try: - subprocess.run(time_tracking_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True) - print(f"Added {spent_time} minutes to issue {issue_id} time tracking.") - except subprocess.CalledProcessError as e: - print(f"Error adding time tracking: {str(e)}") - except Exception as e: - print(f"Error tracking issue time: {str(e)}") - -def get_last_production_deploy(): - try: - project_id = get_project_id() - api_url = f"{API_URL}/projects/{project_id}/pipelines" - headers = {"Private-Token": GITLAB_TOKEN} - - # Set up parameters for the pipeline search - params = { - "per_page": 50, - "order_by": "updated_at", - "sort": "desc" - } - - # Add ref filter if specified in config - if MAIN_BRANCH: - params["ref"] = MAIN_BRANCH - else: - # Use main branch if no specific ref is configured - try: - main_branch = getMainBranch() - params["ref"] = main_branch - except: - # Fallback to common main branch names - params["ref"] = "main" - - response = requests.get(api_url, headers=headers, params=params) - - if response.status_code != 200: - print(f"Failed to fetch pipelines: {response.status_code} - {response.text}") - return - - pipelines = response.json() - production_pipeline = None - - # Look for production pipeline by name pattern - for pipeline in pipelines: - # Get pipeline details to check jobs - pipeline_detail_url = f"{API_URL}/projects/{project_id}/pipelines/{pipeline['id']}/jobs" - detail_response = requests.get(pipeline_detail_url, headers=headers) - - if detail_response.status_code == 200: - jobs = detail_response.json() - - # Check if this pipeline contains production deployment - for job in jobs: - job_name = job.get('name', '') - stage = job.get('stage', '') - job_status = job.get('status', '').lower() - - # Only consider successful jobs - if job_status != 'success': - continue - - # Check project-specific mapping first - project_mapping = PRODUCTION_MAPPINGS.get(str(project_id)) - if project_mapping: - expected_stage = project_mapping.get('stage', '').lower() - expected_job = project_mapping.get('job', '').lower() - - if (stage.lower() == expected_stage or - (expected_job and job_name.lower() == expected_job)): - production_pipeline = { - 'pipeline': pipeline, - 'production_job': job - } - break - else: - print('Didn\'t find deployment pipeline') - - if production_pipeline: - break - - if not production_pipeline: - print(f"No production deployment found matching pattern") - return - - # Display the results - pipeline = production_pipeline['pipeline'] - job = production_pipeline['production_job'] - - print(f"šŸš€ Last Production Deployment:") - print(f" Pipeline: #{pipeline['id']} - {pipeline['status']}") - print(f" Job: {job['name']} ({job['status']})") - print(f" Branch/Tag: {pipeline['ref']}") - print(f" Started: {job.get('started_at', 'N/A')}") - print(f" Finished: {job.get('finished_at', 'N/A')}") - print(f" Duration: {job.get('duration', 'N/A')} seconds" if job.get('duration') else " Duration: N/A") - print(f" Commit: {pipeline['sha'][:8]}") - print(f" URL: {pipeline['web_url']}") - - # Show time since deployment - if job.get('finished_at'): - try: - finished_time = datetime.datetime.fromisoformat(job['finished_at'].replace('Z', '+00:00')) - time_diff = datetime.datetime.now(datetime.timezone.utc) - finished_time - - if time_diff.days > 0: - print(f" ā° {time_diff.days} days ago") - elif time_diff.seconds > 3600: - hours = time_diff.seconds // 3600 - print(f" ā° {hours} hours ago") - else: - minutes = time_diff.seconds // 60 - print(f" ā° {minutes} minutes ago") - except: - pass - - except Exception as e: - print(f"Error fetching last production deploy: {str(e)}") - -def main(): - global MAIN_BRANCH - - parser = argparse.ArgumentParser("Argument description of Git happens") - parser.add_argument("title", nargs="+", help="Title of issue") - parser.add_argument(f"--project_id", type=str, help="Id or URL-encoded path of project") - parser.add_argument("-m", "--milestone", action='store_true', help="Add this flag, if you want to manually select milestone") - parser.add_argument("--no_epic", action="store_true", help="Add this flag if you don't want to pick epic") - parser.add_argument("--no_milestone", action="store_true", help="Add this flag if you don't want to pick milestone") - parser.add_argument("--no_iteration", action="store_true", help="Add this flag if you don't want to pick iteration") - parser.add_argument("--only_issue", action="store_true", help="Add this flag if you don't want to create merge request and branch alongside issue") - parser.add_argument("-am", "--auto_merge", action="store_true", help="Add this flag to review if you want to set merge request to auto merge when pipeline succeeds") - parser.add_argument("--select", action="store_true", help="Manually select reviewers for merge request (interactive)") - - # If no arguments passed, show help - if len(sys.argv) <= 1: - parser.print_help() - exit(1) - - args = parser.parse_args() - if args.title[0] == 'report': - parts = args.title - if len(parts) != 3: - print("Invalid report format. Use: gh report \"text\" minutes") - return - - text = parts[1] - try: - minutes = int(parts[2].strip()) - process_report(text, minutes) - except ValueError: - print("Invalid minutes. Please provide a valid number.") - return - - # So it takes all text until first known argument - title = " ".join(args.title) - - if title == 'open': - openMergeRequestInBrowser() - return - elif title == 'review': - track_issue_time() - reviewers = None - if getattr(args, "select", False): - reviewers = chooseReviewersManually() - addReviewersToMergeRequest(reviewers=reviewers) - - # Run AI code review and post to MR - try: - from ai_code_review import run_review_for_mr - project_id = get_project_id() - mr_id = getActiveMergeRequestId() - run_review_for_mr(project_id, mr_id, GITLAB_TOKEN, API_URL) - except Exception as e: - print(f"AI review skipped: {e}") - - if(args.auto_merge): - setMergeRequestToAutoMerge() - return - elif title == 'summary': - get_two_weeks_commits() - return - elif title == 'summaryAI': - generate_smart_summary() - return - elif title == 'last deploy': - get_last_production_deploy() - return - elif title == 'ai review': - from ai_code_review import run_review - run_review() - return - - # Get settings for issue from template - selectedSettings = getIssueSettings(select_template()) - - # If template is False, ask for each settings - if not len(selectedSettings): - print('Custom selection of issue settings is not supported yet') - pass - - if args.project_id and selectedSettings.get('projectIds'): - print('NOTE: Overwriting project id from argument...') - - project_id = selectedSettings.get('projectIds') or args.project_id or get_project_id() - - milestone = False - if not args.no_milestone: - milestone = get_milestone(args.milestone)['id'] - - iteration = False - if not args.no_iteration: - # manual pick iteration - iteration = get_iteration(True) - - epic = False - if not args.no_epic: - epic = get_epic() - - MAIN_BRANCH = getMainBranch() - - onlyIssue = selectedSettings.get('onlyIssue') or args.only_issue - - if type(project_id) == list: - for id in project_id: - startIssueCreation(id, title, milestone, epic, iteration, selectedSettings, onlyIssue) - else: - startIssueCreation(project_id, title, milestone, epic, iteration, selectedSettings, onlyIssue) - -if __name__ == '__main__': - main() \ No newline at end of file +if __name__ == "__main__": + main() diff --git a/githappens/__init__.py b/githappens/__init__.py new file mode 100644 index 0000000..2fde185 --- /dev/null +++ b/githappens/__init__.py @@ -0,0 +1,5 @@ +"""GitHappens CLI package.""" + +__all__ = ["__version__"] + +__version__ = "0.1.0" diff --git a/githappens/commands/__init__.py b/githappens/commands/__init__.py new file mode 100644 index 0000000..8117035 --- /dev/null +++ b/githappens/commands/__init__.py @@ -0,0 +1 @@ +"""Command handlers for the GitHappens CLI.""" diff --git a/githappens/commands/create_issue.py b/githappens/commands/create_issue.py new file mode 100644 index 0000000..8a0cc7c --- /dev/null +++ b/githappens/commands/create_issue.py @@ -0,0 +1,85 @@ +from .. import git_utils, gitlab_api, interactive + + +def create_issue(ctx, title, project_id, milestone_id, epic, iteration, settings): + if settings: + issue_type = settings.get("type") or "issue" + return gitlab_api.execute_issue_create( + ctx, + project_id, + title, + settings.get("labels"), + milestone_id, + epic, + iteration, + settings.get("weight"), + settings.get("estimated_time"), + issue_type, + ) + print("No settings in template") + raise SystemExit(2) + + +def start_issue_creation( + ctx, project_id, title, milestone_id, epic, iteration, selected_settings, only_issue +): + inquirer = interactive._inquirer() + estimated_time = inquirer.prompt( + [ + inquirer.Text( + "estimated_time", + message="Estimated time to complete this issue (in minutes, optional)", + validate=lambda _, value: value == "" or value.isdigit(), + ) + ] + )["estimated_time"] + + if isinstance(project_id, list): + estimated_time_per_project = ( + int(estimated_time) / len(project_id) if estimated_time else None + ) + else: + estimated_time_per_project = estimated_time + + if estimated_time_per_project: + selected_settings = selected_settings.copy() if selected_settings else {} + selected_settings["estimated_time"] = int(estimated_time_per_project) + + created_issue = create_issue( + ctx, + title, + project_id, + milestone_id, + epic, + iteration, + selected_settings, + ) + print(f"Issue #{created_issue['iid']}: {created_issue['title']} created.") + + if only_issue: + return created_issue + + created_branch = gitlab_api.create_branch(ctx, project_id, created_issue) + created_mr = gitlab_api.create_merge_request( + ctx, + project_id, + created_branch, + created_issue, + selected_settings.get("labels"), + milestone_id, + ) + print(f"Merge request #{created_mr['iid']}: {created_mr['title']} created.") + print("Run:") + print(" git fetch origin") + print( + f" git checkout -b '{created_mr['source_branch']}' " + f"'origin/{created_mr['source_branch']}'" + ) + print("to switch to new branch.") + + return created_issue + + +def resolve_main_branch(ctx): + ctx.main_branch = git_utils.get_main_branch() + return ctx.main_branch diff --git a/githappens/commands/deploy.py b/githappens/commands/deploy.py new file mode 100644 index 0000000..27442ae --- /dev/null +++ b/githappens/commands/deploy.py @@ -0,0 +1,92 @@ +import datetime + +from .. import git_utils, gitlab_api + + +def get_last_production_deploy(ctx): + try: + project_id = gitlab_api.get_project_id(ctx) + params = { + "per_page": 50, + "order_by": "updated_at", + "sort": "desc", + } + if ctx.main_branch: + params["ref"] = ctx.main_branch + else: + try: + params["ref"] = git_utils.get_main_branch() + except Exception: + params["ref"] = "main" + + response = gitlab_api.get_pipelines(ctx, project_id, params) + if response.status_code != 200: + print(f"Failed to fetch pipelines: {response.status_code} - {response.text}") + return + + production_pipeline = None + for pipeline in response.json(): + detail_response = gitlab_api.get_pipeline_jobs(ctx, project_id, pipeline["id"]) + if detail_response.status_code != 200: + continue + + for job in detail_response.json(): + job_name = job.get("name", "") + stage = job.get("stage", "") + job_status = job.get("status", "").lower() + if job_status != "success": + continue + + project_mapping = ctx.production_mappings.get(str(project_id)) + if project_mapping: + expected_stage = project_mapping.get("stage", "").lower() + expected_job = project_mapping.get("job", "").lower() + if stage.lower() == expected_stage or ( + expected_job and job_name.lower() == expected_job + ): + production_pipeline = { + "pipeline": pipeline, + "production_job": job, + } + break + else: + print("Didn't find deployment pipeline") + + if production_pipeline: + break + + if not production_pipeline: + print("No production deployment found matching pattern") + return + + pipeline = production_pipeline["pipeline"] + job = production_pipeline["production_job"] + print("Last Production Deployment:") + print(f" Pipeline: #{pipeline['id']} - {pipeline['status']}") + print(f" Job: {job['name']} ({job['status']})") + print(f" Branch/Tag: {pipeline['ref']}") + print(f" Started: {job.get('started_at', 'N/A')}") + print(f" Finished: {job.get('finished_at', 'N/A')}") + if job.get("duration"): + print(f" Duration: {job.get('duration')} seconds") + else: + print(" Duration: N/A") + print(f" Commit: {pipeline['sha'][:8]}") + print(f" URL: {pipeline['web_url']}") + + if job.get("finished_at"): + try: + finished_time = datetime.datetime.fromisoformat( + job["finished_at"].replace("Z", "+00:00") + ) + time_diff = datetime.datetime.now(datetime.timezone.utc) - finished_time + if time_diff.days > 0: + print(f" {time_diff.days} days ago") + elif time_diff.seconds > 3600: + print(f" {time_diff.seconds // 3600} hours ago") + else: + print(f" {time_diff.seconds // 60} minutes ago") + except Exception: + pass + except Exception as exc: + print(f"Error fetching last production deploy: {str(exc)}") diff --git a/githappens/commands/open_mr.py b/githappens/commands/open_mr.py new file mode 100644 index 0000000..157f007 --- /dev/null +++ b/githappens/commands/open_mr.py @@ -0,0 +1,21 @@ +import subprocess +import webbrowser + +from .. import git_utils, gitlab_api + + +def get_active_merge_request_id(ctx): + branch_to_find = git_utils.get_current_branch() + return gitlab_api.find_merge_request_id_by_branch(ctx, branch_to_find) + + +def open_merge_request_in_browser(ctx): + try: + merge_request_id = get_active_merge_request_id(ctx) + remote_url = subprocess.check_output( + ["git", "config", "--get", "remote.origin.url"], text=True + ).strip() + url = ctx.settings.base_url + "/" + remote_url.split(":")[1][:-4] + webbrowser.open(f"{url}/-/merge_requests/{merge_request_id}") + except (subprocess.CalledProcessError, TypeError, IndexError): + return None diff --git a/githappens/commands/report.py b/githappens/commands/report.py new file mode 100644 index 0000000..53e776e --- /dev/null +++ b/githappens/commands/report.py @@ -0,0 +1,56 @@ +import subprocess + +from .. import gitlab_api, interactive +from .create_issue import create_issue + + +def process_report(ctx, text, minutes): + incident_project_id = ctx.settings.incident_project_id + if not incident_project_id: + print("Error: incident_project_id not found in config.ini") + print("Please add your incident project ID to configs/config.ini under [DEFAULT]:") + print("incident_project_id = your_project_id_here") + return + + issue_title = f"Incident Report: {text}" + selected_label = interactive.select_labels(ctx, "Department") + + incident_settings = { + "labels": ["incident", "report"], + "onlyIssue": True, + "type": "incident", + } + if selected_label: + incident_settings["labels"].append(selected_label) + + try: + iteration = interactive.get_active_iteration(ctx) + created_issue = create_issue( + ctx, + issue_title, + incident_project_id, + False, + False, + iteration, + incident_settings, + ) + issue_iid = created_issue["iid"] + + gitlab_api.close_opened_issue(incident_project_id, issue_iid) + print(f"Incident issue #{issue_iid} created successfully.") + print(f"Title: {issue_title}") + + command = [ + "glab", + "api", + f"/projects/{incident_project_id}/issues/{issue_iid}/add_spent_time", + "-f", + f"duration={minutes}m", + ] + try: + subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + print(f"Added {minutes} minutes to issue time tracking.") + except subprocess.CalledProcessError as exc: + print(f"Error adding time tracking: {str(exc)}") + except Exception as exc: + print(f"Error creating incident issue: {str(exc)}") diff --git a/githappens/commands/review.py b/githappens/commands/review.py new file mode 100644 index 0000000..6c44bfe --- /dev/null +++ b/githappens/commands/review.py @@ -0,0 +1,68 @@ +import subprocess + +from .. import git_utils, gitlab_api, interactive +from .open_mr import get_active_merge_request_id + + +def get_current_issue_id(ctx): + mr = gitlab_api.get_merge_request_for_branch(ctx, git_utils.get_current_branch()) + return mr["description"].replace('"', "").replace("#", "").split()[1] + + +def track_issue_time(ctx): + try: + project_id = gitlab_api.get_project_id(ctx) + issue_id = get_current_issue_id(ctx) + except Exception as exc: + print(f"Error getting issue details: {str(exc)}") + return + + inquirer = interactive._inquirer() + spent_time = inquirer.prompt( + [ + inquirer.Text( + "spent_time", + message="How many minutes did you actually spend on this issue?", + validate=lambda _, value: value.isdigit(), + ) + ] + )["spent_time"] + + command = [ + "glab", + "api", + f"/projects/{project_id}/issues/{issue_id}/notes", + "-f", + f"body=/spend {spent_time}m", + ] + + try: + subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True) + print(f"Added {spent_time} minutes to issue {issue_id} time tracking.") + except subprocess.CalledProcessError as exc: + print(f"Error adding time tracking: {str(exc)}") + except Exception as exc: + print(f"Error tracking issue time: {str(exc)}") + + +def review_current_merge_request(ctx, select_reviewers=False, auto_merge=False): + track_issue_time(ctx) + reviewers = interactive.choose_reviewers_manually(ctx) if select_reviewers else None + mr_id = get_active_merge_request_id(ctx) + gitlab_api.add_reviewers_to_merge_request(ctx, mr_id, reviewers=reviewers) + + try: + from ai_code_review import run_review_for_mr + + project_id = gitlab_api.get_project_id(ctx) + run_review_for_mr( + project_id, + mr_id, + ctx.settings.gitlab_token, + ctx.settings.api_url, + ) + except Exception as exc: + print(f"AI review skipped: {exc}") + + if auto_merge: + gitlab_api.set_merge_request_to_auto_merge(ctx, mr_id) diff --git a/githappens/commands/summary.py b/githappens/commands/summary.py new file mode 100644 index 0000000..ddb5ce0 --- /dev/null +++ b/githappens/commands/summary.py @@ -0,0 +1,49 @@ +from .. import git_utils + + +def print_commit_summary(ctx): + git_utils.get_two_weeks_commits(ctx) + + +def generate_smart_summary(ctx): + commits = git_utils.get_two_weeks_commits(ctx, return_output=True) + if not commits: + return + + if not ctx.settings.openai_api_key: + print("OpenAI API key not set. Skipping AI summary generation.") + return + + try: + import openai + except ImportError: + print("OpenAI package not installed. Please install it using: pip install openai") + return + + openai.api_key = ctx.settings.openai_api_key + + try: + response = openai.chat.completions.create( + model="gpt-3.5-turbo", + messages=[ + { + "role": "system", + "content": ( + "You are a helpful assistant that summarizes git commits. " + "Provide a concise, well-organized summary of the main " + "changes and themes." + ), + }, + { + "role": "user", + "content": ( + "Please summarize these git commits in a clear, bulleted " + f"format:\n\n{commits}" + ), + }, + ], + ) + print("\nAI-Generated Summary of Recent Changes:\n") + print(response.choices[0].message.content) + except Exception as exc: + print(f"Error generating AI summary: {exc}") diff --git a/githappens/config.py b/githappens/config.py new file mode 100644 index 0000000..bf4a45a --- /dev/null +++ b/githappens/config.py @@ -0,0 +1,113 @@ +from dataclasses import dataclass +from pathlib import Path +import configparser +import json + + +@dataclass +class Settings: + base_url: str + api_url: str + group_id: str + custom_template: str + gitlab_token: str + delete_branch_after_merge: bool + developer_email: str + squash_commits: bool + production_pipeline_name: str + production_job_name: str + production_ref: str + incident_project_id: str + openai_api_key: str + + +@dataclass +class AppContext: + root: Path + settings: Settings + templates: list + reviewers: list + production_mappings: dict + main_branch: str = "master" + + +def project_root() -> Path: + return Path(__file__).resolve().parent.parent + + +def _read_required(parser: configparser.ConfigParser, option: str) -> str: + try: + return parser.get("DEFAULT", option) + except (configparser.NoOptionError, configparser.NoSectionError) as exc: + raise RuntimeError( + f"Missing '{option}' in configs/config.ini. " + "Copy configs/config.ini.example to configs/config.ini and fill it in." + ) from exc + + +def _read_bool(parser: configparser.ConfigParser, option: str) -> bool: + return _read_required(parser, option).lower() == "true" + + +def load_settings(root: Path | None = None) -> Settings: + root = root or project_root() + config_path = root / "configs" / "config.ini" + parser = configparser.ConfigParser() + + if not parser.read(config_path): + raise RuntimeError( + "Missing configs/config.ini. Copy configs/config.ini.example to " + "configs/config.ini and fill it in before running GitHappens." + ) + + base_url = _read_required(parser, "base_url") + return Settings( + base_url=base_url, + api_url=f"{base_url}/api/v4", + group_id=_read_required(parser, "group_id"), + custom_template=_read_required(parser, "custom_template"), + gitlab_token=_read_required(parser, "GITLAB_TOKEN").strip("\"'"), + delete_branch_after_merge=_read_bool(parser, "delete_branch_after_merge"), + developer_email=parser.get("DEFAULT", "developer_email", fallback=None), + squash_commits=_read_bool(parser, "squash_commits"), + production_pipeline_name=parser.get( + "DEFAULT", "production_pipeline_name", fallback="deploy" + ), + production_job_name=parser.get("DEFAULT", "production_job_name", fallback=None), + production_ref=parser.get("DEFAULT", "production_ref", fallback=None), + incident_project_id=parser.get("DEFAULT", "incident_project_id", fallback=None), + openai_api_key=parser.get("DEFAULT", "OPENAI_API_KEY", fallback=None), + ) + + +def load_templates(root: Path | None = None) -> tuple[list, list, dict]: + root = root or project_root() + templates_path = root / "configs" / "templates.json" + + if not templates_path.exists(): + raise RuntimeError( + "Missing configs/templates.json. Copy configs/templates.json.example " + "to configs/templates.json and remove JSON comments before running." + ) + + with templates_path.open("r", encoding="utf-8") as handle: + payload = json.load(handle) + + return ( + payload.get("templates", []), + payload.get("reviewers", []), + payload.get("productionMappings", {}), + ) + + +def load_context(root: Path | None = None) -> AppContext: + root = root or project_root() + settings = load_settings(root) + templates, reviewers, production_mappings = load_templates(root) + return AppContext( + root=root, + settings=settings, + templates=templates, + reviewers=reviewers, + production_mappings=production_mappings, + ) diff --git a/githappens/git_utils.py b/githappens/git_utils.py new file mode 100644 index 0000000..a46e773 --- /dev/null +++ b/githappens/git_utils.py @@ -0,0 +1,84 @@ +import datetime +import re +import subprocess + + +def get_project_link_from_current_dir(): + try: + result = subprocess.run( + ["git", "remote", "get-url", "origin"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + if result.returncode == 0: + return result.stdout.strip() + return -1 + except FileNotFoundError: + return -1 + + +def get_current_branch() -> str: + return subprocess.check_output( + ["git", "rev-parse", "--abbrev-ref", "HEAD"], text=True + ).strip() + + +def get_main_branch() -> str: + output = subprocess.check_output( + ["git", "symbolic-ref", "refs/remotes/origin/HEAD"], + stderr=subprocess.STDOUT, + text=True, + ) + return output.strip().replace("refs/remotes/origin/", "") + + +def build_branch_name(issue: dict) -> str: + issue_id = str(issue["iid"]) + title = re.sub(r"\s+", "-", issue["title"]).lower() + title = title.replace(":", "").replace("(", " ").replace(")", "").replace(" ", "-") + return f"{issue_id}-{title}" + + +def get_two_weeks_commits(ctx, return_output=False): + two_weeks_ago = (datetime.datetime.now() - datetime.timedelta(weeks=2)).strftime( + "%Y-%m-%d" + ) + command = [ + "git", + "log", + f"--since={two_weeks_ago}", + '--format=%ad - %ae - %s', + "--date=short", + ] + + try: + output = subprocess.check_output( + command, + text=True, + stderr=subprocess.DEVNULL, + ).strip() + except subprocess.CalledProcessError as exc: + print(f"No commits were found or an error occurred. (exit status {exc.returncode})") + return "" if return_output else None + except FileNotFoundError: + print("Git is not installed or not found in PATH.") + return "" if return_output else None + + commits = [] + for line in output.splitlines(): + if "Merge branch" in line: + continue + if ctx.settings.developer_email and ctx.settings.developer_email not in line: + continue + commits.append(line) + + if commits: + result = "\n".join(commits) + if return_output: + return result + print(result) + return None + + print("No commits found.") + return "" if return_output else None diff --git a/githappens/gitlab_api.py b/githappens/gitlab_api.py new file mode 100644 index 0000000..8997f85 --- /dev/null +++ b/githappens/gitlab_api.py @@ -0,0 +1,288 @@ +import json +import subprocess + +import requests + +from . import git_utils + + +def private_token_headers(ctx): + return {"Private-Token": ctx.settings.gitlab_token} + + +def get_all_projects(ctx, project_link): + url = ( + f"{ctx.settings.api_url}/projects?membership=true&search=" + f"{project_link.split('/')[-1].split('.')[0]}" + ) + response = requests.get(url, headers={"PRIVATE-TOKEN": ctx.settings.gitlab_token}) + + if response.status_code == 200: + return response.json() + if response.status_code == 401: + print( + "Error: Unauthorized (401). Your GitLab token is probably expired, " + "invalid, or missing required permissions." + ) + print("Please generate a new token and update your configs/config.ini.") + raise SystemExit(1) + + print(f"Request failed with status code {response.status_code}") + return [] + + +def get_project_id(ctx): + project_link = git_utils.get_project_link_from_current_dir() + if project_link == -1: + from .interactive import enter_project_id + + return enter_project_id() + + matching_id = None + for project in get_all_projects(ctx, project_link): + if project.get("ssh_url_to_repo") == project_link: + matching_id = project.get("id") + break + return matching_id + + +def list_milestones(ctx, current=False): + result = subprocess.run( + ["glab", "api", f"/groups/{ctx.settings.group_id}/milestones?state=active"], + stdout=subprocess.PIPE, + text=True, + ) + milestones = json.loads(result.stdout) + if current: + import datetime + + today = datetime.date.today().strftime("%Y-%m-%d") + active_milestones = [ + milestone + for milestone in milestones + if milestone["start_date"] + and milestone["due_date"] + and milestone["start_date"] <= today <= milestone["due_date"] + ] + active_milestones.sort(key=lambda item: item["due_date"]) + return active_milestones[0] + return milestones + + +def list_iterations(ctx): + result = subprocess.run( + ["glab", "api", f"/groups/{ctx.settings.group_id}/iterations?state=opened"], + stdout=subprocess.PIPE, + text=True, + ) + return json.loads(result.stdout) + + +def list_epics(ctx): + result = subprocess.run( + [ + "glab", + "api", + f"/groups/{ctx.settings.group_id}/epics?per_page=1000&state=opened", + ], + stdout=subprocess.PIPE, + text=True, + ) + return json.loads(result.stdout) + + +def get_authorized_user(): + output = subprocess.check_output(["glab", "api", "/user"]) + return json.loads(output) + + +def execute_issue_create( + ctx, + project_id, + title, + labels, + milestone_id, + epic, + iteration, + weight, + estimated_time, + issue_type="issue", +): + labels = ",".join(labels) if isinstance(labels, list) else labels + assignee_id = get_authorized_user()["id"] + issue_command = [ + "glab", + "api", + f"/projects/{str(project_id)}/issues", + "-f", + f"title={title}", + "-f", + f"assignee_ids={assignee_id}", + "-f", + f"issue_type={issue_type}", + ] + + optional_fields = { + "labels": labels, + "weight": str(weight) if weight else None, + "milestone_id": str(milestone_id) if milestone_id else None, + } + + if epic: + optional_fields["epic_id"] = str(epic["id"]) + + description = "" + if iteration: + description += f"/iteration *iteration:{str(iteration['id'])} " + if estimated_time: + description += f"\n/estimate {estimated_time}m " + optional_fields["description"] = description + + for field, value in optional_fields.items(): + if value: + issue_command.extend(["-f", f"{field}={value}"]) + + issue_output = subprocess.check_output(issue_command) + return json.loads(issue_output.decode()) + + +def create_branch(ctx, project_id, issue): + issue_id = str(issue["iid"]) + branch_name = git_utils.build_branch_name(issue) + branch_output = subprocess.check_output( + [ + "glab", + "api", + f"/projects/{str(project_id)}/repository/branches", + "-f", + f"branch={branch_name}", + "-f", + f"ref={ctx.main_branch}", + "-f", + f"issue_iid={issue_id}", + ] + ) + return json.loads(branch_output.decode()) + + +def create_merge_request(ctx, project_id, branch, issue, labels, milestone_id): + issue_id = str(issue["iid"]) + branch_name = branch["name"] + assignee_id = get_authorized_user()["id"] + labels = ",".join(labels) if isinstance(labels, list) else labels + command = [ + "glab", + "api", + f"/projects/{str(project_id)}/merge_requests", + "-f", + f"title={issue['title']}", + "-f", + f'description="Closes #{issue_id}"', + "-f", + f"source_branch={branch_name}", + "-f", + f"target_branch={ctx.main_branch}", + "-f", + f"issue_iid={issue_id}", + "-f", + f"assignee_ids={assignee_id}", + ] + + if ctx.settings.squash_commits: + command.extend(["-f", "squash=true"]) + if ctx.settings.delete_branch_after_merge: + command.extend(["-f", "remove_source_branch=true"]) + if labels: + command.extend(["-f", f"labels={labels}"]) + if milestone_id: + command.extend(["-f", f"milestone_id={str(milestone_id)}"]) + + mr_output = subprocess.check_output(command) + return json.loads(mr_output.decode()) + + +def get_merge_request_for_branch(ctx, branch_name): + project_id = get_project_id(ctx) + api_url = f"{ctx.settings.api_url}/projects/{project_id}/merge_requests" + response = requests.get( + api_url, + headers=private_token_headers(ctx), + params={"source_branch": branch_name}, + ) + if response.status_code == 200: + for merge_request in response.json(): + if merge_request["source_branch"] == branch_name: + return merge_request + else: + print(f"Failed to fetch Merge Requests: {response.status_code} - {response.text}") + return None + + +def find_merge_request_id_by_branch(ctx, branch_name): + return get_merge_request_for_branch(ctx, branch_name)["iid"] + + +def add_reviewers_to_merge_request(ctx, mr_id, reviewers=None): + project_id = get_project_id(ctx) + api_url = f"{ctx.settings.api_url}/projects/{project_id}/merge_requests/{mr_id}" + data = {"reviewer_ids": reviewers if reviewers is not None else ctx.reviewers} + requests.put(api_url, headers=private_token_headers(ctx), json=data) + + +def set_merge_request_to_auto_merge(ctx, mr_id): + project_id = get_project_id(ctx) + api_url = f"{ctx.settings.api_url}/projects/{project_id}/merge_requests/{mr_id}/merge" + data = { + "id": project_id, + "merge_request_iid": mr_id, + "should_remove_source_branch": True, + "merge_when_pipeline_succeeds": True, + "auto_merge_strategy": "merge_when_pipeline_succeeds", + } + requests.put(api_url, headers=private_token_headers(ctx), json=data) + + +def get_labels_of_group(ctx, search=""): + try: + result = subprocess.run( + ["glab", "api", f"/groups/{ctx.settings.group_id}/labels?search={search}"], + stdout=subprocess.PIPE, + check=True, + text=True, + ) + return json.loads(result.stdout) + except subprocess.CalledProcessError as exc: + print(f"Error getting labels: {str(exc)}") + return [] + + +def close_opened_issue(project_id, issue_iid): + command = [ + "glab", + "api", + f"/projects/{project_id}/issues/{issue_iid}", + "-X", + "PUT", + "-f", + "state_event=close", + ] + try: + subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + except subprocess.CalledProcessError as exc: + print(f"Error closing issue: {str(exc)}") + + +def get_pipelines(ctx, project_id, params): + response = requests.get( + f"{ctx.settings.api_url}/projects/{project_id}/pipelines", + headers=private_token_headers(ctx), + params=params, + ) + return response + + +def get_pipeline_jobs(ctx, project_id, pipeline_id): + return requests.get( + f"{ctx.settings.api_url}/projects/{project_id}/pipelines/{pipeline_id}/jobs", + headers=private_token_headers(ctx), + ) diff --git a/githappens/interactive.py b/githappens/interactive.py new file mode 100644 index 0000000..ea31cae --- /dev/null +++ b/githappens/interactive.py @@ -0,0 +1,194 @@ +import datetime + +import requests + +from . import gitlab_api + + +def _inquirer(): + try: + import inquirer + except ModuleNotFoundError as exc: + raise SystemExit( + "Missing dependency 'inquirer'. Run: pip install -r requirements.txt" + ) from exc + return inquirer + + +def enter_project_id(): + while True: + project_id = input("Please enter the ID of your GitLab project: ") + if project_id: + return project_id + raise SystemExit("Invalid project ID.") + + +def select_template(ctx): + inquirer = _inquirer() + template_names = [template["name"] for template in ctx.templates] + template_names.append(ctx.settings.custom_template) + answer = inquirer.prompt( + [ + inquirer.List( + "template", + message="Select template:", + choices=template_names, + ) + ] + ) + return answer["template"] + + +def select_milestone(milestones): + inquirer = _inquirer() + choices = [milestone["title"] for milestone in milestones] + answer = inquirer.prompt( + [ + inquirer.List( + "milestones", + message="Select milestone:", + choices=choices, + ) + ] + ) + return answer["milestones"] + + +def get_selected_milestone(title, milestones): + return next((milestone for milestone in milestones if milestone["title"] == title), None) + + +def get_milestone(ctx, manual): + if manual: + milestones = gitlab_api.list_milestones(ctx) + return get_selected_milestone(select_milestone(milestones), milestones) + return gitlab_api.list_milestones(ctx, current=True) + + +def select_iteration(iterations): + inquirer = _inquirer() + choices = [ + f"{iteration['start_date']} - {iteration['due_date']}" for iteration in iterations + ] + answer = inquirer.prompt( + [ + inquirer.List( + "iterations", + message="Select iteration:", + choices=choices, + ) + ] + ) + return answer["iterations"] + + +def get_selected_iteration(iteration_label, iterations): + return next( + ( + iteration + for iteration in iterations + if f"{iteration['start_date']} - {iteration['due_date']}" == iteration_label + ), + None, + ) + + +def get_active_iteration(ctx): + iterations = gitlab_api.list_iterations(ctx) + today = datetime.date.today().strftime("%Y-%m-%d") + active_iterations = [ + iteration + for iteration in iterations + if iteration["start_date"] + and iteration["due_date"] + and iteration["start_date"] <= today <= iteration["due_date"] + ] + active_iterations.sort(key=lambda item: item["due_date"]) + return active_iterations[0] + + +def get_iteration(ctx, manual): + if manual: + iterations = gitlab_api.list_iterations(ctx) + return get_selected_iteration(select_iteration(iterations), iterations) + return get_active_iteration(ctx) + + +def select_epic(epics): + inquirer = _inquirer() + epic_titles = [epic["title"] for epic in epics] + search_query = inquirer.prompt( + [inquirer.Text("search_query", message="Search epic:")] + )["search_query"] + filtered_epics = [ + title for title in epic_titles if search_query.lower() in title.lower() + ] + answer = inquirer.prompt( + [ + inquirer.List( + "epics", + message="Select epic:", + choices=filtered_epics, + ) + ] + ) + return answer["epics"] + + +def get_selected_epic(epic_title, epics): + return next((epic for epic in epics if epic["title"] == epic_title), None) + + +def get_epic(ctx): + epics = gitlab_api.list_epics(ctx) + return get_selected_epic(select_epic(epics), epics) + + +def choose_reviewers_manually(ctx): + """Prompt the user to select reviewers manually from configured reviewers.""" + inquirer = _inquirer() + reviewer_choices = [] + for reviewer_id in ctx.reviewers: + api_url = f"{ctx.settings.api_url}/users/{reviewer_id}" + try: + response = requests.get( + api_url, headers=gitlab_api.private_token_headers(ctx) + ) + if response.status_code == 200: + user = response.json() + display_name = f"{user.get('name')} ({user.get('username')})" + reviewer_choices.append((display_name, str(reviewer_id))) + else: + reviewer_choices.append((str(reviewer_id), str(reviewer_id))) + except Exception: + reviewer_choices.append((str(reviewer_id), str(reviewer_id))) + + answers = inquirer.prompt( + [ + inquirer.Checkbox( + "selected_reviewers", + message="Select reviewers", + choices=reviewer_choices, + ) + ] + ) + if answers and "selected_reviewers" in answers: + return [int(reviewer_id) for reviewer_id in answers["selected_reviewers"]] + return [] + + +def select_labels(ctx, search, multiple=False): + inquirer = _inquirer() + labels = gitlab_api.get_labels_of_group(ctx, search) + choices = sorted([label["name"] for label in labels]) + question_type = inquirer.Checkbox if multiple else inquirer.List + answer = inquirer.prompt( + [ + question_type( + "labels", + message="Select one or more department labels:", + choices=choices, + ) + ] + ) + return answer["labels"] diff --git a/githappens/main.py b/githappens/main.py new file mode 100644 index 0000000..97cba24 --- /dev/null +++ b/githappens/main.py @@ -0,0 +1,157 @@ +import argparse +import sys + +from .config import load_context +from . import gitlab_api, interactive, templates +from .commands import create_issue, deploy, open_mr, report, review, summary + + +def build_parser(): + parser = argparse.ArgumentParser("Argument description of Git happens") + parser.add_argument("title", nargs="+", help="Title of issue") + parser.add_argument("--project_id", type=str, help="Id or URL-encoded path of project") + parser.add_argument( + "-m", + "--milestone", + action="store_true", + help="Add this flag, if you want to manually select milestone", + ) + parser.add_argument( + "--no_epic", + action="store_true", + help="Add this flag if you don't want to pick epic", + ) + parser.add_argument( + "--no_milestone", + action="store_true", + help="Add this flag if you don't want to pick milestone", + ) + parser.add_argument( + "--no_iteration", + action="store_true", + help="Add this flag if you don't want to pick iteration", + ) + parser.add_argument( + "--only_issue", + action="store_true", + help="Add this flag if you don't want to create merge request and branch alongside issue", + ) + parser.add_argument( + "-am", + "--auto_merge", + action="store_true", + help="Add this flag to review if you want to set merge request to auto merge when pipeline succeeds", + ) + parser.add_argument( + "--select", + action="store_true", + help="Manually select reviewers for merge request (interactive)", + ) + return parser + + +def main(argv=None): + argv = sys.argv[1:] if argv is None else argv + parser = build_parser() + if not argv: + parser.print_help() + raise SystemExit(1) + + if any(arg in ("-h", "--help") for arg in argv): + parser.parse_args(argv) + return + + try: + ctx = load_context() + except RuntimeError as exc: + print(exc) + raise SystemExit(1) from exc + + args = parser.parse_args(argv) + title = " ".join(args.title) + + if args.title[0] == "report": + if len(args.title) != 3: + print('Invalid report format. Use: gh report "text" minutes') + return + try: + report.process_report(ctx, args.title[1], int(args.title[2].strip())) + except ValueError: + print("Invalid minutes. Please provide a valid number.") + return + + if title == "open": + open_mr.open_merge_request_in_browser(ctx) + return + if title == "review": + review.review_current_merge_request( + ctx, select_reviewers=args.select, auto_merge=args.auto_merge + ) + return + if title == "summary": + summary.print_commit_summary(ctx) + return + if title == "summaryAI": + summary.generate_smart_summary(ctx) + return + if title == "last deploy": + deploy.get_last_production_deploy(ctx) + return + if title == "ai review": + from ai_code_review import run_review + + run_review() + return + + selected_settings = templates.get_issue_settings( + ctx, interactive.select_template(ctx) + ) + + if not len(selected_settings): + print("Custom selection of issue settings is not supported yet") + + if args.project_id and selected_settings.get("projectIds"): + print("NOTE: Overwriting project id from argument...") + + project_id = ( + selected_settings.get("projectIds") or args.project_id or gitlab_api.get_project_id(ctx) + ) + + milestone = False + if not args.no_milestone: + milestone = interactive.get_milestone(ctx, args.milestone)["id"] + + iteration = False + if not args.no_iteration: + iteration = interactive.get_iteration(ctx, True) + + epic = False + if not args.no_epic: + epic = interactive.get_epic(ctx) + + create_issue.resolve_main_branch(ctx) + only_issue = selected_settings.get("onlyIssue") or args.only_issue + + if isinstance(project_id, list): + for selected_project_id in project_id: + create_issue.start_issue_creation( + ctx, + selected_project_id, + title, + milestone, + epic, + iteration, + selected_settings, + only_issue, + ) + else: + create_issue.start_issue_creation( + ctx, + project_id, + title, + milestone, + epic, + iteration, + selected_settings, + only_issue, + ) diff --git a/githappens/templates.py b/githappens/templates.py new file mode 100644 index 0000000..2db3610 --- /dev/null +++ b/githappens/templates.py @@ -0,0 +1,4 @@ +def get_issue_settings(ctx, template_name): + if template_name == ctx.settings.custom_template: + return {} + return next((template for template in ctx.templates if template["name"] == template_name), None) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..284bd8d --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,19 @@ +import unittest + +from githappens.main import build_parser + + +class CliTest(unittest.TestCase): + def test_parser_collects_title_and_flags(self): + args = build_parser().parse_args( + ["Fix", "login", "--only_issue", "--no_epic", "--project_id", "123"] + ) + + self.assertEqual(args.title, ["Fix", "login"]) + self.assertTrue(args.only_issue) + self.assertTrue(args.no_epic) + self.assertEqual(args.project_id, "123") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_commands.py b/tests/test_commands.py new file mode 100644 index 0000000..9bb181f --- /dev/null +++ b/tests/test_commands.py @@ -0,0 +1,218 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from githappens.commands import create_issue, deploy, open_mr, report, review, summary + + +class FakePrompt: + class Text: + def __init__(self, *args, **kwargs): + pass + + @staticmethod + def prompt(_): + return {"estimated_time": "30", "spent_time": "45"} + + +class IssuesCommandTest(unittest.TestCase): + @patch("githappens.commands.create_issue.gitlab_api.create_merge_request") + @patch("githappens.commands.create_issue.gitlab_api.create_branch") + @patch("githappens.commands.create_issue.gitlab_api.execute_issue_create") + @patch("githappens.commands.create_issue.interactive._inquirer", return_value=FakePrompt) + def test_start_issue_creation_creates_issue_branch_and_mr( + self, _, execute_issue_create, create_branch, create_merge_request + ): + ctx = SimpleNamespace(main_branch="main") + execute_issue_create.return_value = {"iid": 10, "title": "Fix bug"} + create_branch.return_value = {"name": "10-fix-bug"} + create_merge_request.return_value = { + "iid": 20, + "title": "Fix bug", + "source_branch": "10-fix-bug", + } + + with patch("builtins.print"): + created_issue = create_issue.start_issue_creation( + ctx, + project_id=123, + title="Fix bug", + milestone_id=5, + epic=None, + iteration=None, + selected_settings={"labels": ["bug"]}, + only_issue=False, + ) + + self.assertEqual(created_issue["iid"], 10) + execute_issue_create.assert_called_once() + self.assertEqual(execute_issue_create.call_args.args[8], 30) + create_branch.assert_called_once_with(ctx, 123, created_issue) + create_merge_request.assert_called_once() + + @patch("githappens.commands.create_issue.gitlab_api.create_branch") + @patch("githappens.commands.create_issue.gitlab_api.execute_issue_create") + @patch("githappens.commands.create_issue.interactive._inquirer", return_value=FakePrompt) + def test_start_issue_creation_can_stop_after_issue( + self, _, execute_issue_create, create_branch + ): + execute_issue_create.return_value = {"iid": 11, "title": "Docs"} + + with patch("builtins.print"): + result = create_issue.start_issue_creation( + SimpleNamespace(), + 123, + "Docs", + False, + None, + None, + {"labels": ["docs"]}, + True, + ) + + self.assertEqual(result["iid"], 11) + create_branch.assert_not_called() + + +class OpenMrCommandTest(unittest.TestCase): + @patch("githappens.commands.open_mr.webbrowser.open") + @patch("githappens.commands.open_mr.subprocess.check_output") + @patch("githappens.commands.open_mr.get_active_merge_request_id", return_value=7) + def test_open_merge_request_builds_url_from_remote(self, _, check_output, browser_open): + check_output.return_value = "git@gitlab.example.test:group/project.git\n" + ctx = SimpleNamespace(settings=SimpleNamespace(base_url="https://gitlab.example.test")) + + open_mr.open_merge_request_in_browser(ctx) + + browser_open.assert_called_once_with( + "https://gitlab.example.test/group/project/-/merge_requests/7" + ) + + +class ReviewCommandTest(unittest.TestCase): + @patch("githappens.commands.review.subprocess.run") + @patch("githappens.commands.review.interactive._inquirer", return_value=FakePrompt) + @patch("githappens.commands.review.get_current_issue_id", return_value="10") + @patch("githappens.commands.review.gitlab_api.get_project_id", return_value=123) + def test_track_issue_time_posts_spent_time_note(self, _, __, ___, run): + with patch("builtins.print"): + review.track_issue_time(SimpleNamespace()) + + run.assert_called_once() + command = run.call_args.args[0] + self.assertIn("/projects/123/issues/10/notes", command) + self.assertIn("body=/spend 45m", command) + + @patch("ai_code_review.run_review_for_mr") + @patch("githappens.commands.review.gitlab_api.get_project_id", return_value=123) + @patch("githappens.commands.review.gitlab_api.set_merge_request_to_auto_merge") + @patch("githappens.commands.review.gitlab_api.add_reviewers_to_merge_request") + @patch("githappens.commands.review.get_active_merge_request_id", return_value=20) + @patch("githappens.commands.review.track_issue_time") + def test_review_current_merge_request_adds_reviewers_and_auto_merge( + self, track_issue_time, _, add_reviewers, auto_merge, get_project_id, run_review + ): + ctx = SimpleNamespace( + reviewers=[1, 2], + settings=SimpleNamespace(gitlab_token="token", api_url="https://gitlab/api/v4"), + ) + + with patch("builtins.print"): + review.review_current_merge_request(ctx, auto_merge=True) + + track_issue_time.assert_called_once_with(ctx) + add_reviewers.assert_called_once_with(ctx, 20, reviewers=None) + get_project_id.assert_called_once_with(ctx) + run_review.assert_called_once() + auto_merge.assert_called_once_with(ctx, 20) + + +class ReportCommandTest(unittest.TestCase): + @patch("githappens.commands.report.subprocess.run") + @patch("githappens.commands.report.gitlab_api.close_opened_issue") + @patch("githappens.commands.report.create_issue") + @patch("githappens.commands.report.interactive.get_active_iteration") + @patch("githappens.commands.report.interactive.select_labels", return_value="Support") + def test_process_report_creates_closes_and_tracks_incident( + self, _, get_active_iteration, create_issue, close_opened_issue, run + ): + ctx = SimpleNamespace(settings=SimpleNamespace(incident_project_id="incidents")) + get_active_iteration.return_value = {"id": 3} + create_issue.return_value = {"iid": 99} + + with patch("builtins.print"): + report.process_report(ctx, "Checkout outage", 15) + + settings = create_issue.call_args.args[6] + self.assertEqual(settings["labels"], ["incident", "report", "Support"]) + self.assertEqual(settings["type"], "incident") + close_opened_issue.assert_called_once_with("incidents", 99) + self.assertIn("duration=15m", run.call_args.args[0]) + + @patch("githappens.commands.report.create_issue") + def test_process_report_returns_when_incident_project_missing(self, create_issue): + ctx = SimpleNamespace(settings=SimpleNamespace(incident_project_id=None)) + + with patch("builtins.print"): + report.process_report(ctx, "Checkout outage", 15) + + create_issue.assert_not_called() + + +class SummaryCommandTest(unittest.TestCase): + @patch("githappens.commands.summary.git_utils.get_two_weeks_commits") + def test_print_commit_summary_delegates_to_git_helper(self, get_commits): + ctx = SimpleNamespace() + + summary.print_commit_summary(ctx) + + get_commits.assert_called_once_with(ctx) + + +class DeployCommandTest(unittest.TestCase): + @patch("githappens.commands.deploy.gitlab_api.get_pipeline_jobs") + @patch("githappens.commands.deploy.gitlab_api.get_pipelines") + @patch("githappens.commands.deploy.gitlab_api.get_project_id", return_value=123) + def test_get_last_production_deploy_uses_project_mapping( + self, _, get_pipelines, get_pipeline_jobs + ): + ctx = SimpleNamespace( + main_branch="main", + production_mappings={"123": {"stage": "deploy", "job": "production"}}, + ) + get_pipelines.return_value = SimpleNamespace( + status_code=200, + json=lambda: [ + { + "id": 8, + "status": "success", + "ref": "main", + "sha": "abcdef123456", + "web_url": "https://gitlab/pipeline/8", + } + ], + ) + get_pipeline_jobs.return_value = SimpleNamespace( + status_code=200, + json=lambda: [ + { + "name": "production", + "stage": "deploy", + "status": "success", + "started_at": "2026-05-01T00:00:00Z", + "finished_at": "2026-05-01T00:01:00Z", + "duration": 60, + } + ], + ) + + with patch("builtins.print"): + deploy.get_last_production_deploy(ctx) + + get_pipelines.assert_called_once() + self.assertEqual(get_pipelines.call_args.args[2]["ref"], "main") + get_pipeline_jobs.assert_called_once_with(ctx, 123, 8) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..c3f461a --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,53 @@ +import json +import tempfile +import unittest +from pathlib import Path + +from githappens.config import load_context + + +class LoadContextTest(unittest.TestCase): + def test_loads_settings_templates_reviewers_and_mappings(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + configs = root / "configs" + configs.mkdir() + (configs / "config.ini").write_text( + "\n".join( + [ + "[DEFAULT]", + "base_url=https://gitlab.example.test", + "group_id=123", + "custom_template=Custom", + "GITLAB_TOKEN='secret'", + "squash_commits=true", + "delete_branch_after_merge=false", + "developer_email=dev@example.test", + ] + ), + encoding="utf-8", + ) + (configs / "templates.json").write_text( + json.dumps( + { + "templates": [{"name": "Bug", "labels": ["bug"]}], + "reviewers": [1, 2], + "productionMappings": {"123": {"stage": "deploy"}}, + } + ), + encoding="utf-8", + ) + + ctx = load_context(root) + + self.assertEqual(ctx.settings.base_url, "https://gitlab.example.test") + self.assertEqual(ctx.settings.api_url, "https://gitlab.example.test/api/v4") + self.assertEqual(ctx.settings.gitlab_token, "secret") + self.assertFalse(ctx.settings.delete_branch_after_merge) + self.assertEqual(ctx.templates[0]["name"], "Bug") + self.assertEqual(ctx.reviewers, [1, 2]) + self.assertIn("123", ctx.production_mappings) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_git_utils.py b/tests/test_git_utils.py new file mode 100644 index 0000000..e6a9d3a --- /dev/null +++ b/tests/test_git_utils.py @@ -0,0 +1,30 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from githappens import git_utils + + +class GitUtilsTest(unittest.TestCase): + def test_build_branch_name_matches_existing_slug_style(self): + issue = {"iid": 42, "title": "Fix: Login Flow (Mobile)"} + self.assertEqual(git_utils.build_branch_name(issue), "42-fix-login-flow--mobile") + + @patch("githappens.git_utils.subprocess.check_output") + def test_get_two_weeks_commits_filters_merge_commits_and_developer(self, check_output): + check_output.return_value = "\n".join( + [ + "2026-05-01 - dev@example.test - Add feature", + "2026-05-02 - dev@example.test - Merge branch main", + "2026-05-03 - other@example.test - Fix bug", + ] + ) + ctx = SimpleNamespace(settings=SimpleNamespace(developer_email="dev@example.test")) + + output = git_utils.get_two_weeks_commits(ctx, return_output=True) + + self.assertEqual(output, "2026-05-01 - dev@example.test - Add feature") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gitlab_api.py b/tests/test_gitlab_api.py new file mode 100644 index 0000000..3b0a8e7 --- /dev/null +++ b/tests/test_gitlab_api.py @@ -0,0 +1,70 @@ +import json +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from githappens import gitlab_api + + +class GitlabApiTest(unittest.TestCase): + @patch("githappens.gitlab_api.get_authorized_user", return_value={"id": 7}) + @patch("githappens.gitlab_api.subprocess.check_output") + def test_execute_issue_create_builds_command_payload(self, check_output, _): + check_output.return_value = json.dumps({"iid": 10}).encode() + + issue = gitlab_api.execute_issue_create( + SimpleNamespace(), + project_id=123, + title="Fix bug", + labels=["bug", "ui"], + milestone_id=5, + epic={"id": 9}, + iteration={"id": 11}, + weight=3, + estimated_time=30, + ) + + command = check_output.call_args.args[0] + self.assertEqual(issue, {"iid": 10}) + self.assertIn("/projects/123/issues", command) + self.assertIn("title=Fix bug", command) + self.assertIn("assignee_ids=7", command) + self.assertIn("labels=bug,ui", command) + self.assertIn("milestone_id=5", command) + self.assertIn("epic_id=9", command) + self.assertIn("weight=3", command) + self.assertIn("description=/iteration *iteration:11 \n/estimate 30m ", command) + + @patch("githappens.gitlab_api.get_authorized_user", return_value={"id": 7}) + @patch("githappens.gitlab_api.subprocess.check_output") + def test_create_merge_request_honors_configured_flags(self, check_output, _): + check_output.return_value = json.dumps({"iid": 20}).encode() + ctx = SimpleNamespace( + main_branch="main", + settings=SimpleNamespace( + squash_commits=True, + delete_branch_after_merge=True, + ), + ) + + mr = gitlab_api.create_merge_request( + ctx, + project_id=123, + branch={"name": "10-fix-bug"}, + issue={"iid": 10, "title": "Fix bug"}, + labels=["bug"], + milestone_id=5, + ) + + command = check_output.call_args.args[0] + self.assertEqual(mr, {"iid": 20}) + self.assertIn("/projects/123/merge_requests", command) + self.assertIn("source_branch=10-fix-bug", command) + self.assertIn("target_branch=main", command) + self.assertIn("squash=true", command) + self.assertIn("remove_source_branch=true", command) + self.assertIn("labels=bug", command) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_interactive.py b/tests/test_interactive.py new file mode 100644 index 0000000..9b1e4f2 --- /dev/null +++ b/tests/test_interactive.py @@ -0,0 +1,30 @@ +import unittest + +from githappens import interactive + + +class InteractiveSelectionTest(unittest.TestCase): + def test_selected_milestone_iteration_and_epic_match_labels(self): + milestones = [{"id": 1, "title": "Sprint 1"}, {"id": 2, "title": "Sprint 2"}] + iterations = [ + {"id": 3, "start_date": "2026-05-01", "due_date": "2026-05-15"}, + {"id": 4, "start_date": "2026-05-16", "due_date": "2026-05-31"}, + ] + epics = [{"id": 5, "title": "Checkout"}, {"id": 6, "title": "Login"}] + + self.assertEqual( + interactive.get_selected_milestone("Sprint 2", milestones), + {"id": 2, "title": "Sprint 2"}, + ) + self.assertEqual( + interactive.get_selected_iteration("2026-05-01 - 2026-05-15", iterations), + {"id": 3, "start_date": "2026-05-01", "due_date": "2026-05-15"}, + ) + self.assertEqual( + interactive.get_selected_epic("Checkout", epics), + {"id": 5, "title": "Checkout"}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_templates.py b/tests/test_templates.py new file mode 100644 index 0000000..9c1b16b --- /dev/null +++ b/tests/test_templates.py @@ -0,0 +1,26 @@ +import unittest +from types import SimpleNamespace + +from githappens.templates import get_issue_settings + + +class TemplatesTest(unittest.TestCase): + def test_custom_template_returns_empty_settings(self): + ctx = SimpleNamespace( + settings=SimpleNamespace(custom_template="Custom"), + templates=[{"name": "Bug", "labels": ["bug"]}], + ) + + self.assertEqual(get_issue_settings(ctx, "Custom"), {}) + + def test_named_template_returns_matching_settings(self): + ctx = SimpleNamespace( + settings=SimpleNamespace(custom_template="Custom"), + templates=[{"name": "Bug", "labels": ["bug"]}], + ) + + self.assertEqual(get_issue_settings(ctx, "Bug"), {"name": "Bug", "labels": ["bug"]}) + + +if __name__ == "__main__": + unittest.main()