Documentation for workitems

[tillias]

Hi guys, thank you very much for the library. Example provided works like a charm, unfortunately I can't find any possibility doing basic stuff:

• Create area within a project
• Create workitem within an area
• Attach file to workitem

Can you please advise? Autocomplete doesn't work with this library as well.

[jfthuong]

Hi,

Please have a look at following repo for a start:
https://github.com/Microsoft/azure-devops-python-samples

There is not much there but I will try to contribute again and add few examples as soon as I got some time.

There is also some pointers in the "issue" #132

Finally, what do you mean by autocomplete does not work ?

[tillias]

Thank you very much for the response. I have already checked samples and missing following examples:

• Create area within a project
• Create workitem within an area
• Attach file to workitem

I've tried Visual Studio 2017 and when I declare any variable then I can't see any methods of it -> autocomplete is completely missing.

[jfthuong]

Hi,

I am not sure for the first bullet but I suppose it is like creating the creation of a Workitem.

For creating a WorkItem within an area, you just create it and assign the correct area.

To give you some ideas, this is part of my wrapper I use to add some functionalities and clients. By using type hints on the clients, I managed to get code completion for the functions.

from collections import defaultdict
from dataclasses import replace
from datetime import datetime, timedelta, timezone
import logging
import mmap
from msrest.authentication import BasicAuthentication
from msrest.exceptions import ClientRequestError
import os
from pprint import pprint
import re
import sys
from typing import Any, List, Dict, NamedTuple, Optional, Union, Tuple, Generator
from warnings import warn

from vsts.vss_connection import VssConnection

from vsts.work_item_tracking.v4_1.models.work_item import WorkItem
from vsts.work_item_tracking.v4_1.models.wiql import Wiql
from vsts.work_item_tracking.v4_1.models.json_patch_operation import JsonPatchOperation
from vsts.work_item_tracking.v4_1.models.attachment_reference import AttachmentReference
from vsts.work_item_tracking.v4_1.models.work_item_classification_node import (
    WorkItemClassificationNode as Iteration,
)
from vsts.work_item_tracking.v4_1.models.work_item_relation import WorkItemRelation

from vsts.test.v4_1.models.identity_ref import IdentityRef
from vsts.test.v4_1.models.run_create_model import RunCreateModel
from vsts.test.v4_1.models.run_update_model import RunUpdateModel
from vsts.test.v4_1.models.suite_update_model import SuiteUpdateModel
from vsts.test.v4_1.models.test_case_result import TestCaseResult
from vsts.test.v4_1.models.test_point import TestPoint


from vsts.build.v4_0.build_client import BuildClient
from vsts.core.v4_1.core_client import CoreClient
from vsts.test.v4_1.test_client import TestClient
from vsts.work_item_tracking.v4_1.work_item_tracking_client import (
    WorkItemTrackingClient,
)
from vsts.work.v4_1.work_client import WorkClient



GenId = Generator[int, None, None]
GenWit = Generator[WorkItem, None, None]
GenTestCases = Generator[Union[int, WorkItem], None, None]
Wit_Data = List[JsonPatchOperation]

# cspell:words astimezone


def utc_to_local(utc_dt):
    """Transform date to Local Timezone"""
    return utc_dt.replace(tzinfo=timezone.utc).astimezone(tz=None)


def get_date(date: Optional[str] = None, as_utc=False):
    """Get date object based on date in format YYYY/MM/DD

    Args:
        date (optional): if omitted, return today's date
    """
    if date is None:
        date_time = datetime.utcnow()
    else:
        if not re.match(r"\d{4}/\d{2}/\d{2}$", date):
            raise ValueError(f"Incorrect date {date!r}")
        date_time = datetime.strptime(date, r"%Y/%m/%d")
    return date_time if not as_utc else utc_to_local(date_time)


class Client:
    """VSTS Clients to use REST API"""

    VERSION = "v4_1"

    def __init__(
        self,
        team_instance: str,
        token: str,
        project: str,
        logger=None,
    ):
        """Connect to TFS to create clients"""
        self.logger = logger or logging.getLogger(__name__)
        self.offline = offline
        # We will connect and add one client for each required type
        self.project = project
        credentials = BasicAuthentication("", token)
        self.team = team_instance
        self.connection = VssConnection(base_url=self.team, creds=credentials)

        try:
            self.wit: WorkItemTrackingClient = self._connect("work_item_tracking")
            self.test: TestClient = self._connect("test")
            self.core: CoreClient = self._connect("core")  # for Projects
            self.build: BuildClient = self._connect("build")  # for Auto-Tests builds
        except ClientRequestError as e:
            self.logger.critical(f"E004: Cannot connect to '{team_instance}': {e}")
            exit(4)

        # We add the Test Configurations
        if self.offline:
            self.test_config = dict()  # type: Dict[str, int]
            self.active_config = list()  # type: List[str]
        else:
            self.test_config = {
                c.name: c.id for c in self.test.get_test_configurations(project)
            }
            self.active_config = [
                c.name
                for c in self.test.get_test_configurations(project)
                if c.state == "active"
            ]


    def _connect(self, subclient: str) -> Any:
        """Connect to the client by using the magic string"""
        if self.offline:
            return None
        cl_class = "".join([word.title() for word in subclient.split("_")])
        magic = f"vsts.{subclient}.{self.VERSION}.{subclient}_client.{cl_class}Client"
        return self.connection.get_client(magic)

    @property
    def configurations(self) -> Dict[str, Any]:
        """TFS Test Configurations"""
        if self.offline:
            return {}
        return {c.name: c.id for c in self.test.get_test_configurations(self.project)}


    def get_identity(self, username: Optional[str]) -> IdentityRef:
        """Return Identity from username (e.g. jthuong) or display name"""
        if username:
            unique_name = f"ANSYS\\{username}"
            if username in self.identities:
                return self.identities[username]
            elif unique_name not in self.identities:
                warn(f"WARNING: identify for {username!r} not found")
                return None
            else:
                return self.identities[unique_name]
        else:
            return None

    def get_test_config(self, os: Optional[str]) -> Optional[int]:
        """Return identity of Test Configuration from its name"""
        return self.test_config.get(os) if os else None

    def get_id_from_wiql(self, query: str) -> GenId:
        """Return IDs of WorkItems from WIQL query"""
        self.logger.debug(f"I022: Performing WIQL query in {self.team}:\n{query}")
        query_wiql = Wiql(query=query)
        results = self.wit.query_by_wiql(query_wiql).work_items
        return (int(res.id) for res in results)

    def get_wit_from_wiql(self, query: str, expand=None) -> GenWit:
        """Return WorkItems from WIQL query"""
        # WIQL query gives a WorkItemReference
        # => we get the corresponding WorkItem from id
        return (
            self.wit.get_work_item(wid, expand=expand)
            for wid in self.get_id_from_wiql(query)
        )

    def get_id_from_query_id(self, query_id: str) -> GenId:
        """Return IDs of WorkItems from query identifier"""
        self.logger.debug(f"I022: Performing Query {query_id} in {self.team}")
        results = self.wit.query_by_id(query_id).work_items
        return (int(res.id) for res in results)

    def get_wit_from_query_id(self, query_id: str, expand=None) -> GenWit:
        """Return WorkItems from WIQL query"""
        return (
            self.wit.get_work_item(i, expand=expand)
            for i in self.get_id_from_query_id(query_id)
        )

    def get_tests_from_areas(self, areas: List[str], as_wit=False) -> GenTestCases:
        """Retrieve all the test cases from a list of Areas

        Args:
            areas: list of Area Paths
            as_wit (optional): return WorkItem if True, and WorkItem ID (int) if False

        Raises:
            AttributeError: if a empty list or another type is passed as argument
        """
        # TODO: consider adding "fields" parameter to filter
        if not areas or not isinstance(areas, list):
            raise AttributeError("List of areas shall contain at least one element")

        query_area = " OR ".join(
            [f"[System.AreaPath] UNDER '{area}'" for area in areas]
        )
        # NOTE: might consider " IN ..."
        query = f"""
        SELECT  [System.Id]
        FROM workitems
        WHERE [System.WorkItemType] =  'Test Case'
        AND ({query_area})
        ORDER BY [System.Id]
        """
        if as_wit:
            return self.get_wit_from_wiql(query)
        else:
            return self.get_id_from_wiql(query)



    @classmethod
    def _get_json(
        cls,
        title: str = None,
        description: str = None,
        state: str = None,
        area: str = None,
        iteration: str = None,
        fields: List[str] = None,
    ) -> Wit_Data:
        """Return a list of JsonPatch for creation/update of WorkItems"""

        def _create_json(field, value):
            return JsonPatchOperation(op="add", path=f"/fields/{field}", value=value)

        patch_document = list()  # type: Wit_Data

        fields_values = [
            (title, "System.Title"),
            (description, "System.Description"),
            (state, "System.State"),
            (area, "System.AreaPath"),
            (iteration, "System.IterationPath"),
        ]
        for value, field in fields_values:
            if value is not None:
                patch_document.append(_create_json(field, value))

        if fields is not None and fields:
            for field in fields:
                kvp = field.split("=", 1)
                if len(kvp) == 2:
                    patch_document.append(_create_json(kvp[0], kvp[1]))
                else:
                    raise ValueError(
                        'The --fields argument should consist of "field=value" strings.'
                    )
        return patch_document

    def push_work_item(
        self,
        wit_type: str,  # For "Update", similar code with wit_id instead of wit_type
        title: str = None,
        description: str = None,
        state: str = None,
        area: str = None,
        iteration: str = None,
        fields: List[str] = None,
        wit_id: int = None,
        validate_only: bool = None,
    ):
        """Create a work item.

        Args:
            wit_type: Name of the work item type (e.g. "Bug" or "Test Case").
            title: Title of the work item.
            description: Description of the work item.
            state: State of the work item (e.g. active)
            area: Area the work item is assigned to (e.g. Demos)
            iteration: Iteration path of the work item (e.g. Demos\\Iteration 1)
            fields: List of "field=value" pairs for custom fields you would like to set
            wit_id: If set (integer), the corresponding WorkItem will be updated
            validate_only: If you only want to validate the changes without saving them

        Return:
            WorkItem object created or updated [in case "update" is set]
        """
        update = not wit_id is None
        try:
            # For Creation of WorkItem, we need a Title
            if not update and title is None:
                raise ValueError("--title is a required argument.")
            patch_document = self._get_json(
                title, description, state, area, iteration, fields
            )

            # SIMULATION MODE
            if self.simu:
                if update:
                    # TODO: maybe convert patch_document in a more printable format
                    msg = f"SIMU: Updating {wit_type} {wit_id} with {patch_document}"
                    self.logger.info(msg)
                    return self.wit.get_work_item(wit_id, self.project)
                else:
                    self.logger.info(
                        f"SIMU: Creating new {wit_type} with title {title}"
                    )
                    return None

            if update:
                work_item = self.wit.update_work_item(
                    patch_document, wit_id, self.project, validate_only=validate_only
                )
            else:
                work_item = self.wit.create_work_item(
                    patch_document, self.project, wit_type, validate_only=validate_only
                )
            return work_item

        except Exception as e:
            operation = f"update of {wit_id}" if update else f"creation of {wit_type}"
            raise Exception(f"Error during {operation}: {e}")

    def get_attachments(self, tcid: int) -> List[AttachmentReference]:
        """Get the list of attachments of a given Test Case"""
        tc = self.wit.get_work_item(int(tcid), expand="Relations")
        return [rel for rel in tc.relations if rel.rel == "AttachedFile"]

    def link_attachment_to_step(
        self, tcid: int, attach: AttachmentReference, name: str, step: int, comment=""
    ) -> None:
        """Add an attachment to a given step of a Test Case
        Args:
            tcid: identifier of the test case
            idx: index of the attachment (sorting order)
            name: name of the attached file
            step: index of the step
        """
        tc = self.wit.get_work_item(int(tcid), expand="Relations")
        comment = f"[TestStep={step}]:{comment}"
        try:
            idx = tc.relations.index(attach)
            # TODO: consider attachment ID instead of attachment object (based on need)
            # def get_relation_idx(tc, attachment_id):
            #     """Get relation index of attachment to be able to update attribute / comment"""
            #     for idx, rel in enumerate(tc.relations):
            #         if rel.rel == "AttachedFile" and attachment_id in rel.url:
            #             return idx
            #     return None

        except ValueError:
            self.logger.error(f"ERROR: attachment {attach} not found in Test {tcid}")
            self.logger.debug("Details below...", exc_info=True)

        def replace(attribute: str, value: str):
            """Try to replace the value, and add it if it does not exist"""
            try:
                patch_document = [
                    JsonPatchOperation(
                        op="replace",
                        path=f"/relations/{idx}/attributes/{attribute}",
                        value=value,
                    )
                ]
                self.wit.update_work_item(patch_document, tcid, self.project)
            except:
                patch_document = [
                    JsonPatchOperation(
                        op="add",
                        path=f"/relations/{idx}/attributes/{attribute}",
                        value=value,
                    )
                ]
                self.wit.update_work_item(patch_document, tcid, self.project)

        replace("comment", comment)

    def upload_work_item_attachment(
        self, wit_id: int, file_path: str, comment="", callback=None
    ):
        """Upload a file and link it to a WorkItem"""
        file_name = os.path.basename(file_path)
        with open(file_path, "r+b") as file:
            # We use mmap, so we don't load entire file in memory at the same time,
            # and so we can start streaming before we are done reading the file.
            mm = mmap.mmap(file.fileno(), 0)
            attachment = self.wit.create_attachment(
                mm, file_name=file_name, callback=callback
            )

        # Link Work Item to attachment
        patch_document = [
            JsonPatchOperation(
                op="add",
                path="/relations/-",
                value={
                    "rel": "AttachedFile",
                    "url": attachment.url,
                    "attributes": {"comment": comment},
                },
            )
        ]
        return self.wit.update_work_item(patch_document, wit_id)

    def upload_attachment_to_step(
        self, tc_id: int, file_path: str, step: int, comment="", callback=None
    ):
        """Add an attachment to a given step of a Test Case"""
        comment = f"[TestStep={step}]:{comment}"
        return self.upload_work_item_attachment(tc_id, file_path, comment, callback)

    def download_attachments(self, tc_id: int, outdir: str):
        """"Download all attachments of a WorkItem into a given directory"""
        # Attachments are stored in "Relations" of the Test Case as "AttachedFile"
        tc = self.wit.get_work_item(tc_id, self.project, expand="Relations")
        attachments = [rel for rel in tc.relations if rel.rel == "AttachedFile"]
        for attach in attachments:
            name = attach.attributes["name"]
            id_ = re.sub(r".*/", "", attach.url)
            # TODO: Understand why we can't have attach.id (not found)
            content = self.wit.get_attachment_content(id_, download=True)

            file_path = os.path.join(outdir, f"export_{id_}_{name}")
            if os.path.exists(file_path):
                os.remove(file_path)
            with open(file_path, "wb") as f:
                for c in content:
                    f.write(c)
                self.logger.debug(f"Successfully downloaded {file_path}")

[jfthuong]

... and in general, by looking at <PYTHON_INSTAL_DIR>\Lib\site-packages\vsts\work_item_tracking_process\v4_1\work_item_tracking_process_client.py, you can find out what you need.

[tedchamb]

Thanks for your responses @jfthuong .

@tillias , this python sdk is a thin wrapper around our REST APIs. You can find documentation and examples for the WIT REST API here. The parameter names and contracts should all match with this parameters and contracts in this sdk.

Create area within a project: Rest Doc, Python method: WorkItemTrackingClient.create_or_update_classification_node

Create workitem within an area: Rest Doc, Python method: WorkItemTrackingClient.create_work_item

Creating a work item attachment: Rest Doc, Python method: WorkItemTrackingClient.create_attachment

Linking Attachment to work item: Rest Doc Python method: WorkItemTrackingClient.update_work_item

Example of creating and attaching a file to a work item:

def upload_work_item_attachment(wit_client, work_item_id, file_name, callback=None):
    # upload file / create attachment
    with open(file_name, 'r+b') as file:
        # use mmap, so we don't load entire file in memory at the same time, and so we can start
        # streaming before we are done reading the file.
        mm = mmap.mmap(file.fileno(), 0)
        attachment = wit_client.create_attachment(mm, file_name=file_name, callback=callback)

    # Link Work Item to attachment
    patch_document = [
        JsonPatchOperation(
            op="add",
            path="/relations/-",
            value={
                "rel": "AttachedFile",
                "url": attachment.url
            }
        )
    ]
    wit_client.update_work_item(patch_document, work_item_id)

@vtbassmatt , just fyi, on suggestions for new samples.

[ErickFu]

woooooo....