Added examples (Wiql requests for WorkItems and Tests elements)

README.md

@@ -23,12 +23,12 @@ Now you can run `runner.py` with no arguments to see available options.
 1. Get a [personal access token](https://docs.microsoft.com/azure/devops/organizations/accounts/use-personal-access-tokens-to-authenticate?view=vsts).
 
 2. Store the PAT and organization URL you'll be running samples against (note: some samples are destructive, so use a test organization):
-   * `runner.py config url --set-to https://fabrikam.visualstudio.com`
-   * `runner.py config pat --set-to ABC123`
+   * `python runner.py config url --set-to https://fabrikam.visualstudio.com`
+   * `python runner.py config pat --set-to ABC123`
    * If you don't want your PAT persisted to a file, you can put it in an environment variable called `AZURE_DEVOPS_PAT` instead
 
-3. Run `runner.py run {area} {resource}` with the 2 required arguments:
-   * `{area}`: API area (currently core, git, and work_item_tracking) to run the client samples for. Use `all` to include all areas.
+3. Run `python runner.py run {area} {resource}` with the 2 required arguments:
+   * `{area}`: API area (currently `core`, `git`, and `work_item_tracking`) to run the client samples for. Use `all` to include all areas.
    * `{resource}`: API resource to run the client samples for. Use `all` to include all resources.
    * You can optionally pass `--url {url}` to override your configured URL
 

src/samples/test.py

@@ -0,0 +1,77 @@
+"""
+TEST samples
+"""
+import datetime
+import logging
+
+from samples import resource
+from utils import emit
+
+
+logger = logging.getLogger(__name__)
+
+
+def get_project_names(context):
+    core_client = context.connection.get_client("vsts.core.v4_1.core_client.CoreClient")
+    return (project.name for project in core_client.get_projects())
+
+
+@resource("test_plans")
+def get_plans(context):
+    test_client = context.connection.get_client("vsts.test.v4_1.test_client.TestClient")
+    for project in get_project_names(context):
+        try:
+            for plan in test_client.get_plans(project):
+                emit("Test Plan {}: {} ({})".format(plan.id, plan.name, plan.area.name))
+        except Exception as e:
+            emit("Project '{}' raised error: {}".format(project, e))
+
+
+@resource("test_suites")
+def get_test_suites_for_plan(context):
+    test_client = context.connection.get_client("vsts.test.v4_1.test_client.TestClient")
+    for project in get_project_names(context):
+        try:
+            for plan in test_client.get_plans(project):
+                for suite in test_client.get_test_suites_for_plan(project, plan.id):
+                    emit(
+                        "Test Suite {}: {} ({}.{})".format(
+                            suite.id, suite.name, plan.id, plan.name
+                        )
+                    )
+        except Exception as e:
+            emit("Project '{}' raised error: {}".format(project, e))
+
+
+@resource("test_runs")
+def get_test_runs(context):
+    test_client = context.connection.get_client("vsts.test.v4_1.test_client.TestClient")
+    for project in get_project_names(context):
+        try:
+            for run in test_client.get_test_runs(project, top=16):
+                emit(
+                    "Test Run {}: {} => {} ({})".format(
+                        run.id, run.name, run.state, project
+                    )
+                )
+        except Exception as e:
+            emit("Project '{}' raised error: {}".format(project, e))
+
+
+@resource("test_results")
+def get_test_results(context):
+    test_client = context.connection.get_client("vsts.test.v4_1.test_client.TestClient")
+    for project in get_project_names(context):
+        try:
+            for run in test_client.get_test_runs(project, top=10):
+                # Limiting Test Results is not something one shall do!
+                for res in test_client.get_test_results(project, run.id, top=3):
+                    tc = res.test_case
+                    tester = res.run_by.display_name
+                    emit(
+                        "Test Result {}: {} => {} by {} ({})".format(
+                            run.id, tc.name, res.outcome, tester, project
+                        )
+                    )
+        except Exception as e:
+            emit("Project '{}' raised error: {}".format(project, e))

src/samples/work_item_tracking.py

@@ -7,44 +7,89 @@
 from samples import resource
 from utils import emit
 
+from vsts.work_item_tracking.v4_1.models.wiql import Wiql
 
 logger = logging.getLogger(__name__)
 
 
-@resource('work_items')
+def print_work_item(work_item):
+    emit(
+        "{0} {1}: {2}".format(
+            work_item.fields["System.WorkItemType"],
+            work_item.id,
+            work_item.fields["System.Title"],
+        )
+    )
+
+
+@resource("work_items")
 def get_work_items(context):
     wit_client = context.connection.get_client(
-        "vsts.work_item_tracking.v4_1.work_item_tracking_client.WorkItemTrackingClient")
+        "vsts.work_item_tracking.v4_1.work_item_tracking_client.WorkItemTrackingClient"
+    )
 
     desired_ids = range(1, 51)
     work_items = wit_client.get_work_items(ids=desired_ids, error_policy="omit")
 
     for id_, work_item in zip(desired_ids, work_items):
         if work_item:
-            emit("{0} {1}: {2}".format(work_item.fields['System.WorkItemType'],
-                                       work_item.id,
-                                       work_item.fields['System.Title']))
+            print_work_item(work_item)
         else:
             emit("(work item {0} omitted by server)".format(id_))
 
     return work_items
 
 
-@resource('work_items')
+@resource("work_items")
 def get_work_items_as_of(context):
     wit_client = context.connection.get_client(
-        "vsts.work_item_tracking.v4_1.work_item_tracking_client.WorkItemTrackingClient")
+        "vsts.work_item_tracking.v4_1.work_item_tracking_client.WorkItemTrackingClient"
+    )
 
     desired_ids = range(1, 51)
     as_of_date = datetime.datetime.now() + datetime.timedelta(days=-7)
-    work_items = wit_client.get_work_items(ids=desired_ids, as_of=as_of_date, error_policy="omit")
+    work_items = wit_client.get_work_items(
+        ids=desired_ids, as_of=as_of_date, error_policy="omit"
+    )
 
     for id_, work_item in zip(desired_ids, work_items):
         if work_item:
-            emit("{0} {1}: {2}".format(work_item.fields['System.WorkItemType'],
-                                       work_item.id,
-                                       work_item.fields['System.Title']))
+            print_work_item(work_item)
         else:
             emit("(work item {0} omitted by server)".format(id_))
 
     return work_items
+
+
+@resource("wiql_query")
+def wiql_query(context):
+    wit_client = context.connection.get_client(
+        "vsts.work_item_tracking.v4_1.work_item_tracking_client.WorkItemTrackingClient"
+    )
+    wiql = Wiql(
+        query="""
+        select [System.Id],
+            [System.WorkItemType],
+            [System.Title],
+            [System.State],
+            [System.AreaPath],
+            [System.IterationPath],
+            [System.Tags]
+        from WorkItems
+        where [System.WorkItemType] = 'Test Case'
+        order by [System.ChangedDate] desc"""
+    )
+    # We limit number of results to 30 on purpose
+    wiql_results = wit_client.query_by_wiql(wiql, top=30).work_items
+    emit("Results: {0}".format(len(wiql_results)))
+    if wiql_results:
+        # WIQL query gives a WorkItemReference with ID only
+        # => we get the corresponding WorkItem from id
+        work_items = (
+            wit_client.get_work_item(int(res.id)) for res in wiql_results
+        )
+        for work_item in work_items:
+            print_work_item(work_item)
+        return work_items
+    else:
+        return []