consumer, scheduler: support testing dist-git PRs (#104)

This makes it possible to request openQA tests for dist-git PRs.
Anyone (for now) can comment "/openqa test" on a dist-git PR,
and if the consumer is configured to listen on the appropriate
topics, it should schedule update tests (always the full set, for
now) on the scratch build associated with the PR. If the scratch
build is complete when the comment is added, the tests should be
scheduled immediately; if not, the tests will be scheduled when
we get a message indicating a scratch build has completed for the
PR. We also automatically test any subsequent scratch builds for
the same PR when they complete.

As part of this, we move the openqa_hostname and openqa_baseurl
definitions from consumer config to fedora_openqa config. They
are always the same for all consumers in real-world use, so it
doesn't really make sense to require them to be set for every
consumer.

Signed-off-by: Adam Williamson <awilliam@redhat.com>

sample-configs/fedora_openqa_resultsdb_reporter.stg.toml

@@ -56,8 +56,6 @@ routing_keys = [
 ]
 
 [consumer_config]
-openqa_hostname = "openqa.stg.fedoraproject.org"
-openqa_baseurl = "https://openqa.stg.fedoraproject.org"
 resultsdb_url = "http://resultsdb-stg01.qa.fedoraproject.org/resultsdb_api/api/v2.0/"
 do_report = false
 

sample-configs/fedora_openqa_resultsdb_reporter.toml

@@ -55,8 +55,6 @@ routing_keys = [
 ]
 
 [consumer_config]
-openqa_hostname = "openqa.fedoraproject.org"
-openqa_baseurl = "https://openqa.fedoraproject.org"
 resultsdb_url = "http://resultsdb01.qa.fedoraproject.org/resultsdb_api/api/v2.0/"
 do_report = false
 

sample-configs/fedora_openqa_scheduler.stg.toml

@@ -59,8 +59,6 @@ routing_keys = ["org.fedoraproject.prod.pungi.compose.status.change",
                 "org.fedoraproject.prod.odcs.compose.state-changed"]
 
 [consumer_config]
-# host to schedule tests on
-openqa_hostname = "openqa.stg.fedoraproject.org"
 # arches to schedule update tests for
 update_arches = ["x86_64", "aarch64"]
 

sample-configs/fedora_openqa_scheduler.toml

@@ -53,8 +53,6 @@ routing_keys = ["org.fedoraproject.prod.pungi.compose.status.change",
                 "org.fedoraproject.prod.odcs.compose.state-changed"]
 
 [consumer_config]
-# host to schedule tests on
-openqa_hostname = "openqa.fedoraproject.org"
 # arches to schedule update tests for
 update_arches = ["x86_64"]
 

sample-configs/fedora_openqa_wiki_reporter.stg.toml

@@ -48,8 +48,6 @@ exchange = "zmq.topic"
 routing_keys = ["org.fedoraproject.stg.openqa.job.done"]
 
 [consumer_config]
-openqa_hostname = "openqa.stg.fedoraproject.org"
-openqa_baseurl = "https://openqa.stg.fedoraproject.org"
 wiki_hostname = "stg.fedoraproject.org"
 do_report = false
 

sample-configs/fedora_openqa_wiki_reporter.toml

@@ -47,8 +47,6 @@ exchange = "zmq.topic"
 routing_keys = ["org.fedoraproject.prod.openqa.job.done"]
 
 [consumer_config]
-openqa_hostname = "openqa.fedoraproject.org"
-openqa_baseurl = "https://openqa.fedoraproject.org"
 wiki_hostname = "fedoraproject.org"
 do_report = false
 

src/fedora_openqa/config.py

@@ -35,6 +35,7 @@ CONFIG = configparser.ConfigParser()
 CONFIG.add_section('cli')
 CONFIG.add_section('report')
 CONFIG.add_section('schedule')
+CONFIG.add_section('consumer')
 
 CONFIG.set('cli', 'log-file', '')
 CONFIG.set('cli', 'log-level', 'info')
@@ -46,6 +47,11 @@ CONFIG.set('report', 'wiki_hostname', 'stg.fedoraproject.org')
 
 CONFIG.set('schedule', 'arches', 'x86_64')
 
+CONFIG.set('consumer', 'openqa_baseurl', 'https://localhost')
+CONFIG.set('consumer', 'openqa_hostname', 'localhost')
+# dist-git token for setting flags when running tests on PRs
+CONFIG.set('consumer', 'dist-git_token', '')
+
 CONFIG.read('/etc/fedora-openqa/schedule.conf')
 CONFIG.read('{0}/.config/fedora-openqa/schedule.conf'.format(os.path.expanduser('~')))
 

src/fedora_openqa/consumer.py

@@ -24,16 +24,70 @@ openQA jobs."""
 # standard libraries
 import logging
 import time
+from urllib.parse import urlencode
 
 # external imports
 import fedfind.helpers
 import fedora_messaging.config
 import mwclient.errors
 from openqa_client.client import OpenQA_Client
+from openqa_client.const import JOB_FINAL_STATES, JOB_OK_RESULTS, JOB_NOT_COMPLETE_RESULTS
+from requests import Session
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
 
 # internal imports
 from . import schedule
 from . import report
+from .config import CONFIG
+
+# requests session with retries
+SESSION = Session()
+SESSION.mount(
+    "https://",
+    HTTPAdapter(
+        max_retries=Retry(
+            total=5,
+            backoff_factor=0.5,
+            status_forcelist=[502, 503, 504],
+            allowed_methods={"GET", "POST"},
+        )
+    )
+)
+
+# SHARED FUNCTIONS
+
+def _set_pr_flag(flagurl, status, task, overviewurl, logger):
+    """"""
+    token = CONFIG.get("consumer", "dist-git_token")
+    params = {
+        "username": f"Fedora openQA tests - scratch build {task}",
+        "comment": f"openQA tests {status}",
+        "url": overviewurl,
+        "status": status,
+        "uid": f"openqa-{task}",
+    }
+    headers = {}
+    if token:
+        headers = {"Authorization": f"token {token}"}
+    resp = SESSION.post(flagurl, headers=headers, data=params, timeout=60)
+    if not resp.ok:
+        logger.error("Setting PR flag failed! Response: %s", resp)
+
+def _version_from_branch(body, logger):
+    """"""
+    branch = body.get("pullrequest", {}).get("branch", "")
+    if branch == "rawhide":
+        version = str(fedfind.helpers.get_current_release(branched=True) + 1)
+    elif branch.startswith("f"):
+        version = branch[1:]
+    else:
+        logger.debug("PR branch %s appears to not be a Fedora mainline release branch", branch)
+        return ("", "")
+    if not version.isdigit():
+        logger.debug("PR branch %s appears to not be a Fedora mainline release branch", branch)
+        return ("", "")
+    return (version, branch)
 
 # SCHEDULER
 
@@ -44,7 +98,8 @@ class OpenQAScheduler(object):
     """
 
     def __init__(self):
-        self.openqa_hostname = fedora_messaging.config.conf["consumer_config"]["openqa_hostname"]
+        self.openqa_hostname = CONFIG.get("consumer", "openqa_hostname")
+        self.openqa_baseurl = CONFIG.get("consumer", "openqa_baseurl")
         self.update_arches = fedora_messaging.config.conf["consumer_config"]["update_arches"]
         self.logger = logging.getLogger(self.__class__.__name__)
 
@@ -60,6 +115,10 @@ class OpenQAScheduler(object):
             # from Bodhi 8.0 onwards, this message should always and
             # only be published when we want to run tests:
             return self._consume_update(message.body, force=True)
+        elif 'pagure.pull-request.comment.added' in message.topic:
+            return self._consume_pr_comment(message.body)
+        elif 'pagure.pull-request.flag.updated' in message.topic:
+            return self._consume_pr_flag(message.body)
 
     def _check_mainline(self, body):
         """
@@ -139,7 +198,7 @@ class OpenQAScheduler(object):
         except ValueError as err:
             # I have some thoughts for Past Adam re: exception strategy
             if "urlopen_retries: Failed to open" in str(err):
-                self.logger.exception(f"Scheduling for {builddir} failed!")
+                self.logger.exception("Scheduling for %s failed!", builddir)
                 jobs = []
             else:
                 raise
@@ -190,6 +249,92 @@ class OpenQAScheduler(object):
             self.logger.info(tmpl, ', '.join(flavors), advisory)
         self._update_schedule(advisory, version, flavors, force=force, updic=update)
 
+    def _jobs_from_flag(self, flag, version, branch, prid, pkgname):
+        """"""
+        if flag.get("comment") == "RPM build succeeded." and flag.get("user", {}).get("name") == "packit":
+            if not f" {branch} " in flag.get("username", ""):
+                self.logger.debug("This scratch build is not for the PR branch, ignoring")
+                return False
+            task = flag.get("url", "").split("taskID=")[-1]
+            if not task.isdigit():
+                self.logger.warning("Task ID %s unexpected!", task)
+                return False
+            # discover appropriate flavors
+            critpath = SESSION.get("https://bodhi.fedoraproject.org/get_critpath_components", timeout=60).json()
+            groups = [group for group in critpath if pkgname in critpath[group]]
+            # this is hella janky, but we'll just construct an update
+            # dict realistic enough we can reuse get_critpath_flavors
+            fakeupd = {"critpath_groups": " ".join(groups), "release": {"version": version}}
+            flavors = schedule.get_critpath_flavors(fakeupd) or None
+            flavtext = "all"
+            if flavors:
+                flavtext = ", ".join(flavors)
+            build = f"PR-{task}-{prid}"
+            self.logger.info("Running tests for %s flavors on pull request %s", flavtext, prid)
+            self._update_schedule(build, version, flavors, force=True, updic=None)
+            overviewurl = f"{self.openqa_baseurl}/tests/overview?"
+            params = {"distri": "fedora", "build": f"{build}-NOREPORT", "groupid": 2, "version": version}
+            overviewurl += urlencode(params)
+            return (task, overviewurl)
+        return False
+
+    def _consume_pr_comment(self, body):
+        """Handle dist-git pull request comment messages. If this is
+        a comment requesting tests on a package PR for a release
+        branch, check whether the scratch build is done already. If it
+        is, schedule tests. If it's not, log the pull request URL in
+        a file in /run that lists PRs that should be tested when their
+        scratch builds complete.
+        """
+        pr = body.get("pullrequest", {})
+        fullurl = pr.get("full_url", "")
+        if "src.fedoraproject.org/rpms" not in fullurl:
+            self.logger.debug("Message is not for a package PR")
+            return
+        comment = pr.get("comments", [{}])[-1]
+        if not comment.get("comment") == "/openqa test":
+            self.logger.debug("Not a comment requesting openQA tests")
+            return
+        prid = pr.get("project", {}).get("fullname", "") + "#" + str(pr.get("id", ""))
+        pkgname = pr.get("project", {}).get("name", "")
+        version, branch = _version_from_branch(body, self.logger)
+        if not version:
+            return
+        flagurl = fullurl.replace("/rpms", "/api/0/rpms") + "/flag"
+        flags = SESSION.get(flagurl, timeout=60).json().get("flags", [{}])
+        flags.sort(key=lambda x: int(x.get("date_updated")))
+        for flag in flags:
+            ret = self._jobs_from_flag(flag, version, branch, prid, pkgname)
+            if ret:
+                task, overviewurl = ret
+                _set_pr_flag(flagurl, "pending", task, overviewurl, self.logger)
+                # only schedule for most recent scratch build
+                break
+        # if scratch build is not done yet, _consume_pr_flag will
+        # schedule tests later
+
+    def _consume_pr_flag(self, body):
+        """Handle dist-git pull request flag update messages. If this
+        message indicates a scratch build for a package PR completed,
+        and that PR has a "/openqa test" comment, schedule tests.
+        """
+        pr = body.get("pullrequest", {})
+        fullurl = pr.get("full_url", "")
+        version, branch = _version_from_branch(body, self.logger)
+        if not version:
+            return
+        # check for the magic comment
+        if not any(comment.get("comment", "") == "/openqa test" for comment in pr.get("comments", [])):
+            return
+        flag = body.get("flag", {})
+        prid = pr.get("project", {}).get("fullname", "") + "#" + str(pr["id"])
+        pkgname = pr.get("project", {}).get("name", "")
+        ret = self._jobs_from_flag(flag, version, branch, prid, pkgname)
+        if ret:
+            task, overviewurl = ret
+            flagurl = fullurl.replace("/rpms", "/api/0/rpms") + "/flag"
+            _set_pr_flag(flagurl, "pending", task, overviewurl, self.logger)
+
 
 # WIKI REPORTER
 
@@ -202,8 +347,8 @@ class OpenQAWikiReporter(object):
     def __init__(self):
         self.logger = logging.getLogger(self.__class__.__name__)
         self.do_report = fedora_messaging.config.conf["consumer_config"]["do_report"]
-        self.openqa_hostname = fedora_messaging.config.conf["consumer_config"]["openqa_hostname"]
-        self.openqa_baseurl = fedora_messaging.config.conf["consumer_config"]["openqa_baseurl"]
+        self.openqa_hostname = CONFIG.get("consumer", "openqa_hostname")
+        self.openqa_baseurl = CONFIG.get("consumer", "openqa_baseurl")
         self.wiki_hostname = fedora_messaging.config.conf["consumer_config"]["wiki_hostname"]
 
     def __call__(self, message):
@@ -240,8 +385,8 @@ class OpenQAResultsDBReporter(object):
     def __init__(self):
         self.logger = logging.getLogger(self.__class__.__name__)
         self.do_report = fedora_messaging.config.conf["consumer_config"]["do_report"]
-        self.openqa_hostname = fedora_messaging.config.conf["consumer_config"]["openqa_hostname"]
-        self.openqa_baseurl = fedora_messaging.config.conf["consumer_config"]["openqa_baseurl"]
+        self.openqa_hostname = CONFIG.get("consumer", "openqa_hostname")
+        self.openqa_baseurl = CONFIG.get("consumer", "openqa_baseurl")
         self.resultsdb_url = fedora_messaging.config.conf["consumer_config"]["resultsdb_url"]
 
     def __call__(self, message):
@@ -257,4 +402,71 @@ class OpenQAResultsDBReporter(object):
             resultsdb_url=self.resultsdb_url, jobs=newjobs, do_report=self.do_report,
             openqa_hostname=self.openqa_hostname, openqa_baseurl=self.openqa_baseurl)
 
+
+# DIST-GIT REPORTER
+
+
+class OpenQADistgitReporter(object):
+    """A fedora-messaging consumer that reports openQA results to
+    dist-git when all tests for a pull request are completed.
+    """
+
+    def __init__(self):
+        self.logger = logging.getLogger(self.__class__.__name__)
+        self.do_report = fedora_messaging.config.conf["consumer_config"]["do_report"]
+        self.openqa_baseurl = CONFIG.get("consumer", "openqa_baseurl")
+        self.openqa_hostname = CONFIG.get("consumer", "openqa_hostname")
+
+    def __call__(self, message):
+        """Consume incoming message."""
+        body = message.body
+        build = body.get("BUILD", "")
+        if not build.startswith("PR-"):
+            # not a pull request test job
+            self.logger.info("Build %s is not for a PR job", build)
+            return
+        remaining = body.get('remaining', 1)
+        if remaining != 0:
+            # not all tests done. FIXME: do progress? we need total
+            return
+        _, task, prid = build.replace("-NOREPORT", "").split("-", 2)
+        prrepo, prnum = prid.split("#")
+        # magic wait for restarted jobs
+        time.sleep(5)
+        # get all jobs
+        client = OpenQA_Client(self.openqa_hostname)
+        jobs = client.get_jobs(build=build, filter_dupes=True)
+        if not jobs:
+            # wat
+            return
+        # this is the same for all jobs, just get it from the first
+        version = jobs[0]["settings"]["VERSION"]
+        if not all(job["state"] in JOB_FINAL_STATES for job in jobs):
+            # even though remaining was 0, actually we have live jobs
+            self.logger.info("found live jobs:")
+            self.logger.info(jobs)
+            return
+        if any(job["result"] in JOB_NOT_COMPLETE_RESULTS for job in jobs):
+            status = "error"
+        elif all(job["result"] in JOB_OK_RESULTS for job in jobs):
+            status = "success"
+        else:
+            status = "failure"
+        overviewurl = f"{self.openqa_baseurl}/tests/overview?"
+        params = {"distri": "fedora", "build": build, "groupid": 2, "version": version}
+        overviewurl += urlencode(params)
+        # ugh this is kinda magic do we want to pass it via job settings?
+        flagurl = f"https://src.fedoraproject.org/api/0/{prrepo}/pull-request/{prnum}/flag"
+        if self.do_report:
+            _set_pr_flag(flagurl, status, task, overviewurl, self.logger)
+            return
+        self.logger.info(
+            "dist-git: would update flag %s to status %s for task %s with overview %s",
+            flagurl,
+            status,
+            task,
+            overviewurl
+        )
+        return
+
 # vim: set textwidth=120 ts=8 et sw=4:

src/fedora_openqa/schedule.py

@@ -574,6 +574,7 @@ def jobs_from_update(
     updic (dict or None): the Bodhi update dict, from the message or
     the web API. Must be provided to schedule update jobs
     """
+    prid = ""
     if version:
         version = str(version)
     if not flavors:
@@ -589,12 +590,18 @@ def jobs_from_update(
     if isinstance(update, str) and update.isdigit():
         update = [update]
 
-    if isinstance(update, list):
-        if not all(item.isdigit() for item in update):
-            raise TriggerException("Can only pass multiple Koji tasks, not updates or side tags or COPRs!")
-        idstring = "_".join(update)
-        # Koji task ID: treat as a non-reported scratch build test
-        build = f"Kojitask-{idstring}-NOREPORT"
+    if isinstance(update, list) or update.startswith("PR-"):
+        if isinstance(update, list):
+            if not all(item.isdigit() for item in update):
+                raise TriggerException("Can only pass multiple Koji tasks, not updates or side tags or COPRs!")
+            idstring = "_".join(update)
+            # Koji task ID: treat as a non-reported scratch build test
+            build = f"Kojitask-{idstring}-NOREPORT"
+        else:
+            # format: PR-(taskid)-(pullrequestid), we want 'idstring'
+            # to be the task ID
+            build = f"{update}-NOREPORT"
+            _, idstring, prid = update.split("-", 2)
         # we'll set ADVISORY and ADVISORY_OR_TASK for updates, and KOJITASK and
         # ADVISORY_OR_TASK for Koji tasks. I'd probably have designed this more
         # cleanly if doing it from scratch, but we started with updates and added
@@ -693,6 +700,8 @@ def jobs_from_update(
         'QEMU_HOST_IP': '172.16.2.2',
         'NICTYPE_USER_OPTIONS': 'net=172.16.2.0/24',
     })
+    if prid:
+        baseparams["PULL_REQUEST"] = prid
 
     # include whether updates-testing is active for the release; this
     # determines whether we use the buildroot repo or not

tests/test_consumers.py

@@ -19,6 +19,7 @@
 
 # these are all kinda inappropriate for pytest patterns
 # pylint: disable=old-style-class, no-init, protected-access, no-self-use, unused-argument
+# pylint: disable=too-many-arguments, too-many-positional-arguments
 
 """Tests for the fedmsg consumers."""
 
@@ -390,56 +391,193 @@ FCOSBUILDNOTF.body["state"] = "STARTED"
 FCOSBUILDNOTS = copy.deepcopy(FCOSBUILD)
 FCOSBUILDNOTS.body["result"] = "FAILURE"
 
+# dist-git (Pagure) "/openqa test" comment message
+DGPLZTEST = Message(
+    topic="org.fedoraproject.prod.pagure.pull-request.comment.added",
+    body={
+        "agent": "adamwill",
+        "pullrequest": {
+            "branch": "rawhide",
+            "branch_from": "my_pr_branch",
+            "comments": [
+                {
+                    "comment": "/openqa test",
+                    "commit": None,
+                    "date_created": "1782862483",
+                    "user": {
+                        "full_url": "https://src.fedoraproject.org/user/adamwill",
+                        "name": "adamwill",
+                    }
+                }
+            ],
+            "full_url": "https://src.fedoraproject.org/rpms/unetbootin/pull-request/1",
+            "id": 1,
+            "project": {
+                "full_url": "https://src.fedoraproject.org/rpms/fedfind",
+                "fullname": "rpms/fedfind",
+                "name": "fedfind",
+            }
+        }
+    }
+)
+# not a "/openqa test" comment message
+DGNOTEST = copy.deepcopy(DGPLZTEST)
+DGNOTEST.body["pullrequest"]["comments"][0]["comment"] = "hi there"
+
+# not a PR message (this is unlikely to really happen, but oh well)
+DGNOPR = copy.deepcopy(DGPLZTEST)
+DGNOPR.body["pullrequest"]["full_url"] = "https://src.fedoraproject.org/modules/postgresql/pull-request/1"
+
+# good comment, good URL, but weird branch
+DGBADBRANCH = copy.deepcopy(DGPLZTEST)
+DGBADBRANCH.body["pullrequest"]["branch"] = "someotherbranch"
+
+# another branch variant to test another codepath
+DGBADBRANCH2 = copy.deepcopy(DGPLZTEST)
+DGBADBRANCH2.body["pullrequest"]["branch"] = "f43foobar"
+
+# flag showing a successful RPM scratch build (reused later)
+GOODFLAG = {
+    "comment": "RPM build succeeded.",
+    "date_updated": "1781908960",
+    "user": {"name": "packit"},
+    "username": "Packit - scratch build - rawhide [8e56523]",
+    "url": "https://koji.fedoraproject.org/koji/taskinfo?taskID=147313590",
+}
+
+# same, but a bit older
+OLDGOODFLAG = copy.deepcopy(GOODFLAG)
+OLDGOODFLAG["date_updated"] = "1781905000"
+OLDGOODFLAG["username"] = "Packit - scratch build - rawhide [abc1234]"
+OLDGOODFLAG["url"] = "https://koji.fedoraproject.org/koji/taskinfo?taskID=147313400"
+
+# dist-git PR flag update message indicating a completed scratch build
+# with "/openqa test" comment
+DGFLAGSCRATCH = Message(
+    topic="org.fedoraproject.prod.pagure.pull-request.flag.updated",
+    body={
+        "agent": "adamwill",
+        "flag": copy.deepcopy(GOODFLAG),
+        "pullrequest": {
+            "branch": "rawhide",
+            "branch_from": "my_pr_branch",
+            "comments": [
+                {"comment": "/openqa test"}
+            ],
+            "id": 1,
+            "project": {
+                "full_url": "https://src.fedoraproject.org/rpms/fedfind",
+                "fullname": "rpms/fedfind",
+                "name": "fedfind",
+            }
+        }
+    }
+)
+# same, but no /openqa test comment
+DGFLAGNOCOMMENT = copy.deepcopy(DGFLAGSCRATCH)
+DGFLAGNOCOMMENT.body["pullrequest"]["comments"][0]["comment"] = "something else"
+
+# bad branch
+DGFLAGBADBRANCH = copy.deepcopy(DGFLAGSCRATCH)
+DGFLAGBADBRANCH.body["pullrequest"]["branch"] = "someotherbranch"
+
+# completed test for a dist-git PR with 0 remaining (dist-git reporter
+# should report)
+DGPASSMSG = Message(
+    topic="org.fedoraproject.stg.openqa.job.done",
+    body={
+        "BUILD": "PR-146808563-rpms/fedfind#2-NOREPORT",
+        "remaining": 0,
+    }
+)
+
+# non-zero remaining (should not report)
+DGPASSNZ = copy.deepcopy(DGPASSMSG)
+DGPASSNZ.body["remaining"] = 2
+
+# completed test for some other build with 0 remaining (dist-git
+# reporter should not report)
+DGPASSOTHER = copy.deepcopy(DGPASSMSG)
+DGPASSOTHER.body["BUILD"] = "Update-FEDORA-2026-8eabdaaa4f"
+
 # initialize a few test consumers with different configs
 PRODCONF = {
     'consumer_config': {
-        'do_report':True,
-        'openqa_hostname':'openqa.fedoraproject.org',
-        'openqa_baseurl':'https://openqa.fedoraproject.org',
-        'wiki_hostname':'fedoraproject.org',
-        'resultsdb_url':'http://resultsdb01.qa.fedoraproject.org/resultsdb_api/api/v2.0/',
-        'update_arches':["x86_64"]
+        'do_report': True,
+        'wiki_hostname': 'fedoraproject.org',
+        'resultsdb_url': 'http://resultsdb01.qa.fedoraproject.org/resultsdb_api/api/v2.0/',
+        'update_arches': ["x86_64"]
     }
 }
 STGCONF = {
     'consumer_config': {
-        'do_report':False,
-        'openqa_hostname':'openqa.stg.fedoraproject.org',
-        'openqa_baseurl':'https://openqa.stg.fedoraproject.org',
-        'wiki_hostname':'stg.fedoraproject.org',
-        'resultsdb_url':'http://resultsdb-stg01.qa.fedoraproject.org/resultsdb_api/api/v2.0/',
-        'update_arches':["x86_64", "ppc64le"]
+        'do_report': False,
+        'wiki_hostname': 'stg.fedoraproject.org',
+        'resultsdb_url': 'http://resultsdb-stg01.qa.fedoraproject.org/resultsdb_api/api/v2.0/',
+        'update_arches': ["x86_64", "ppc64le"]
     }
 }
 TESTCONF = {
     'consumer_config': {
-        'do_report':False,
-        'openqa_hostname':'localhost',
-        'openqa_baseurl':'https://localhost',
-        'wiki_hostname':'stg.fedoraproject.org',
-        'resultsdb_url':'http://localhost:5001/api/v2.0/',
-        'update_arches':["x86_64", "ppc64le"]
+        'do_report': False,
+        'wiki_hostname': 'stg.fedoraproject.org',
+        'resultsdb_url': 'http://localhost:5001/api/v2.0/',
+        'update_arches': ["x86_64", "ppc64le"]
     }
 }
 
 with mock.patch.dict('fedora_messaging.config.conf', PRODCONF):
+    fedora_openqa.consumer.CONFIG.set("consumer", "openqa_baseurl", "https://openqa.fedoraproject.org")
+    fedora_openqa.consumer.CONFIG.set("consumer", "openqa_hostname", "openqa.fedoraproject.org")
     PRODSCHED = fedora_openqa.consumer.OpenQAScheduler()
     PRODWIKI = fedora_openqa.consumer.OpenQAWikiReporter()
     PRODRDB = fedora_openqa.consumer.OpenQAResultsDBReporter()
+    PRODDG = fedora_openqa.consumer.OpenQADistgitReporter()
 with mock.patch.dict('fedora_messaging.config.conf', STGCONF):
+    fedora_openqa.consumer.CONFIG.set("consumer", "openqa_baseurl", "https://openqa.stg.fedoraproject.org")
+    fedora_openqa.consumer.CONFIG.set("consumer", "openqa_hostname", "openqa.stg.fedoraproject.org")
     STGSCHED = fedora_openqa.consumer.OpenQAScheduler()
     STGWIKI = fedora_openqa.consumer.OpenQAWikiReporter()
     STGRDB = fedora_openqa.consumer.OpenQAResultsDBReporter()
+    STGDG = fedora_openqa.consumer.OpenQADistgitReporter()
 with mock.patch.dict('fedora_messaging.config.conf', TESTCONF):
+    fedora_openqa.consumer.CONFIG.set("consumer", "openqa_baseurl", "https://localhost")
+    fedora_openqa.consumer.CONFIG.set("consumer", "openqa_hostname", "localhost")
     TESTSCHED = fedora_openqa.consumer.OpenQAScheduler()
     TESTWIKI = fedora_openqa.consumer.OpenQAWikiReporter()
     TESTRDB = fedora_openqa.consumer.OpenQAResultsDBReporter()
+    TESTDG = fedora_openqa.consumer.OpenQADistgitReporter()
 
 PRODS = (PRODWIKI, PRODRDB, PRODSCHED)
 STGS = (STGWIKI, STGRDB, STGSCHED)
 # we don't include TESTSCHED as its values are hardcoded
 TESTS = (TESTWIKI, TESTRDB)
 
+# these are the values MockResponse returns for a flag query and
+# any other query (respectively), so tests can update them on the fly
+RESPFLAGS = [copy.deepcopy(OLDGOODFLAG), copy.deepcopy(GOODFLAG)]
+RESPGROUPS = {}
+
+def reset_resp():
+    global RESPFLAGS
+    global RESPGROUPS
+    RESPFLAGS = [copy.deepcopy(OLDGOODFLAG), copy.deepcopy(GOODFLAG)]
+    RESPGROUPS = {}
+
+
+class MockResponse:
+    """Mock requests response which returns different content
+    depending on the requested URL.
+    """
+    def __init__(self, *args, **kwargs):
+        self.url = args[0]
+
+    def json(self):
+        if "/flag" in self.url:
+            return {"flags": RESPFLAGS}
+        else:
+            return RESPGROUPS
+
 
 @pytest.mark.usefixtures("ffmock")
 class TestConsumers:
@@ -507,7 +645,7 @@ class TestConsumers:
             (CRITPATHEDITUC, True, False, None, None),
             (FCOSBUILD, False, None, None, None),
             (FCOSBUILDNOTF, False, False, None, None),
-            (FCOSBUILDNOTS, False, False, None, None)
+            (FCOSBUILDNOTS, False, False, None, None),
         ]
     )
     def test_scheduler(self, fake_fcosbuild, fake_update, fake_schedule, consumer,
@@ -560,6 +698,131 @@ class TestConsumers:
         #fake_schedule.reset_mock()
 
 
+    @mock.patch('fedora_openqa.consumer.SESSION.get', autospec=True, side_effect=MockResponse)
+    @mock.patch('fedora_openqa.consumer._set_pr_flag', autospec=True)
+    @mock.patch('fedfind.helpers.get_current_release', return_value=38, autospec=True)
+    @mock.patch('fedora_openqa.schedule.jobs_from_update', return_value=[1], autospec=True)
+    @pytest.mark.parametrize(
+        "consumer,oqah",
+        [
+            (PRODSCHED, 'openqa.fedoraproject.org'),
+            (STGSCHED, 'openqa.stg.fedoraproject.org'),
+            (TESTSCHED, 'localhost'),
+        ]
+    )
+    def test_scheduler_distgit_comment(self, mockjfu, mockgcr, mockspf, mockget, consumer, oqah):
+        """Test OpenQAScheduler with dist-git PR comments. The code
+        path here is pretty different from all others and requires
+        different mocks, so it's split out.
+        """
+        reset_resp()
+        archcount = len(consumer.update_arches)
+        consumer(DGPLZTEST)
+        # should only have scheduled for *one* of the flags, so only
+        # archcount jobs
+        assert mockjfu.call_count == archcount
+        assert mockjfu.call_args[1]["openqa_hostname"] == oqah
+        assert mockjfu.call_args[1]["flavors"] == None
+        assert mockjfu.call_args[0][1] == "39"
+        mockjfu.reset_mock()
+        # test flavor discovery
+        global RESPGROUPS
+        RESPGROUPS = {"critical-path-apps": ["fedfind"]}
+        consumer(DGPLZTEST)
+        assert mockjfu.call_args[1]["flavors"] == {"kde", "server", "workstation"}
+        mockjfu.reset_mock()
+        # test different flag values
+        # only return one flag - the second - for easy tweaking
+        global RESPFLAGS
+        flag = RESPFLAGS[1]
+        RESPFLAGS = [flag]
+        # tweak flag attributes to expect no jobs
+        flag["comment"] = "RPM build failed."
+        consumer(DGPLZTEST)
+        assert mockjfu.call_count == 0
+        flag["comment"] = "RPM build succeeded."
+        flag["user"] = {"name": "foo"}
+        consumer(DGPLZTEST)
+        assert mockjfu.call_count == 0
+        flag["user"] = {"name": "packit"}
+        flag["username"] = "moo"
+        consumer(DGPLZTEST)
+        assert mockjfu.call_count == 0
+        flag["username"] = "Packit - scratch build - rawhide [8e56523]"
+        flag["url"] = "https://foo.com"
+        consumer(DGPLZTEST)
+        assert mockjfu.call_count == 0
+        flag["url"] = "https://koji.fedoraproject.org/koji/taskinfo?taskID=147313590"
+        # test with tweaked messages that should not produce jobs
+        consumer(DGNOTEST)
+        assert mockjfu.call_count == 0
+        consumer(DGNOPR)
+        assert mockjfu.call_count == 0
+        consumer(DGBADBRANCH)
+        assert mockjfu.call_count == 0
+        consumer(DGBADBRANCH2)
+        assert mockjfu.call_count == 0
+
+
+    @mock.patch('fedora_openqa.consumer.SESSION.get', autospec=True)
+    @mock.patch('fedora_openqa.consumer._set_pr_flag', autospec=True)
+    @mock.patch('fedfind.helpers.get_current_release', return_value=38, autospec=True)
+    @mock.patch('fedora_openqa.schedule.jobs_from_update', return_value=[1], autospec=True)
+    @pytest.mark.parametrize(
+        "consumer,oqah",
+        [
+            (PRODSCHED, 'openqa.fedoraproject.org'),
+            (STGSCHED, 'openqa.stg.fedoraproject.org'),
+            (TESTSCHED, 'localhost'),
+        ]
+    )
+    def test_scheduler_distgit_flag(self, mockjfu, mockgcr, mockspf, mockget, consumer, oqah):
+        """Test OpenQAScheduler with dist-git PR flag updates. The
+        code path here is pretty different from all others and
+        requires different mocks, so it's split out.
+        """
+        archcount = len(consumer.update_arches)
+        consumer(DGFLAGSCRATCH)
+        assert mockjfu.call_count == archcount
+        assert mockjfu.call_args[1]["openqa_hostname"] == oqah
+        assert mockjfu.call_args[0][1] == "39"
+        mockjfu.reset_mock()
+        consumer(DGFLAGNOCOMMENT)
+        assert mockjfu.call_count == 0
+        consumer(DGFLAGBADBRANCH)
+        assert mockjfu.call_count == 0
+
+
+    @mock.patch("fedora_openqa.consumer.SESSION.post", autospec=True)
+    def test_set_pr_flags(self, mockpost):
+        """Test the _set_pr_flag convenience function."""
+        fedora_openqa.consumer.CONFIG.set("consumer", "dist-git_token", "testtoken")
+        flagurl = "https://src.fedoraproject.org/api/0/rpms/fedfind/pull-request/2/flag"
+        mocklog = mock.MagicMock()
+        expdata = {
+            "username": "Fedora openQA tests - scratch build 146822887",
+            "comment": "openQA tests pending",
+            "url": "https://test.url",
+            "status": "pending",
+            "uid": "openqa-146822887",
+        }
+        fedora_openqa.consumer._set_pr_flag(flagurl, "pending", "146822887", "https://test.url", mocklog)
+        assert mockpost.call_count == 1
+        assert mockpost.call_args == (
+            (flagurl,),
+            {
+                "headers": {"Authorization": "token testtoken"},
+                "data": expdata,
+                "timeout": 60,
+            }
+        )
+        assert mocklog.call_count == 0
+        mockpost.return_value.ok = False
+        fedora_openqa.consumer._set_pr_flag(flagurl, "pending", "146822887", "https://test.url", mocklog)
+        assert mocklog.error.call_count == 1
+        assert mocklog.error.call_args[0][0].startswith("Setting PR flag failed")
+
+
     @mock.patch('fedora_openqa.report.wiki_report', autospec=True)
     @pytest.mark.parametrize(
         "consumer,expected",
@@ -671,4 +934,86 @@ class TestConsumers:
         assert fake_report.call_count == 2
         assert fake_sleep.call_count == 1
 
+    @mock.patch("fedora_openqa.consumer._set_pr_flag", autospec=True)
+    def test_distgit_reporter_noreport(self, mockspf, caplog):
+        """Test cases where dist-git PR result reporter bails early."""
+        caplog.set_level(logging.INFO)
+        # just use the prod consumer; behavior here does not differ
+        PRODDG(DGPASSNZ)
+        assert mockspf.call_count == 0
+        assert caplog.text == ""
+        PRODDG(DGPASSOTHER)
+        assert mockspf.call_count == 0
+        assert "is not for a PR job" in caplog.text
+
+
+    @mock.patch("time.sleep", autospec=True)
+    @mock.patch("openqa_client.client.OpenQA_Client.openqa_request", autospec=True)
+    @mock.patch("fedora_openqa.consumer._set_pr_flag", autospec=True)
+    @pytest.mark.parametrize(
+        "consumer,oqah",
+        [
+            (PRODDG, "openqa.fedoraproject.org"),
+            (STGDG, "openqa.stg.fedoraproject.org"),
+            (TESTDG, "localhost"),
+        ]
+    )
+    @pytest.mark.parametrize(
+        # jobs: we'll have openqa_request return as many jobs as there
+        # are items here, with the specified state and result
+        # status: the expected status to be reported back to dist-git
+        # if empty, we expect no report to dist-git
+        # text: some log output to check for in caplog if set
+        "jobs,status,text",
+        [
+            ([("done", "passed"), ("done", "passed")], "success", ""),
+            ([("done", "passed"), ("done", "failed")], "failure", ""),
+            ([("done", "passed"), ("done", "failed"), ("done", "incomplete")], "error", ""),
+            ([("done", "passed"), ("done", "failed"), ("running", "none")], "", "found live jobs"),
+            ([], "", ""),
+        ]
+    )
+    def test_distgit_reporter(self, mockspf, mockrequest, _, jobs, status, text, consumer, oqah, caplog):
+        """Test the dist-git PR result reporter consumer."""
+        caplog.set_level(logging.INFO)
+        flagurl = "https://src.fedoraproject.org/api/0/rpms/fedfind/pull-request/2/flag"
+        ovurl = (
+            f"https://{oqah}/tests/overview?"
+            "distri=fedora&build=PR-146808563-rpms%2Ffedfind%232-NOREPORT&groupid=2&version=39"
+        )
+        mockrequest.return_value = {
+            "jobs": [
+                {
+                    "id": 1,
+                    "settings": {"FLAVOR": "updates-server", "VERSION": "39"},
+                    "state": job[0],
+                    "result": job[1],
+                    "clone_id": None
+                } for job in jobs
+            ]
+        }
+        consumer(DGPASSMSG)
+        if status:
+            if consumer == PRODDG:
+                assert mockspf.call_count == 1
+                assert mockspf.call_args == (
+                    (
+                        "https://src.fedoraproject.org/api/0/rpms/fedfind/pull-request/2/flag",
+                        status,
+                        "146808563",
+                        ovurl,
+                        consumer.logger
+                    ),
+                    {}
+                )
+            else:
+                exptext = f"would update flag {flagurl} to status {status} for task 146808563 with overview {ovurl}"
+                assert exptext in caplog.text
+        else:
+            assert mockspf.call_count == 0
+            assert "would update flag" not in caplog.text
+
+        if text:
+            assert text in caplog.text
+
 # vim: set textwidth=120 ts=8 et sw=4:

tests/test_schedule.py

@@ -1081,6 +1081,53 @@ def test_jobs_from_update_kojitask(fakeclient, fakecurrr, fakecurrs, fakeget):
             flavors=['everything-boot-iso']
         )
 
+@mock.patch('requests.get', autospec=True)
+@mock.patch('fedfind.helpers.get_current_stables', return_value=[28, 29])
+@mock.patch('fedfind.helpers.get_current_release', return_value=29)
+@mock.patch('fedora_openqa.schedule.OpenQA_Client', autospec=True)
+def test_jobs_from_update_pullrequest(fakeclient, fakecurrr, fakecurrs, fakeget):
+    """Test jobs_from_update works as expected when passed a dist-git
+    PR. We don't need to recheck everything, just the differing vars.
+    """
+    # act as if u-t is enabled
+    fakeget.return_value.json.return_value = {'create_automatic_updates': False}
+    # the OpenQA_Client instance mock
+    fakeinst = fakeclient.return_value
+    # for now, return no 'jobs' (for the dupe query), one 'id' (for
+    # the post request)
+    fakeinst.openqa_request.return_value = {'jobs': [], 'ids': [1]}
+    ret = schedule.jobs_from_update('PR-146808563-rpms/fedfind#2', version='28', flavors=['everything-boot-iso'])
+    # should get one job for one flavor
+    assert ret == [1]
+    # find the POST calls
+    posts = [call for call in fakeinst.openqa_request.call_args_list if call[0][0] == 'POST']
+    # one flavor, one call
+    assert len(posts) == 1
+    parmdict = posts[0][1]["data"]
+    assert parmdict == {
+        'DISTRI': 'fedora',
+        'VERSION': '28',
+        'ARCH': 'x86_64',
+        'BUILD': 'PR-146808563-rpms/fedfind#2-NOREPORT',
+        'KOJITASK': '146808563',
+        'ADVISORY_OR_TASK': '146808563',
+        'PULL_REQUEST': 'rpms/fedfind#2',
+        'UPDATE_OR_TAG_REPO': 'nfs://172.16.2.110:/mnt/update_repo',
+        '_OBSOLETE': '1',
+        '_ONLY_OBSOLETE_SAME_BUILD': '1',
+        'START_AFTER_TEST': '',
+        'QEMU_HOST_IP': '172.16.2.2',
+        'NICTYPE_USER_OPTIONS': 'net=172.16.2.0/24',
+        'FLAVOR': 'updates-everything-boot-iso',
+        'CURRREL': '29',
+        'RAWREL': '30',
+        'UEFI_PFLASH_CODE': '%INSECURE_PFLASH_CODE%',
+        'UEFI_PFLASH_VARS': '%INSECURE_PFLASH_VARS%',
+        'UEFI_SECURE': '',
+        'UP1REL': '27',
+        'UP2REL': '26',
+    }
+
 @mock.patch('requests.get', autospec=True)
 @mock.patch('fedfind.helpers.get_current_stables', return_value=[28, 29])
 @mock.patch('fedfind.helpers.get_current_release', return_value=29)