Merge lp:~cprov/britney/testclient-api into lp:~canonical-ci-engineering/britney/queued-announce-and-collect

[Celso Providelo]

Initial implementation of a generic interface for proposed-migration (britney) to produce new-candidates events and consume test results from the CI-infrastructure.

Initial implementation of a generic interface for proposed-migration (britney) to produce new-candidates events and consume test results from the CI-infrastructure.

[Para Siva (psivaa)]

Thanks a lot cprov for the MP.
I really like your way of handling the cache. Making keys with name and version is a clever way. And thanks for adding tests too.

I have a few points,
1. To include the conf file changes,
2. An inline comment in avoiding duplicate entries in the reg
3. I think the cleanup will handle entries with one source having two or more versions at the same time, but want to double check.
4. Also i'm not sure if we're handling acking and nacking etc. I'm asking this because we're removing entries from the cache and that may lead to messages being in the queue forever?

review: Needs Fixing

[Celso Providelo (cprov)]

Psivaa,

Thanks for the review.

Regarding config-changes, they will be only necessary when we do integration tests (which I've mentioned are missing, on purpose, from this MP).
On point 2) I don't get the registry duplication, it's impossible, since it's a dictionary, perhaps you mean re-announcement ?
On point 3) take a look at the corresponding test, it may clarify things about cleanup() effects.
On message ack-ing, you are right it was missing in the collect() action, fixed.

[]

[Para Siva (psivaa)]

Thanks for fixing the comments cprov.

> On point 2) I don't get the registry duplication, it's impossible, since it's
> a dictionary, perhaps you mean re-announcement ?
>
Point 2 was about a notification about my inline comment, not using make_key bit, which you've fixed.

And thanks a lot again for this MP. Whole lot different than what I proposed.

review: Approve

[lp:~cprov/britney/testclient-api]

442. By Celso Providelo on 2015-05-22

addressing review comments

443. By Celso Providelo on 2015-05-22

extend existing tests for ignored test results (series mismatch).

=== modified file 'britney.py'
--- britney.py	2015-02-20 19:02:00 +0000
+++ britney.py	2015-05-22 17:40:09 +0000
@@ -227,6 +227,7 @@
                    PROVIDES, RDEPENDS, RCONFLICTS, MULTIARCH, ESSENTIAL)
 from autopkgtest import AutoPackageTest, ADT_PASS, ADT_EXCUSES_LABELS
 from boottest import BootTest
+from testclient import TestClient
 
 
 __author__ = 'Fabio Tranchitella and the Debian Release Team'
@@ -1984,6 +1985,62 @@
                         upgrade_me.remove(excuse.name)
                         unconsidered.append(excuse.name)
 
+        if (getattr(self.options, "testclient_enable", "no") == "yes" and
+            self.options.series):
+
+            # Filter only new source candidates excuses.
+            testing_excuses = []
+            for excuse in self.excuses:
+                # Skip removals, binary-only candidates, proposed-updates
+                # and unknown versions.
+                if (excuse.name.startswith("-") or
+                    "/" in excuse.name or
+                    "_" in excuse.name or
+                    excuse.ver[1] == "-"):
+                    continue
+                testing_excuses.append(excuse)
+
+            amqp_uris = getattr(
+                self.options, "testclient_amqp_uris", "").split()
+            testclient = TestClient(self.options.series, amqp_uris)
+
+            # Announce new candidates and collect new test results.
+            if not self.options.dry_run:
+                testclient.announce(testing_excuses)
+                testclient.collect()
+                testclient.cleanup(testing_excuses)
+
+            # Update excuses considering hints and required_tests (for gating).
+            required_tests = getattr(
+                self.options, "testclient_required_tests", "").split()
+            for excuse in testing_excuses:
+                hints = self.hints.search('force', package=excuse.name)
+                hints.extend(
+                    self.hints.search('force-badtest', package=excuse.name))
+                forces = [x for x in hints
+                          if same_source(excuse.ver[1], x.version)]
+                for test in testclient.getTests(excuse.name):
+                    label = TestClient.EXCUSE_LABELS.get(
+                        test.status, 'UNKNOWN STATUS')
+                    excuse.addhtml(
+                        "%s result: %s (<a href=\"%s\">results</a>)" % (
+                            test.name, label, test.result_url))
+                    if forces:
+                        excuse.addhtml(
+                            "Should wait for %s %s %s, but forced by "
+                            "%s" % (excuse.name, excuse.ver[1],
+                                    test.name, forces[0].user))
+                        continue
+                    if test.name not in required_tests:
+                        continue
+                    if test.status not in TestClient.VALID_STATUSES:
+                        excuse.addreason(test.name)
+                        if excuse.is_valid:
+                            excuse.is_valid = False
+                            excuse.addhtml("Not considered")
+                            upgrade_me.remove(excuse.name)
+                            unconsidered.append(excuse.name)
+
         # invalidate impossible excuses
         for e in self.excuses:
             # parts[0] == package name

=== added file 'testclient.py'
--- testclient.py	1970-01-01 00:00:00 +0000
+++ testclient.py	2015-05-22 17:40:09 +0000
@@ -0,0 +1,178 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2015 Canonical Ltd.
+
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+from __future__ import print_function
+
+
+from contextlib import (
+    contextmanager,
+    nested,
+)
+import json
+import os
+
+
+import kombu
+from kombu.pools import producers
+
+
+@contextmanager
+def json_cached_info(path):
+    """Context manager for caching a JSON object on disk."""
+    info = {}
+    if os.path.exists(path):
+        with open(path) as fp:
+            try:
+                info = json.load(fp)
+            except ValueError:
+                # cache is empty or corrupted (!).
+                info = {}
+    else:
+        dirname = os.path.dirname(path)
+        if not os.path.exists(dirname):
+            os.makedirs(dirname)
+    yield info
+    with open(path, 'w') as fp:
+        json.dump(info, fp, indent=2)
+
+
+def make_cache_key(name, version):
+    """Return a json-hashable key for given source & version."""
+    return '{}_{}'.format(name, version)
+
+
+class TestClient(object):
+    """Generic test client implementation.
+
+    announce: announcing new source candidates to a pre-defined rabbitmq
+              exchange (testing subsystems can hook/subscribe).
+
+    collect: collect test results for the context series from a pre-defined
+             rabbitmq exchange (other promotion agents can do the same).
+
+    cleanup: sanitize internal announcement/testing registry after runs. 
+    """
+
+    EXCHANGE_CANDIDATES = 'candidates.exchange'
+    EXCHANGE_RESULTS = 'results.exchange'
+
+    VALID_STATUSES = ('PASS', 'SKIP')
+
+    LABELS = {
+        "PASS": '<span style="background:#87d96c">Pass</span>',
+        "SKIP": '<span style="background:#ffff00">Skip</span>',
+        "FAIL": '<span style="background:#ff6666">Regression</span>',
+        "RUNNING": '<span style="background:#99ddff">Test in progress</span>',
+    }
+
+    def __init__(self, series, amqp_uris):
+        self.series = series
+        self.amqp_uris = amqp_uris
+
+    @property
+    def cache_path(self):
+        """Series-specific test announcement/result cache."""
+        return 'testclient/{}.json'.format(self.series)
+
+    @property
+    def results_queue(self):
+        """Series-specific queue for collecting tests results."""
+        return 'pm.results.{}'.format(self.series)
+
+    def announce(self, excuses):
+        """Announce new source candidates.
+
+        Post a message to the EXCHANGE_CANDATIDATES for every new given
+        excuses (cache announcementes so excuses do not get re-annouced).
+        """            
+        with nested(json_cached_info(self.cache_path),
+                    kombu.Connection(self.amqp_uris)) as (cache, connection):
+            # XXX cprov 20150521: nested() is deprecated, and multi-statement
+            # 'with' does not support nesting (i.e. the previous context
+            # manager is not available to the next, in this case
+            # 'connection').
+            with producers[connection].acquire(block=True) as producer:
+                publisher = connection.ensure(
+                    producer, producer.publish, max_retries=3)
+                exchange = kombu.Exchange(
+                    self.EXCHANGE_CANDIDATES, type="fanout")
+                for excuse in excuses:
+                    if make_cache_key(
+                            excuse.name, excuse.ver[1]) in cache.keys():
+                        continue
+                    payload = {
+                        'source_name': excuse.name,
+                        'source_version': excuse.ver[1],
+                        'series': self.series,
+                    }
+                    publisher(payload, exchange=exchange, declare=[exchange])
+                    cache[make_cache_key(excuse.name, excuse.ver[1])] = []
+
+    def collect(self):
+        """Collect available test results.
+
+        Consume all messages from the EXCHANGE_RESULTS (routed to a series-
+        specific queue). Ignore test results for other series and update
+        test results registry.
+        """
+        with nested(json_cached_info(self.cache_path),
+                    kombu.Connection(self.amqp_uris)) as (cache, connection):
+            exchange = kombu.Exchange(
+                self.EXCHANGE_RESULTS, type="fanout")
+            queue = kombu.Queue(self.results_queue, exchange)
+            # XXX cprov 20150521: same as above about nested context managers.
+            with connection.SimpleQueue(queue) as q:
+                for i in range(len(q)):
+                    msg = q.get()
+                    payload = msg.payload
+                    if payload.get('series') != self.series:
+                        continue
+                    tests = cache.setdefault(
+                        make_cache_key(
+                            payload.get('source_name'),
+                            payload.get('source_version')
+                        ), [])
+                    tests.append({
+                        'name': payload.get('test_name'),
+                        'status': payload.get('test_status'),
+                        'url': payload.get('test_url'),
+                    })
+                    msg.ack()
+
+    def cleanup(self, excuses):
+        """Remove test result entries without corresponding excuse.
+
+        If there is not excuse the test results are not relevant anymore.
+        """
+        with json_cached_info(self.cache_path) as cache:
+            current_keys = [
+                make_cache_key(e.name, e.ver[1]) for e in excuses]
+            cached_keys = list(cache.keys())
+            for k in cached_keys:
+                if k not in current_keys:
+                    del cache[k]
+
+    def getTests(self, name, version):
+        """Yields test results for a given source-version pair.
+
+        Tests results are a list of dictionaries container test-name, status
+        and url.
+        """
+        with json_cached_info(self.cache_path) as cache:
+            tests = cache.get(make_cache_key(name, version), [])
+            for test in tests:
+                yield {
+                    'name': test.get('name'),
+                    'status': test.get('status'),
+                    'url': test.get('url'),
+                }

=== added file 'tests/test_testclient.py'
--- tests/test_testclient.py	1970-01-01 00:00:00 +0000
+++ tests/test_testclient.py	2015-05-22 17:40:09 +0000
@@ -0,0 +1,211 @@
+#!/usr/bin/python
+# (C) 2015 Canonical Ltd.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+import kombu
+from kombu.pools import producers
+
+PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, PROJECT_DIR)
+
+from excuse import Excuse
+from testclient import (
+    json_cached_info,
+    make_cache_key,
+    TestClient,
+)
+
+
+class TestJsonCachedInfo(unittest.TestCase):
+
+    def setUp(self):
+        super(TestJsonCachedInfo, self).setUp()
+        (_dummy, self.test_cache) = tempfile.mkstemp()
+        self.addCleanup(os.unlink, self.test_cache)
+
+    def test_simple(self):
+        # `json_cached_info` context manager correctly persists a
+        # python dictionary on disk.
+        with json_cached_info(self.test_cache) as cache:
+            self.assertEqual({}, cache)
+            cache['foo'] = 'bar'
+
+        with open(self.test_cache) as fp:
+            self.assertEqual(
+                ['{\n',
+                 '  "foo": "bar"\n',
+                 '}'], fp.readlines())
+
+        with json_cached_info(self.test_cache) as cache:
+            self.assertEqual(cache['foo'], 'bar')
+
+
+def make_excuse(name, version):
+    """Return a `Excuse` for the give source name and version."""
+    e = Excuse(name)
+    e.set_vers('-', version)
+    return e
+
+
+class TestTestClient(unittest.TestCase):
+
+    def setUp(self):
+        super(TestTestClient, self).setUp()
+        self.path = tempfile.mkdtemp(prefix='testclient')
+        os.makedirs(os.path.join(self.path, 'testclient/'))
+        self.addCleanup(shutil.rmtree, self.path)
+        os.chdir(self.path)
+
+    def test_announce(self):
+        # 'announce' post messages to the EXCHANGE_CANDIDATES exchange and
+        # updates its internal cache.
+        amqp_uris = ['memory://']
+        testclient = TestClient('vivid', amqp_uris)
+        test_excuses = [
+            make_excuse('foo', '1.0'),
+            make_excuse('bar', '2.0'),
+        ]
+
+        with kombu.Connection(amqp_uris) as connection:
+            exchange = kombu.Exchange(
+                testclient.EXCHANGE_CANDIDATES, type="fanout")
+            queue = kombu.Queue('testing', exchange)
+            with connection.SimpleQueue(queue) as q:
+                q.queue.purge()
+                testclient.announce(test_excuses)
+                self.assertEqual(
+                    [{'series': 'vivid',
+                      'source_name': 'foo',
+                      'source_version': '1.0'},
+                     {'series': 'vivid',
+                      'source_name': 'bar',
+                      'source_version': '2.0'}],
+                    [q.get().payload for i in range(len(q))])
+
+        with json_cached_info(testclient.cache_path) as cache:
+            self.assertEqual(
+                {'bar_2.0': [], 'foo_1.0': []},
+                cache)
+
+    def test_collect(self):
+        # 'collect' collects test results and aggregates them in its
+        # internal cache.
+        amqp_uris = ['memory://']
+        testclient = TestClient('vivid', amqp_uris)
+
+        result_payloads = [
+            {'source_name': 'foo',
+             'source_version': '1.0',
+             'series': testclient.series,
+             'test_name': 'snappy',
+             'test_status': 'RUNNING',
+             'test_url': 'http://snappy.com/foo'},
+            {'source_name': 'bar',
+             'source_version': '1.0',
+             'series': testclient.series,
+             'test_name': 'ubuntu',
+             'test_status': 'RUNNING',
+             'test_url': 'http://ubuntu.com/foo'},
+            {'source_name': 'foo',
+             'source_version': '1.0',
+             'series': testclient.series,
+             'test_name': 'bbb',
+             'test_status': 'RUNNING',
+             'test_url': 'http://bbb.com/foo'},
+            # This result will be ignored due to the series mismatch.
+            {'source_name': 'zoing',
+             'source_version': '1.0',
+             'series': 'some-other-series',
+             'test_name': 'ubuntu',
+             'test_status': 'RUNNING',
+             'test_url': 'http://ubuntu.com/foo'},
+        ]
+
+        with kombu.Connection(amqp_uris) as connection:
+            with producers[connection].acquire(block=True) as producer:
+                # Just for binding destination queue to the exchange.
+                testclient.collect()
+                exchange = kombu.Exchange(
+                    testclient.EXCHANGE_RESULTS, type="fanout")
+                publisher = connection.ensure(
+                    producer, producer.publish, max_retries=3)
+                for payload in result_payloads:
+                    publisher(payload, exchange=exchange, declare=[exchange])
+                testclient.collect()
+
+        with json_cached_info(testclient.cache_path) as cache:
+            self.assertEqual(
+                {'foo_1.0': [{'name': 'snappy',
+                              'status': 'RUNNING',
+                              'url': 'http://snappy.com/foo'},
+                             {'name': 'bbb',
+                              'status': 'RUNNING',
+                              'url': 'http://bbb.com/foo'}],
+                 'bar_1.0': [{'name': 'ubuntu',
+                              'status': 'RUNNING',
+                              'url': 'http://ubuntu.com/foo'}]},
+                cache)
+
+    def test_cleanup(self):
+        # `cleanup` remove cache entries that are not present in the
+        # given excuses list (i.e. not relevant for promotion anymore).
+        amqp_uris = ['memory://']
+        testclient = TestClient('vivid', amqp_uris)
+        test_excuses = [
+            make_excuse('foo', '1.0'),
+            make_excuse('bar', '2.0'),
+        ]
+
+        with json_cached_info(testclient.cache_path) as cache:
+            cache[make_cache_key('foo', '0.9')] = []
+            cache[make_cache_key('foo', '1.0')] = []
+            cache[make_cache_key('bar', '2.0')] = []
+
+        testclient.cleanup(test_excuses)
+
+        with json_cached_info(testclient.cache_path) as cache:
+            self.assertEqual(
+                {'bar_2.0': [], 'foo_1.0': []},
+                cache)
+
+    def test_getTests(self):
+        # `getTests` yields cached tests results for a given source name
+        # and version.
+        amqp_uris = ['memory://']
+        testclient = TestClient('vivid', amqp_uris)
+
+        with json_cached_info(testclient.cache_path) as cache:
+            cache[make_cache_key('foo', '1.0')] = [
+                {'name': 'snappy',
+                 'status': 'RUNNING',
+                 'url': 'http://snappy.com/foo'},
+                {'name': 'bbb',
+                 'status': 'RUNNING',
+                 'url': 'http://bbb.com/foo'}
+            ]
+
+        self.assertEqual(
+            [{'name': 'snappy',
+              'status': 'RUNNING',
+              'url': 'http://snappy.com/foo'},
+             {'name': 'bbb',
+              'status': 'RUNNING',
+              'url': 'http://bbb.com/foo'}],
+            list(testclient.getTests('foo', '1.0')))
+
+        self.assertEqual(
+            [], list(testclient.getTests('bar', '1.0')))
+
+
+if __name__ == '__main__':
+    unittest.main()