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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
|
# --------------------------------------------------------------------------------------------
# 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 azext_devops.dev.common.services import resolve_instance_and_project, get_build_client
from azext_devops.dev.common.git import resolve_git_ref_heads
from azext_devops.dev.common.identities import resolve_identity_as_id
logger = get_logger(__name__)
def pipeline_run_show(id, open=False, organization=None, project=None, detect=None): # pylint: disable=redefined-builtin
""" Show details of a pipeline run.
:param id: ID of the pipeline run.
:type id: int
:param open: Open the build results page in your web browser.
:type open: bool
:param project: Name or ID of the team project.
:type project: str
:param detect: Automatically detect organization and project. Default is "on".
:type detect: str
:rtype: :class:`<Build> <build.v5_1.models.Build>`
"""
organization, project = resolve_instance_and_project(
detect=detect, organization=organization, project=project)
client = get_build_client(organization)
build = client.get_build(build_id=id, project=project)
if open:
_open_pipeline_run(build, organization)
return build
def pipeline_run_list(pipeline_ids=None, branch=None, organization=None, project=None, detect=None, top=None,
query_order=None, result=None, status=None, reason=None, tags=None, requested_for=None):
""" List the pipeline runs in a project.
:param pipeline_ids: IDs (space separated) of definitions to list builds for.
For multiple pipeline ids: --pipeline-ids 1 2
:type pipeline_ids: list of int
:param branch: Filter by builds for this branch.
:type branch: str
:param top: Maximum number of builds to list.
:type top: int
:param query_order: Order of pipeline runs.
:type query_order: str
:param result: Limit to builds with this result.
:type result: str
:param status: Limit to builds with this status.
:type status: str
:param reason: Limit to builds with this reason.
:type reason: str
:param tags: Limit to builds with each of the specified tags. Space separated.
:type tags: list of str
:param requested_for: Limit to builds requested for this user or group.
:type requested_for: str
"""
organization, project = resolve_instance_and_project(
detect=detect, organization=organization, project=project)
client = get_build_client(organization)
if pipeline_ids is not None and pipeline_ids:
pipeline_ids = list(set(pipeline_ids)) # make distinct
if tags is not None and tags:
tags = list(set(tags)) # make distinct
query_order = _resolve_runs_query_order(query_order)
builds = client.get_builds(definitions=pipeline_ids,
project=project,
branch_name=resolve_git_ref_heads(branch),
top=top,
result_filter=result,
status_filter=status,
reason_filter=reason,
tag_filters=tags,
query_order=query_order,
requested_for=resolve_identity_as_id(requested_for, organization))
return builds
def _resolve_runs_query_order(query_order):
if query_order:
query_order_vals = ['finishTimeAscending', 'finishTimeDescending', 'queueTimeAscending',
'queueTimeDescending', 'startTimeAscending', 'startTimeDescending']
for val in query_order_vals:
if query_order.lower() in val.lower():
return val
logger.warning("Cannot resolve --query-order, continuing with None")
return None
def pipeline_run_add_tag(run_id, tags, organization=None, project=None, detect=None):
""" Add tag(s) for a pipeline run.
:param run_id: ID of the pipeline run.
:type run_id: int
:param tags: Tag(s) to be added to the pipeline run. [Comma separated values]
:type tags: str
:rtype: list of str
"""
organization, project = resolve_instance_and_project(detect=detect,
organization=organization,
project=project)
client = get_build_client(organization)
tags = list(map(str, tags.split(',')))
if len(tags) == 1:
tags = client.add_build_tag(
project=project, build_id=run_id, tag=tags[0])
else:
tags = client.add_build_tags(
tags=tags, project=project, build_id=run_id)
return tags
def pipeline_run_delete_tag(run_id, tag, organization=None, project=None, detect=None):
""" Delete a pipeline run tag.
:param run_id: ID of the pipeline run.
:type run_id: int
:param tag: Tag to be deleted from the pipeline run.
:type tag: str
:rtype: list of str
"""
organization, project = resolve_instance_and_project(detect=detect,
organization=organization,
project=project)
client = get_build_client(organization)
tags = client.delete_build_tag(project=project, build_id=run_id, tag=tag)
return tags
def pipeline_run_get_tags(run_id, organization=None, project=None, detect=None):
""" Get tags for a pipeline run.
:param run_id: ID of the pipeline run.
:type run_id: int
:rtype: list of str
"""
organization, project = resolve_instance_and_project(detect=detect,
organization=organization,
project=project)
client = get_build_client(organization)
tags = client.get_build_tags(build_id=run_id, project=project)
return tags
def _open_pipeline_run(run, organization):
"""Open the build results page in your web browser.
:param :class:`<Build> <build.v5_1.models.Build>` build:
:param str organization:
"""
from webbrowser import open_new
from azext_devops.dev.common.uri import uri_quote
# https://dev.azure.com/OrgName/ProjectName/_build/results?buildId=1234
project = run.project.name
url = organization.rstrip('/') + '/' + uri_quote(project) + '/_build/results?buildid='\
+ uri_quote(str(run.id))
logger.debug('Opening web page: %s', url)
open_new(url=url)
def _open_pipeline_run6_0(run, project, organization):
"""Open the build results page in your web browser.
:param :class:`<Run> <azure.devops.v6_0.pipelines.models.Run>`
:param str project:
:param str organization:
"""
from webbrowser import open_new
from azext_devops.dev.common.uri import uri_quote
url = f"{organization.rstrip('/')}/{uri_quote(project)}/_build/results?buildid={uri_quote(str(run.id))}"
logger.debug('Opening web page: %s', url)
open_new(url=url)
|
