File: relations.py

package info (click to toggle)
azure-devops-cli-extension 1.0.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 20,384 kB
  • sloc: python: 160,782; xml: 198; makefile: 56; sh: 51
file content (151 lines) | stat: -rw-r--r-- 6,379 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

from knack.log import get_logger
from knack.util import CLIError
from azext_devops.devops_sdk.v5_0.work_item_tracking.models import JsonPatchOperation, Wiql

from azext_devops.dev.common.services import (get_work_item_tracking_client,
                                              resolve_instance)

logger = get_logger(__name__)


def get_relation_types_show(organization=None, detect=None):
    """ List work item relations supported in the organization.
    """
    organization = resolve_instance(detect=detect, organization=organization)
    client = get_work_item_tracking_client(organization)
    return client.get_relation_types()


def add_relation(id, relation_type, target_id=None, target_url=None, organization=None, detect=None):  # pylint: disable=redefined-builtin
    """ Add relation(s) to work item.
    """

    if target_id is None and target_url is None:
        raise CLIError('--target-id or --target-url must be provided')

    organization = resolve_instance(detect=detect, organization=organization)
    patch_document = []
    client = get_work_item_tracking_client(organization)

    relation_types_from_service = client.get_relation_types()
    relation_type_system_name = get_system_relation_name(relation_types_from_service, relation_type)

    patch_document = []
    if target_id is not None:
        target_work_item_ids = target_id.split(',')
        work_item_query_clause = []
        for target_work_item_id in target_work_item_ids:
            work_item_query_clause.append('[System.Id] = {}'.format(target_work_item_id))

        wiql_query_format = 'SELECT [System.Id] FROM WorkItems WHERE ({})'
        wiql_query_to_get_target_work_items = wiql_query_format.format(' OR '.join(work_item_query_clause))

        wiql_object = Wiql()
        wiql_object.query = wiql_query_to_get_target_work_items
        target_work_items = client.query_by_wiql(wiql=wiql_object).work_items

        if len(target_work_items) != len(target_work_item_ids):
            raise CLIError('Id(s) supplied in --target-id is not valid')

        for target_work_item in target_work_items:
            op = _create_patch_operation('add', '/relations/-', relation_type_system_name, target_work_item.url)
            patch_document.append(op)

    if target_url is not None:
        target_urls = target_url.split(',')

        for url in target_urls:
            op = _create_patch_operation('add', '/relations/-', relation_type_system_name, url)
            patch_document.append(op)

    client.update_work_item(document=patch_document, id=id)
    work_item = client.get_work_item(id, expand='All')
    work_item = fill_friendly_name_for_relations_in_work_item(relation_types_from_service, work_item)

    return work_item


def remove_relation(id, relation_type, target_id, organization=None, detect=None):  # pylint: disable=redefined-builtin
    """ Remove relation(s) from work item.
    """
    organization = resolve_instance(detect=detect, organization=organization)
    patch_document = []
    client = get_work_item_tracking_client(organization)

    relation_types_from_service = client.get_relation_types()
    relation_type_system_name = get_system_relation_name(relation_types_from_service, relation_type)
    target_work_item_ids = target_id.split(',')

    main_work_item = client.get_work_item(id, expand='All')

    if main_work_item.relations:
        for target_work_item_id in target_work_item_ids:
            target_work_item = client.get_work_item(target_work_item_id, expand='All')
            target_work_item_url = target_work_item.url

            index = 0
            for relation in main_work_item.relations:
                if relation.rel == relation_type_system_name and relation.url == target_work_item_url:
                    po = _create_patch_operation('remove', '/relations/{}'.format(index))
                    patch_document.append(po)
                    break
                index = index + 1

    if len(patch_document) != len(target_work_item_ids):
        raise CLIError('Id(s) supplied in --target-id is not valid')

    client.update_work_item(document=patch_document, id=id)
    work_item = client.get_work_item(id, expand='All')
    work_item = fill_friendly_name_for_relations_in_work_item(relation_types_from_service, work_item)

    return work_item


def show_work_item(id, organization=None, detect=None):  # pylint: disable=redefined-builtin
    """ Get work item, fill relations with friendly name
    """
    organization = resolve_instance(detect=detect, organization=organization)
    client = get_work_item_tracking_client(organization)

    work_item = client.get_work_item(id, expand='All')
    relation_types_from_service = client.get_relation_types()
    work_item = fill_friendly_name_for_relations_in_work_item(relation_types_from_service, work_item)
    return work_item


def fill_friendly_name_for_relations_in_work_item(relation_types_from_service, wi):
    if not wi.relations:
        return wi

    for relation in wi.relations:
        for relation_type_from_service in relation_types_from_service:
            if relation_type_from_service.reference_name == relation.rel:
                relation.rel = relation_type_from_service.name
    return wi


def get_system_relation_name(relation_types_from_service, relation_type):
    for relation_type_from_service in relation_types_from_service:
        if relation_type_from_service.name.lower() == relation_type.lower():
            return relation_type_from_service.reference_name

    raise CLIError("--relation-type is not valid. Use \"az boards work-item relation list-type\" " +
                   "command to list possible relation types in your project")


def _create_patch_operation(op, path, rel=None, url=None):
    patch_operation = JsonPatchOperation()
    patch_operation.op = op
    patch_operation.path = path
    if rel is not None and url is not None:
        patch_operation.value = {
            'rel': rel,
            'url': url
        }

    return patch_operation