""" Azure DevOps Test Management Tool A Streamlit application for managing test work items in Azure DevOps. """ import streamlit as st import requests import base64 import json import pandas as pd import io from typing import List, Dict, Optional, Any # Page configuration st.set_page_config( page_title="Azure DevOps Test Manager", page_icon="๐งช", layout="wide", initial_sidebar_state="expanded" ) # Custom CSS for dark theme styling st.markdown(""" """, unsafe_allow_html=True) class AzureDevOpsClient: """Client for interacting with Azure DevOps API.""" def __init__(self, organization: str, pat: str, project: Optional[str] = None): self.organization = organization self.project = project self.pat = pat self.base_url = f"https://dev.azure.com/{organization}" self.auth_header = self._get_auth_header() self.debug_info = [] # Store debug information def _get_auth_header(self) -> Dict[str, str]: """Create authorization header with PAT.""" credentials = base64.b64encode(f":{self.pat}".encode()).decode() return { "Authorization": f"Basic {credentials}", "Content-Type": "application/json" } def test_connection(self) -> bool: """Test if the connection to Azure DevOps is valid.""" url = f"{self.base_url}/_apis/projects?top=1&api-version=7.0" try: response = requests.get(url, headers=self.auth_header) response.raise_for_status() return True except requests.exceptions.RequestException: return False def get_projects(self) -> List[Dict]: """Fetch all projects from the organization.""" url = f"{self.base_url}/_apis/projects?api-version=7.0" try: response = requests.get(url, headers=self.auth_header) response.raise_for_status() return response.json().get("value", []) except requests.exceptions.RequestException as e: st.error(f"Error fetching projects: {str(e)}") return [] def get_work_items(self, wiql_query: Optional[str] = None, work_item_type: Optional[str] = None, iteration_path: Optional[str] = None, area_path: Optional[str] = None, debug: bool = False) -> List[Dict]: """Fetch work items using WIQL query or get all work items.""" self.debug_info = [] # Reset debug info if not self.project: st.error("No project selected!") return [] # Build query - simplified to get all work items first if wiql_query is None: # Build WHERE clause conditions conditions = ["[System.TeamProject] = @project"] if work_item_type and work_item_type != "All Types": conditions.append(f"[System.WorkItemType] = '{work_item_type}'") if iteration_path and iteration_path != "All Iterations": conditions.append(f"[System.IterationPath] = '{iteration_path}'") if area_path and area_path != "All Areas": conditions.append(f"[System.AreaPath] = '{area_path}'") where_clause = " AND ".join(conditions) wiql_query = f"""SELECT [System.Id], [System.Title], [System.State], [System.WorkItemType], [System.IterationPath] FROM workitems WHERE {where_clause} ORDER BY [System.ChangedDate] DESC""" url = f"{self.base_url}/{self.project}/_apis/wit/wiql?api-version=7.0" if debug: self.debug_info.append(f"URL: {url}") self.debug_info.append(f"Query: {wiql_query}") self.debug_info.append(f"Project: {self.project}") try: response = requests.post( url, headers=self.auth_header, json={"query": wiql_query} ) if debug: self.debug_info.append(f"Status Code: {response.status_code}") response.raise_for_status() result = response.json() if debug: self.debug_info.append(f"Query returned {len(result.get('workItems', []))} work items") work_item_ids = [item["id"] for item in result.get("workItems", [])] if not work_item_ids: if debug: self.debug_info.append("No work item IDs returned from query") return [] # Fetch detailed work item information return self._get_work_item_details(work_item_ids, debug) except requests.exceptions.RequestException as e: error_msg = f"Error fetching work items: {str(e)}" # Check for 401 error specifically if hasattr(e, 'response') and e.response is not None and e.response.status_code == 401: st.error("๐ **Authentication Error (401)**") st.markdown(""" Your PAT (Personal Access Token) doesn't have the required permissions. **Required PAT Scopes:** - โ Work Items: **Read & Write** - โ Project and Team: **Read** **How to create a new PAT:** 1. Go to: https://dev.azure.com/{org}/_usersSettings/tokens 2. Click **"New Token"** 3. Give it a name (e.g., "Test Manager") 4. Set expiration 5. Under **Scopes**, select: - **Work Items**: Read & Write - **Project and Team**: Read 6. Click **Create** and copy the token 7. Paste it in the sidebar and reconnect """) else: st.error(error_msg) if debug: self.debug_info.append(error_msg) if hasattr(e, 'response') and e.response is not None: self.debug_info.append(f"Status Code: {e.response.status_code}") self.debug_info.append(f"Response: {e.response.text[:500]}") return [] def get_work_item_types(self) -> List[str]: """Get available work item types for the project.""" if not self.project: return [] url = f"{self.base_url}/{self.project}/_apis/wit/workitemtypes?api-version=7.0" try: response = requests.get(url, headers=self.auth_header) response.raise_for_status() types = response.json().get("value", []) type_names = [t.get("name", "") for t in types if t.get("name")] return sorted(type_names) except requests.exceptions.RequestException as e: st.error(f"Error fetching work item types: {str(e)}") return [] def _get_work_item_details(self, work_item_ids: List[int], debug: bool = False) -> List[Dict]: """Get detailed information for work items.""" if not work_item_ids: return [] # Azure DevOps has a limit on URL length, so batch if needed batch_size = 200 all_items = [] for i in range(0, len(work_item_ids), batch_size): batch_ids = work_item_ids[i:i + batch_size] ids_str = ",".join(map(str, batch_ids)) url = f"{self.base_url}/_apis/wit/workitems?ids={ids_str}&$expand=all&api-version=7.0" try: response = requests.get(url, headers=self.auth_header) response.raise_for_status() batch_items = response.json().get("value", []) all_items.extend(batch_items) if debug: self.debug_info.append(f"Fetched {len(batch_items)} work items in batch {i//batch_size + 1}") except requests.exceptions.RequestException as e: error_msg = f"Error fetching work item details for batch: {str(e)}" st.error(error_msg) if debug: self.debug_info.append(error_msg) return all_items def update_work_item_status(self, work_item_id: int, new_state: str, comment: Optional[str] = None) -> bool: """Update work item state and optionally add a comment.""" url = f"{self.base_url}/_apis/wit/workitems/{work_item_id}?api-version=7.0" # Prepare the patch document patch_document = [ { "op": "add", "path": "/fields/System.State", "value": new_state } ] try: response = requests.patch( url, headers={**self.auth_header, "Content-Type": "application/json-patch+json"}, json=patch_document ) response.raise_for_status() # Add comment if provided if comment: self._add_comment(work_item_id, comment) return True except requests.exceptions.RequestException as e: st.error(f"Error updating work item {work_item_id}: {str(e)}") if hasattr(e, 'response') and e.response is not None: st.error(f"Response: {e.response.text}") return False def _add_comment(self, work_item_id: int, comment: str) -> bool: """Add a comment to a work item.""" url = f"{self.base_url}/_apis/wit/workitems/{work_item_id}/comments?api-version=7.0" try: response = requests.post( url, headers=self.auth_header, json={"text": comment} ) response.raise_for_status() return True except requests.exceptions.RequestException as e: st.error(f"Error adding comment to work item {work_item_id}: {str(e)}") return False def create_work_item(self, title: str, description: str = "", work_item_type: str = "Task", priority: int = 2, assigned_to: str = "", tags: str = "", iteration: str = "", area: str = "") -> Optional[Dict]: """Create a new work item.""" if not self.project: st.error("No project selected!") return None url = f"{self.base_url}/{self.project}/_apis/wit/workitems/${work_item_type}?api-version=7.0" # Prepare the patch document patch_document = [ { "op": "add", "path": "/fields/System.Title", "value": title } ] if description: patch_document.append({ "op": "add", "path": "/fields/System.Description", "value": description }) if priority: patch_document.append({ "op": "add", "path": "/fields/Microsoft.VSTS.Common.Priority", "value": priority }) if assigned_to: patch_document.append({ "op": "add", "path": "/fields/System.AssignedTo", "value": assigned_to }) if tags: patch_document.append({ "op": "add", "path": "/fields/System.Tags", "value": tags }) if iteration: patch_document.append({ "op": "add", "path": "/fields/System.IterationPath", "value": iteration if iteration.startswith(self.project) else f"{self.project}\\{iteration}" }) if area: patch_document.append({ "op": "add", "path": "/fields/System.AreaPath", "value": area if area.startswith(self.project) else f"{self.project}\\{area}" }) try: response = requests.post( url, headers={**self.auth_header, "Content-Type": "application/json-patch+json"}, json=patch_document ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: st.error(f"Error creating work item: {str(e)}") if hasattr(e, 'response') and e.response is not None: st.error(f"Response: {e.response.text[:500]}") return None def bulk_create_work_items(self, df: pd.DataFrame) -> tuple[int, List[str]]: """Create multiple work items from a DataFrame.""" created_count = 0 errors = [] required_columns = ['Title'] for col in required_columns: if col not in df.columns: return 0, [f"Required column '{col}' not found in CSV"] for index, row in df.iterrows(): try: title = str(row.get('Title', '')).strip() if not title: errors.append(f"Row {index + 1}: Title is empty") continue description = str(row.get('Description', '')).strip() work_item_type = str(row.get('WorkItemType', 'Task')).strip() priority = int(row.get('Priority', 2)) if pd.notna(row.get('Priority')) else 2 assigned_to = str(row.get('AssignedTo', '')).strip() tags = str(row.get('Tags', '')).strip() iteration = str(row.get('Iteration', '')).strip() area = str(row.get('Area', '')).strip() result = self.create_work_item( title=title, description=description, work_item_type=work_item_type, priority=priority, assigned_to=assigned_to, tags=tags, iteration=iteration, area=area ) if result: created_count += 1 else: errors.append(f"Row {index + 1}: Failed to create '{title}'") except Exception as e: errors.append(f"Row {index + 1}: {str(e)}") return created_count, errors def get_test_plans(self) -> List[Dict]: """Fetch test plans from the project.""" if not self.project: st.error("No project selected!") return [] url = f"{self.base_url}/{self.project}/_apis/testplan/plans?api-version=7.0" try: response = requests.get(url, headers=self.auth_header) response.raise_for_status() return response.json().get("value", []) except requests.exceptions.RequestException as e: st.error(f"Error fetching test plans: {str(e)}") return [] def set_project(self, project: str): """Set the current project.""" self.project = project def get_iterations(self) -> List[Dict]: """Fetch all iterations (sprints) for the project.""" if not self.project: return [] url = f"{self.base_url}/{self.project}/_apis/work/teamsettings/iterations?api-version=7.0" try: response = requests.get(url, headers=self.auth_header) response.raise_for_status() return response.json().get("value", []) except requests.exceptions.RequestException as e: st.error(f"Error fetching iterations: {str(e)}") return [] def get_areas(self) -> List[Dict]: """Fetch all areas for the project.""" if not self.project: return [] url = f"{self.base_url}/{self.project}/_apis/work/teamsettings/areas?api-version=7.0" try: response = requests.get(url, headers=self.auth_header) response.raise_for_status() return response.json().get("value", []) except requests.exceptions.RequestException as e: st.error(f"Error fetching areas: {str(e)}") return [] def initialize_session_state(): """Initialize Streamlit session state variables.""" if "client" not in st.session_state: st.session_state.client = None if "projects" not in st.session_state: st.session_state.projects = [] if "selected_project" not in st.session_state: st.session_state.selected_project = None if "work_items" not in st.session_state: st.session_state.work_items = [] if "test_plans" not in st.session_state: st.session_state.test_plans = [] if "comment_work_item_id" not in st.session_state: st.session_state.comment_work_item_id = None if "success_message" not in st.session_state: st.session_state.success_message = None if "connection_step" not in st.session_state: st.session_state.connection_step = "connect" # connect, select_project, connected if "work_item_types" not in st.session_state: st.session_state.work_item_types = [] if "debug_mode" not in st.session_state: st.session_state.debug_mode = False if "iterations" not in st.session_state: st.session_state.iterations = [] if "selected_iteration" not in st.session_state: st.session_state.selected_iteration = "All Iterations" if "areas" not in st.session_state: st.session_state.areas = [] if "selected_area" not in st.session_state: st.session_state.selected_area = "All Areas" def reset_connection(): """Reset connection state.""" st.session_state.client = None st.session_state.projects = [] st.session_state.selected_project = None st.session_state.work_items = [] st.session_state.test_plans = [] st.session_state.work_item_types = [] st.session_state.iterations = [] st.session_state.selected_iteration = "All Iterations" st.session_state.areas = [] st.session_state.selected_area = "All Areas" st.session_state.connection_step = "connect" st.session_state.success_message = "Disconnected successfully" st.rerun() def render_sidebar(): """Render the sidebar with connection settings.""" with st.sidebar: st.header("๐ง Connection Settings") # Step 1: Connect to Organization if st.session_state.connection_step == "connect": st.subheader("Step 1: Connect to Organization") # Organization input org = st.text_input( "Organization", placeholder="your-organization", help="Your Azure DevOps organization name (from dev.azure.com/{organization})" ) # PAT input pat = st.text_input( "Personal Access Token (PAT)", type="password", placeholder="Enter your PAT", help="Create a PAT at: https://dev.azure.com/{org}/_usersSettings/tokens" ) # Connect button if st.button("๐ Connect to Organization", use_container_width=True, type="primary"): if org and pat: with st.spinner("Connecting..."): client = AzureDevOpsClient(org, pat) if client.test_connection(): st.session_state.client = client # Fetch projects st.session_state.projects = client.get_projects() if st.session_state.projects: st.session_state.connection_step = "select_project" st.session_state.success_message = f"โ Connected! Found {len(st.session_state.projects)} projects" else: st.session_state.success_message = "โ Connected! No projects found" st.rerun() else: st.error("โ Connection failed. Check your organization and PAT.") else: st.error("โ ๏ธ Please fill in all fields") # Step 2: Select Project elif st.session_state.connection_step == "select_project": st.subheader("Step 2: Select Project") if st.session_state.projects: project_names = [p.get("name", "") for p in st.session_state.projects] selected = st.selectbox( "Choose a Project", options=project_names, index=0 if project_names else None ) col1, col2 = st.columns(2) with col1: if st.button("โ Select Project", use_container_width=True, type="primary"): if selected: st.session_state.selected_project = selected st.session_state.client.set_project(selected) st.session_state.connection_step = "connected" # Fetch available work item types, iterations, and areas with st.spinner("Loading work item types, iterations, and areas..."): st.session_state.work_item_types = st.session_state.client.get_work_item_types() st.session_state.iterations = st.session_state.client.get_iterations() st.session_state.areas = st.session_state.client.get_areas() st.session_state.success_message = f"โ Project '{selected}' selected!" st.rerun() with col2: if st.button("๐ Back", use_container_width=True): st.session_state.connection_step = "connect" st.session_state.projects = [] st.rerun() else: st.warning("No projects found") if st.button("๐ Back", use_container_width=True): st.session_state.connection_step = "connect" st.rerun() # Connected State elif st.session_state.connection_step == "connected" and st.session_state.client: st.markdown(f"""