Merge lp:~matiasb/launchpad/upload-snap-to-track-based-channels into lp:launchpad

[Matias Bordese]

Added initial channel track support when uploading snaps to the store.

[Colin Watson (cjwatson)]

Overall very good, thanks! I think this needs a few tweaks especially around making sure that invalid values can't be set over the API, but it's pretty close.

review: Needs Fixing

[Matias Bordese (matiasb)]

Replied comments, pushing updates in a minute.

[Colin Watson (cjwatson)]

review: Approve

=== modified file 'lib/lp/snappy/browser/snap.py'
--- lib/lp/snappy/browser/snap.py	2017-03-08 22:45:20 +0000
+++ lib/lp/snappy/browser/snap.py	2017-03-30 14:10:04 +0000
@@ -32,7 +32,6 @@
     List,
     TextLine,
     )
-from zope.schema.interfaces import IVocabularyFactory
 
 from lp import _
 from lp.app.browser.launchpadform import (
@@ -62,7 +61,6 @@
 from lp.registry.interfaces.pocket import PackagePublishingPocket
 from lp.services.features import getFeatureFlag
 from lp.services.helpers import english_list
-from lp.services.propertycache import cachedproperty
 from lp.services.scripts import log
 from lp.services.webapp import (
     canonical_url,
@@ -82,6 +80,7 @@
 from lp.services.webapp.url import urlappend
 from lp.services.webhooks.browser import WebhookTargetNavigationMixin
 from lp.snappy.browser.widgets.snaparchive import SnapArchiveWidget
+from lp.snappy.browser.widgets.storechannels import StoreChannelsWidget
 from lp.snappy.interfaces.snap import (
     CannotAuthorizeStoreUploads,
     ISnap,
@@ -190,14 +189,9 @@
         else:
             return 'Built on request'
 
-    @cachedproperty
+    @property
     def store_channels(self):
-        vocabulary = getUtility(
-            IVocabularyFactory, name='SnapStoreChannel')(self.context)
-        channel_titles = []
-        for channel in self.context.store_channels:
-            channel_titles.append(vocabulary.getTermByToken(channel).title)
-        return ', '.join(channel_titles)
+        return ', '.join(self.context.store_channels)
 
 
 def builds_for_snap(snap):
@@ -388,7 +382,7 @@
         ]
     custom_widget('store_distro_series', LaunchpadRadioWidget)
     custom_widget('auto_build_archive', SnapArchiveWidget)
-    custom_widget('store_channels', LabeledMultiCheckBoxWidget)
+    custom_widget('store_channels', StoreChannelsWidget)
     custom_widget('auto_build_pocket', LaunchpadDropdownWidget)
 
     help_links = {
@@ -694,7 +688,7 @@
         'store_channels',
         ]
     custom_widget('store_distro_series', LaunchpadRadioWidget)
-    custom_widget('store_channels', LabeledMultiCheckBoxWidget)
+    custom_widget('store_channels', StoreChannelsWidget)
     custom_widget('vcs', LaunchpadRadioWidget)
     custom_widget('git_ref', GitRefWidget, allow_external=True)
     custom_widget('auto_build_archive', SnapArchiveWidget)
 
=== modified file 'lib/lp/snappy/browser/tests/test_snap.py'
--- lib/lp/snappy/browser/tests/test_snap.py	2017-02-03 11:30:05 +0000
+++ lib/lp/snappy/browser/tests/test_snap.py	2017-03-30 14:10:04 +0000
@@ -393,6 +393,7 @@
         browser.getControl("Registered store package name").value = (
             "store-name")
         self.assertFalse(browser.getControl("Stable").selected)
+        browser.getControl(name="field.store_channels.track").value = "track"
         browser.getControl("Edge").selected = True
         root_macaroon = Macaroon()
         root_macaroon.add_third_party_caveat(
@@ -419,7 +420,7 @@
             name=u"snap-name", source=branch, store_upload=True,
             store_series=self.snappyseries, store_name=u"store-name",
             store_secrets={"root": root_macaroon_raw},
-            store_channels=["edge"]))
+            store_channels=["track/edge"]))
         self.assertThat(self.request, MatchesStructure.byEquality(
             url="http://sca.example/dev/api/acl/", method="POST"))
         expected_body = {
@@ -946,12 +947,14 @@
             registrant=self.person, owner=self.person,
             distroseries=self.distroseries, store_upload=True,
             store_series=self.snappyseries, store_name=u"one",
-            store_channels=["edge"])
+            store_channels=["track/edge"])
         view_url = canonical_url(snap, view_name="+edit")
         browser = self.getNonRedirectingBrowser(url=view_url, user=self.person)
         browser.getControl("Registered store package name").value = "two"
+        self.assertEqual("track", browser.getControl("Track").value)
+        self.assertTrue(browser.getControl("Edge").selected)
+        browser.getControl("Track").value = ""
         browser.getControl("Stable").selected = True
-        self.assertTrue(browser.getControl("Edge").selected)
         root_macaroon = Macaroon()
         root_macaroon.add_third_party_caveat(
             urlsplit(config.launchpad.openid_provider_root).netloc, "",
@@ -1373,24 +1376,11 @@
         view = create_initialized_view(snap, "+index")
         self.assertEqual("", view.store_channels)
 
-    def test_store_channels_uses_titles(self):
-        snap_store_client = FakeMethod()
-        snap_store_client.listChannels = FakeMethod(result=[
-            {"name": "stable", "display_name": "Stable"},
-            {"name": "edge", "display_name": "Edge"},
-            {"name": "old", "display_name": "Old channel"},
-            ])
-        self.useFixture(
-            ZopeUtilityFixture(snap_store_client, ISnapStoreClient))
-        snap = self.factory.makeSnap(store_channels=["stable", "old"])
-        view = create_initialized_view(snap, "+index")
-        self.assertEqual("Stable, Old channel", view.store_channels)
-        snap_store_client.listChannels.result = [
-            {"name": "stable", "display_name": "Stable"},
-            {"name": "edge", "display_name": "Edge"},
-            ]
-        view = create_initialized_view(snap, "+index")
-        self.assertEqual("Stable, old", view.store_channels)
+    def test_store_channels_display(self):
+        snap = self.factory.makeSnap(
+            store_channels=["track/stable", "track/edge"])
+        view = create_initialized_view(snap, "+index")
+        self.assertEqual("track/stable, track/edge", view.store_channels)
 
 
 class TestSnapRequestBuildsView(BaseTestSnapView):
 
=== added file 'lib/lp/snappy/browser/widgets/storechannels.py'
--- lib/lp/snappy/browser/widgets/storechannels.py	1970-01-01 00:00:00 +0000
+++ lib/lp/snappy/browser/widgets/storechannels.py	2017-03-30 14:10:04 +0000
@@ -0,0 +1,152 @@
+# Copyright 2017 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__metaclass__ = type
+
+__all__ = [
+    'StoreChannelsWidget',
+    ]
+
+from z3c.ptcompat import ViewPageTemplateFile
+from zope.formlib.interfaces import IInputWidget, WidgetInputError
+from zope.formlib.utility import setUpWidget
+from zope.formlib.widget import (
+    BrowserWidget,
+    CustomWidgetFactory,
+    InputErrors,
+    InputWidget,
+    )
+from zope.interface import implementer
+from zope.schema import (
+    Choice,
+    List,
+    TextLine,
+    )
+
+from lp import _
+from lp.app.errors import UnexpectedFormData
+from lp.app.validators import LaunchpadValidationError
+from lp.app.widgets.itemswidgets import LabeledMultiCheckBoxWidget
+from lp.services.webapp.interfaces import (
+    IAlwaysSubmittedWidget,
+    ISingleLineWidgetLayout,
+    )
+from lp.snappy.validators.channels import (
+    channel_components_delimiter,
+    split_channel_name,
+    )
+
+
+@implementer(ISingleLineWidgetLayout, IAlwaysSubmittedWidget, IInputWidget)
+class StoreChannelsWidget(BrowserWidget, InputWidget):
+
+    template = ViewPageTemplateFile("templates/storechannels.pt")
+    display_label = False
+    _separator = channel_components_delimiter
+    _default_track = 'latest'
+    _widgets_set_up = False
+
+    def __init__(self, field, value_type, request):
+        # We don't use value_type.
+        super(StoreChannelsWidget, self).__init__(field, request)
+
+    def setUpSubWidgets(self):
+        if self._widgets_set_up:
+            return
+        fields = [
+            TextLine(__name__="track", title=u"Track", required=False,
+                     description=_(
+                         "Track defines a series for your software. "
+                         "If not specified, the default track ('latest') is "
+                         "assumed.")
+                     ),
+            List(__name__="risks", title=u"Risk", required=False,
+                 value_type=Choice(vocabulary="SnapStoreChannel"),
+                 description=_(
+                     "Risks denote the stability of your software.")),
+            ]
+
+        self.risks_widget = CustomWidgetFactory(LabeledMultiCheckBoxWidget)
+        for field in fields:
+            setUpWidget(
+                self, field.__name__, field, IInputWidget, prefix=self.name)
+        self.risks_widget.orientation = 'horizontal'
+        self._widgets_set_up = True
+
+    @property
+    def has_risks_vocabulary(self):
+        risks_widget = getattr(self, 'risks_widget', None)
+        return risks_widget and bool(risks_widget.vocabulary)
+
+    def buildChannelName(self, track, risk):
+        """Return channel name composed from given track and risk."""
+        channel = risk
+        if track and track != self._default_track:
+            channel = track + self._separator + risk
+        return channel
+
+    def splitChannelName(self, channel):
+        """Return extracted track and risk from given channel name."""
+        try:
+            track, risk = split_channel_name(channel)
+        except ValueError:
+            raise AssertionError("Not a valid value: %r" % channel)
+        return track, risk
+
+    def setRenderedValue(self, value):
+        """See `IWidget`."""
+        self.setUpSubWidgets()
+        if value:
+            # NOTE: atm target channels must belong to the same track
+            tracks = set()
+            risks = []
+            for channel in value:
+                track, risk = self.splitChannelName(channel)
+                tracks.add(track)
+                risks.append(risk)
+            if len(tracks) != 1:
+                raise AssertionError("Not a valid value: %r" % value)
+            track = tracks.pop()
+            self.track_widget.setRenderedValue(track)
+            self.risks_widget.setRenderedValue(risks)
+        else:
+            self.track_widget.setRenderedValue(None)
+            self.risks_widget.setRenderedValue(None)
+
+    def hasInput(self):
+        """See `IInputWidget`."""
+        return ("%s.risks" % self.name) in self.request.form
+
+    def hasValidInput(self):
+        """See `IInputWidget`."""
+        try:
+            self.getInputValue()
+            return True
+        except (InputErrors, UnexpectedFormData):
+            return False
+
+    def getInputValue(self):
+        """See `IInputWidget`."""
+        self.setUpSubWidgets()
+        risks = self.risks_widget.getInputValue()
+        track = self.track_widget.getInputValue()
+        if track and self._separator in track:
+            error_msg = "Track name cannot include '%s'." % self._separator
+            raise WidgetInputError(
+                self.name, self.label, LaunchpadValidationError(error_msg))
+        channels = [self.buildChannelName(track, risk) for risk in risks]
+        return channels
+
+    def error(self):
+        """See `IBrowserWidget`."""
+        try:
+            if self.hasInput():
+                self.getInputValue()
+        except InputErrors as error:
+            self._error = error
+        return super(StoreChannelsWidget, self).error()
+
+    def __call__(self):
+        """See `IBrowserWidget`."""
+        self.setUpSubWidgets()
+        return self.template()
 
=== added file 'lib/lp/snappy/browser/widgets/templates/storechannels.pt'
--- lib/lp/snappy/browser/widgets/templates/storechannels.pt	1970-01-01 00:00:00 +0000
+++ lib/lp/snappy/browser/widgets/templates/storechannels.pt	2017-03-30 14:10:04 +0000
@@ -0,0 +1,22 @@
+<table>
+  <tr>
+    <td>
+      <tal:widget define="widget nocall:view/track_widget">
+        <metal:block
+            use-macro="context/@@launchpad_widget_macros/launchpad_widget_row" />
+      </tal:widget>
+      <p class="formHelp">
+         To open a new track, <a href="https://snapcraft.io/community">ask the store admins</a>.
+      </p>
+    </td>
+  </tr>
+  <tr>
+    <td>
+      <tal:widget define="widget nocall:view/risks_widget"
+                  condition="widget/context/value_type/vocabulary">
+        <metal:block
+            use-macro="context/@@launchpad_widget_macros/launchpad_widget_row" />
+      </tal:widget>
+    </td>
+  </tr>
+</table>
 
=== added file 'lib/lp/snappy/browser/widgets/tests/test_storechannelswidget.py'
--- lib/lp/snappy/browser/widgets/tests/test_storechannelswidget.py	1970-01-01 00:00:00 +0000
+++ lib/lp/snappy/browser/widgets/tests/test_storechannelswidget.py	2017-03-30 14:10:04 +0000
@@ -0,0 +1,194 @@
+# Copyright 2017 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__metaclass__ = type
+
+import re
+
+from BeautifulSoup import BeautifulSoup
+from zope.formlib.interfaces import (
+    IBrowserWidget,
+    IInputWidget,
+    WidgetInputError,
+    )
+from zope.schema import List
+
+from lp.app.validators import LaunchpadValidationError
+from lp.services.webapp.escaping import html_escape
+from lp.services.webapp.servers import LaunchpadTestRequest
+from lp.snappy.browser.widgets.storechannels import StoreChannelsWidget
+from lp.snappy.interfaces.snapstoreclient import ISnapStoreClient
+from lp.snappy.vocabularies import SnapStoreChannelVocabulary
+from lp.testing import (
+    TestCaseWithFactory,
+    verifyObject,
+    )
+from lp.testing.fakemethod import FakeMethod
+from lp.testing.fixture import ZopeUtilityFixture
+from lp.testing.layers import DatabaseFunctionalLayer
+
+
+class TestStoreChannelsWidget(TestCaseWithFactory):
+
+    layer = DatabaseFunctionalLayer
+
+    def setUp(self):
+        super(TestStoreChannelsWidget, self).setUp()
+        field = List(__name__="channels", title=u"Store channels")
+        self.context = self.factory.makeSnap()
+        field = field.bind(self.context)
+        request = LaunchpadTestRequest()
+        self.widget = StoreChannelsWidget(field, None, request)
+
+        # setup fake store client response for available channels/risks
+        self.risks = [
+            {"name": "stable", "display_name": "Stable"},
+            {"name": "candidate", "display_name": "Candidate"},
+            {"name": "beta", "display_name": "Beta"},
+            {"name": "edge", "display_name": "Edge"},
+            ]
+        snap_store_client = FakeMethod()
+        snap_store_client.listChannels = FakeMethod(result=self.risks)
+        self.useFixture(
+            ZopeUtilityFixture(snap_store_client, ISnapStoreClient))
+
+    def test_implements(self):
+        self.assertTrue(verifyObject(IBrowserWidget, self.widget))
+        self.assertTrue(verifyObject(IInputWidget, self.widget))
+
+    def test_template(self):
+        self.assertTrue(
+            self.widget.template.filename.endswith("storechannels.pt"),
+            "Template was not set up.")
+
+    def test_setUpSubWidgets_first_call(self):
+        # The subwidgets are set up and a flag is set.
+        self.widget.setUpSubWidgets()
+        self.assertTrue(self.widget._widgets_set_up)
+        self.assertIsNotNone(getattr(self.widget, "track_widget", None))
+        self.assertIsInstance(
+            self.widget.risks_widget.vocabulary, SnapStoreChannelVocabulary)
+        self.assertTrue(self.widget.has_risks_vocabulary)
+
+    def test_setUpSubWidgets_second_call(self):
+        # The setUpSubWidgets method exits early if a flag is set to
+        # indicate that the widgets were set up.
+        self.widget._widgets_set_up = True
+        self.widget.setUpSubWidgets()
+        self.assertIsNone(getattr(self.widget, "track_widget", None))
+        self.assertIsNone(getattr(self.widget, "risks_widget", None))
+        self.assertIsNone(self.widget.has_risks_vocabulary)
+
+    def test_buildChannelName_no_track(self):
+        self.assertEqual("edge", self.widget.buildChannelName(None, "edge"))
+
+    def test_buildChannelName_with_track(self):
+        self.assertEqual(
+            "track/edge", self.widget.buildChannelName("track", "edge"))
+
+    def test_splitChannelName_no_track(self):
+        self.assertEqual((None, "edge"), self.widget.splitChannelName("edge"))
+
+    def test_splitChannelName_with_track(self):
+        self.assertEqual(
+            ("track", "edge"), self.widget.splitChannelName("track/edge"))
+
+    def test_splitChannelName_invalid(self):
+        self.assertRaises(
+            AssertionError, self.widget.splitChannelName, "track/edge/invalid")
+
+    def test_setRenderedValue_empty(self):
+        self.widget.setRenderedValue([])
+        self.assertIsNone(self.widget.track_widget._getCurrentValue())
+        self.assertIsNone(self.widget.risks_widget._getCurrentValue())
+
+    def test_setRenderedValue_no_track(self):
+        # Channels do not include a track
+        risks = ['candidate', 'edge']
+        self.widget.setRenderedValue(risks)
+        self.assertIsNone(self.widget.track_widget._getCurrentValue())
+        self.assertEqual(risks, self.widget.risks_widget._getCurrentValue())
+
+    def test_setRenderedValue_with_track(self):
+        # Channels including a track
+        channels = ['2.2/candidate', '2.2/edge']
+        self.widget.setRenderedValue(channels)
+        self.assertEqual('2.2', self.widget.track_widget._getCurrentValue())
+        self.assertEqual(
+            ['candidate', 'edge'], self.widget.risks_widget._getCurrentValue())
+
+    def test_setRenderedValue_invalid_value(self):
+        # Multiple channels, different tracks, unsupported
+        channels = ['2.2/candidate', '2.1/edge']
+        self.assertRaises(
+            AssertionError, self.widget.setRenderedValue, channels)
+
+    def test_hasInput_false(self):
+        # hasInput is false when there is no risk set in the form data.
+        self.widget.request = LaunchpadTestRequest(
+            form={"field.channels.track": "track"})
+        self.assertFalse(self.widget.hasInput())
+
+    def test_hasInput_true(self):
+        # hasInput is true if there are risks set in the form data.
+        self.widget.request = LaunchpadTestRequest(
+            form={"field.channels.risks": ["beta"]})
+        self.assertTrue(self.widget.hasInput())
+
+    def test_hasValidInput_false(self):
+        # The field input is invalid if any of the submitted parts are
+        # invalid.
+        form = {
+            "field.channels.track": "",
+            "field.channels.risks": ["invalid"],
+            }
+        self.widget.request = LaunchpadTestRequest(form=form)
+        self.assertFalse(self.widget.hasValidInput())
+
+    def test_hasValidInput_true(self):
+        # The field input is valid when all submitted parts are valid.
+        form = {
+            "field.channels.track": "track",
+            "field.channels.risks": ["stable", "beta"],
+            }
+        self.widget.request = LaunchpadTestRequest(form=form)
+        self.assertTrue(self.widget.hasValidInput())
+
+    def assertGetInputValueError(self, form, message):
+        self.widget.request = LaunchpadTestRequest(form=form)
+        e = self.assertRaises(WidgetInputError, self.widget.getInputValue)
+        self.assertEqual(LaunchpadValidationError(message), e.errors)
+        self.assertEqual(html_escape(message), self.widget.error())
+
+    def test_getInputValue_invalid_track(self):
+        # An error is raised when the track includes a '/'.
+        form = {"field.channels.track": "tra/ck",
+                "field.channels.risks": ["beta"]}
+        self.assertGetInputValueError(form, "Track name cannot include '/'.")
+
+    def test_getInputValue_no_track(self):
+        self.widget.request = LaunchpadTestRequest(
+            form={"field.channels.track": "",
+                  "field.channels.risks": ["beta", "edge"]})
+        expected = ["beta", "edge"]
+        self.assertEqual(expected, self.widget.getInputValue())
+
+    def test_getInputValue_with_track(self):
+        self.widget.request = LaunchpadTestRequest(
+            form={"field.channels.track": "track",
+                  "field.channels.risks": ["beta", "edge"]})
+        expected = ["track/beta", "track/edge"]
+        self.assertEqual(expected, self.widget.getInputValue())
+
+    def test_call(self):
+        # The __call__ method sets up the widgets.
+        markup = self.widget()
+        self.assertIsNotNone(self.widget.track_widget)
+        self.assertIsNotNone(self.widget.risks_widget)
+        soup = BeautifulSoup(markup)
+        fields = soup.findAll(["input"], {"id": re.compile(".*")})
+        expected_ids = [
+            "field.channels.risks.%d" % i for i in range(len(self.risks))]
+        expected_ids.append("field.channels.track")
+        ids = [field["id"] for field in fields]
+        self.assertContentEqual(expected_ids, ids)
 
=== modified file 'lib/lp/snappy/interfaces/snap.py'
--- lib/lp/snappy/interfaces/snap.py	2017-02-06 14:34:35 +0000
+++ lib/lp/snappy/interfaces/snap.py	2017-03-30 14:10:04 +0000
@@ -94,6 +94,7 @@
     ISnappyDistroSeries,
     ISnappySeries,
     )
+from lp.snappy.validators.channels import channels_validator
 from lp.soyuz.interfaces.archive import IArchive
 from lp.soyuz.interfaces.distroarchseries import IDistroArchSeries
 
@@ -537,11 +538,12 @@
             "authorize uploads of this snap package."))
 
     store_channels = exported(List(
-        value_type=Choice(vocabulary="SnapStoreChannel"),
-        title=_("Store channels"), required=False, readonly=False,
+        title=_("Store channels"),
+        required=False, readonly=False, constraint=channels_validator,
         description=_(
             "Channels to release this snap package to after uploading it to "
-            "the store.")))
+            "the store. A channel is defined by a combination of an optional "
+            " track and a risk, e.g. '2.1/stable', or 'stable'.")))
 
 
 class ISnapAdminAttributes(Interface):
 
=== modified file 'lib/lp/snappy/templates/snap-edit.pt'
--- lib/lp/snappy/templates/snap-edit.pt	2016-07-28 12:42:21 +0000
+++ lib/lp/snappy/templates/snap-edit.pt	2017-03-30 14:10:04 +0000
@@ -87,7 +87,7 @@
                   <metal:block use-macro="context/@@launchpad_form/widget_row" />
                 </tal:widget>
                 <tal:widget define="widget nocall:view/widgets/store_channels"
-                            condition="widget/context/value_type/vocabulary">
+                            condition="widget/has_risks_vocabulary">
                   <metal:block use-macro="context/@@launchpad_form/widget_row" />
                 </tal:widget>
               </table>
 
=== modified file 'lib/lp/snappy/templates/snap-new.pt'
--- lib/lp/snappy/templates/snap-new.pt	2017-02-01 06:31:30 +0000
+++ lib/lp/snappy/templates/snap-new.pt	2017-03-30 14:10:04 +0000
@@ -63,7 +63,7 @@
                   <metal:block use-macro="context/@@launchpad_form/widget_row" />
                 </tal:widget>
                 <tal:widget define="widget nocall:view/widgets/store_channels"
-                            condition="widget/context/value_type/vocabulary">
+                            condition="widget/has_risks_vocabulary">
                   <metal:block use-macro="context/@@launchpad_form/widget_row" />
                 </tal:widget>
               </table>
 
=== added directory 'lib/lp/snappy/validators'
=== added file 'lib/lp/snappy/validators/__init__.py'
=== added file 'lib/lp/snappy/validators/channels.py'
--- lib/lp/snappy/validators/channels.py	1970-01-01 00:00:00 +0000
+++ lib/lp/snappy/validators/channels.py	2017-03-30 14:10:04 +0000
@@ -0,0 +1,53 @@
+# Copyright 2017 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Validators for the .store_channels attribute."""
+
+__metaclass__ = type
+
+from lp import _
+from lp.app.validators import LaunchpadValidationError
+from lp.services.webapp.escaping import (
+    html_escape,
+    structured,
+    )
+
+
+# delimiter separating channel components
+channel_components_delimiter = '/'
+
+
+def split_channel_name(channel):
+    """Return extracted track and risk from given channel name."""
+    components = channel.split(channel_components_delimiter)
+    if len(components) == 2:
+        track, risk = components
+    elif len(components) == 1:
+        track = None
+        risk = components[0]
+    else:
+        raise ValueError("Invalid channel name: %r" % channel)
+    return track, risk
+
+
+def channels_validator(channels):
+    """Return True if the channels in a list are valid, or raise a
+    LaunchpadValidationError.
+    """
+    tracks = set()
+    for name in channels:
+        try:
+            track, risk = split_channel_name(name)
+        except ValueError:
+            message = _(
+                "Invalid channel name '${name}'. Channel names must be of the "
+                "form 'track/risk' or 'risk'.",
+                mapping={'name': html_escape(name)})
+            raise LaunchpadValidationError(structured(message))
+        tracks.add(track)
+
+    if len(tracks) != 1:
+        message = _("Channels must belong to the same track.")
+        raise LaunchpadValidationError(structured(message))
+
+    return True
 
=== added directory 'lib/lp/snappy/validators/tests'
=== added file 'lib/lp/snappy/validators/tests/__init__.py'
=== added file 'lib/lp/snappy/validators/tests/test_channels.py'
--- lib/lp/snappy/validators/tests/test_channels.py	1970-01-01 00:00:00 +0000
+++ lib/lp/snappy/validators/tests/test_channels.py	2017-03-30 14:10:04 +0000
@@ -0,0 +1,40 @@
+# Copyright 2017 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__metaclass__ = type
+
+from lp.app.validators import LaunchpadValidationError
+from lp.snappy.validators.channels import (
+    channels_validator,
+    split_channel_name,
+    )
+from lp.testing import TestCaseWithFactory
+from lp.testing.layers import LaunchpadFunctionalLayer
+
+
+class TestChannelsValidator(TestCaseWithFactory):
+
+    layer = LaunchpadFunctionalLayer
+
+    def test_split_channel_name_no_track(self):
+        self.assertEqual((None, "edge"), split_channel_name("edge"))
+
+    def test_split_channel_name_with_track(self):
+        self.assertEqual(("track", "edge"), split_channel_name("track/edge"))
+
+    def test_split_channel_name_invalid(self):
+        self.assertRaises(ValueError, split_channel_name, "track/edge/invalid")
+
+    def test_channels_validator_valid(self):
+        self.assertTrue(channels_validator(['1.1/beta', '1.1/edge']))
+        self.assertTrue(channels_validator(['beta', 'edge']))
+
+    def test_channels_validator_multiple_tracks(self):
+        self.assertRaises(
+            LaunchpadValidationError, channels_validator,
+            ['1.1/stable', '2.1/edge'])
+
+    def test_channels_validator_invalid_channel(self):
+        self.assertRaises(
+            LaunchpadValidationError, channels_validator,
+            ['1.1/stable/invalid'])