test_bot_proposed_constraints.py

#!/usr/bin/python
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import json
import requests
from requests import auth

base_url = 'https://review.openstack.org/'
digest_auth = auth.HTTPDigestAuth('dims-v', 'xxxxxx')
url = '%sa/changes/' % base_url


def get_latest_bot_proposed_review():
    project = "openstack/requirements"
    branch = 'master'
    query = ('project:%s branch:%s status:open '
             'topic:openstack/requirements/constraints' % (project, branch))
    params = {'q': query, 'o': 'CURRENT_REVISION'}
    r = requests.get(url, params=params, auth=digest_auth)
    if r.status_code != 200:
        return None
    res_text = r.text
    (dummy_magic_prefix, dummy_nl, res_json) = res_text.partition('\n')
    res_json = json.loads(res_json)
    if res_json and len(res_json) > 0:
        return res_json[0]['change_id']
    return None


def get_constraints_test_reviews():
    query = ('branch:master status:open '
             'topic:dims/test/constraints')
    params = {'q': query, 'o': 'CURRENT_REVISION'}
    r = requests.get(url, params=params, auth=digest_auth)
    if r.status_code == 200:
        res_text = r.text
        (dummy_magic_prefix, dummy_nl, res_json) = res_text.partition('\n')
        res_json = json.loads(res_json)
        if res_json:
            for x in range(0, len(res_json)):
                yield res_json[x]['change_id']


def rebase_test_review(change_id):
    url = ('%sa/changes/%s/rebase' %
           (base_url, change_id))
    headers = {'Content-Type': 'application/json'}
    payload = {}
    r = requests.post(url, auth=digest_auth, headers=headers, json=payload)
    if r.status_code == 200:
        print('Review was rebased : %s' % change_id)
        return True
    print('Review was NOT rebased : %s' % change_id)
    return False


def update_test_review(change_id, bot_change_id):
    url = ('%sa/changes/%s/edit:message' %
           (base_url, change_id))
    headers = {'Content-Type': 'application/json'}
    payload = {
        "message": """[WIP] Testing latest u-c

Depends-On: %s
Change-Id: %s
""" % (bot_change_id, change_id)
    }
    r = requests.put(url, auth=digest_auth, headers=headers, json=payload)
    if r.status_code == 204:
        print('Review commit message was updated : %s' % change_id)
        url = ('%sa/changes/%s/edit:publish' %
               (base_url, change_id))
        payload = {}
        r = requests.post(url, auth=digest_auth, headers=headers, json=payload)
        if r.status_code == 204:
            print('Review commit message was published : %s' % change_id)
            return True
        print('Review commit message was NOT published : %s' % change_id)
    else:
        print('Review commit message was NOT updated : %s' % change_id)
    return False


def add_recheck_comment_on_review(change_id):
    url = ('%sa/changes/%s/revisions/current/review' %
           (base_url, change_id))
    headers = {'Content-Type': 'application/json'}
    payload = {
        'message': 'recheck',
        'labels': {
            'Code-Review': 0,
        },
    }
    r = requests.post(url, auth=digest_auth, headers=headers, json=payload)
    if r.status_code == 200:
        print('Review rechecked : %s' % change_id)
        return True
    print('Review recheck failed : %s' % change_id)
    return False


def main():
    bot_id = get_latest_bot_proposed_review()
    print('Proposal Bot review: %s' % bot_id)
    if bot_id:
        for id in get_constraints_test_reviews():
            print('Test review : %s' % id)
            updated = update_test_review(id, bot_id)
            rebased = rebase_test_review(id)
            if not updated and not rebased:
                add_recheck_comment_on_review(id)


if __name__ == "__main__":
    main()