launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #28403
[Merge] ~cjwatson/launchpad:channel-strings into launchpad:master
Colin Watson has proposed merging ~cjwatson/launchpad:channel-strings into launchpad:master with ~cjwatson/launchpad:hardcode-store-risks as a prerequisite.
Commit message:
Move channel (de)composition to its own module
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/+git/launchpad/+merge/420743
Rather than having conversion between composed channel names and decomposed (track, risk, branch) tuples be spread around between `lp.snappy.validators.channels` and `lp.snappy.browser.widgets.storechannels`, let's have a simple common `lp.services.channels` module that just does the string manipulation. I chose the function names based on those in lp:snapdevicegw, although the implementations diverge somewhat at the moment; it may be worth trying to bring them into sync as a future step.
--
Your team Launchpad code reviewers is requested to review the proposed merge of ~cjwatson/launchpad:channel-strings into launchpad:master.
diff --git a/lib/lp/services/channels.py b/lib/lp/services/channels.py
new file mode 100644
index 0000000..79d4e75
--- /dev/null
+++ b/lib/lp/services/channels.py
@@ -0,0 +1,56 @@
+# Copyright 2022 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Utilities for dealing with channels in Canonical's stores."""
+
+__all__ = [
+ "channel_components_delimiter",
+ "channel_list_to_string",
+ "channel_string_to_list",
+ ]
+
+from lp.registry.enums import StoreRisk
+
+
+# delimiter separating channel components
+channel_components_delimiter = '/'
+
+
+def _is_risk(component):
+ """Does this channel component identify a risk?"""
+ return component in {item.title for item in StoreRisk.items}
+
+
+def channel_string_to_list(channel):
+ """Return extracted track, risk, and branch from given channel name."""
+ components = channel.split(channel_components_delimiter)
+ if len(components) == 3:
+ track, risk, branch = components
+ elif len(components) == 2:
+ # Identify risk to determine if this is track/risk or risk/branch.
+ if _is_risk(components[0]):
+ if _is_risk(components[1]):
+ raise ValueError(
+ "Branch name cannot match a risk name: %r" % channel)
+ track = None
+ risk, branch = components
+ elif _is_risk(components[1]):
+ track, risk = components
+ branch = None
+ else:
+ raise ValueError("No valid risk provided: %r" % channel)
+ elif len(components) == 1:
+ track = None
+ risk = components[0]
+ branch = None
+ else:
+ raise ValueError("Invalid channel name: %r" % channel)
+ return track, risk, branch
+
+
+def channel_list_to_string(track, risk, branch):
+ """Return channel name composed from given track, risk, and branch."""
+ if track == "latest":
+ track = None
+ return channel_components_delimiter.join(
+ [c for c in (track, risk, branch) if c is not None])
diff --git a/lib/lp/services/tests/test_channels.py b/lib/lp/services/tests/test_channels.py
new file mode 100644
index 0000000..a8de373
--- /dev/null
+++ b/lib/lp/services/tests/test_channels.py
@@ -0,0 +1,63 @@
+# Copyright 2022 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for handling of channels in Canonical's stores."""
+
+from lp.services.channels import (
+ channel_list_to_string,
+ channel_string_to_list,
+ )
+from lp.testing import TestCase
+from lp.testing.layers import BaseLayer
+
+
+class TestChannels(TestCase):
+
+ layer = BaseLayer
+
+ def test_channel_string_to_list_no_track_or_branch(self):
+ self.assertEqual((None, "edge", None), channel_string_to_list("edge"))
+
+ def test_channel_string_to_list_with_track(self):
+ self.assertEqual(
+ ("track", "edge", None), channel_string_to_list("track/edge"))
+
+ def test_channel_string_to_list_with_branch(self):
+ self.assertEqual(
+ (None, "edge", "fix-123"), channel_string_to_list("edge/fix-123"))
+
+ def test_channel_string_to_list_with_track_and_branch(self):
+ self.assertEqual(
+ ("track", "edge", "fix-123"),
+ channel_string_to_list("track/edge/fix-123"))
+
+ def test_channel_string_to_list_no_risk(self):
+ self.assertRaisesWithContent(
+ ValueError, "No valid risk provided: 'track/fix-123'",
+ channel_string_to_list, "track/fix-123")
+
+ def test_channel_string_to_list_ambiguous_risk(self):
+ self.assertRaisesWithContent(
+ ValueError, "Branch name cannot match a risk name: 'edge/stable'",
+ channel_string_to_list, "edge/stable")
+
+ def test_channel_string_to_list_too_many_components(self):
+ self.assertRaisesWithContent(
+ ValueError, "Invalid channel name: 'track/edge/invalid/too-long'",
+ channel_string_to_list, "track/edge/invalid/too-long")
+
+ def test_channel_list_to_string_no_track_or_branch(self):
+ self.assertEqual("edge", channel_list_to_string(None, "edge", None))
+
+ def test_channel_list_to_string_with_track(self):
+ self.assertEqual(
+ "track/edge", channel_list_to_string("track", "edge", None))
+
+ def test_channel_list_to_string_with_branch(self):
+ self.assertEqual(
+ "edge/fix-123", channel_list_to_string(None, "edge", "fix-123"))
+
+ def test_channel_list_to_string_with_track_and_branch(self):
+ self.assertEqual(
+ "track/edge/fix-123",
+ channel_list_to_string("track", "edge", "fix-123"))
diff --git a/lib/lp/snappy/browser/widgets/storechannels.py b/lib/lp/snappy/browser/widgets/storechannels.py
index 7b0badf..d855669 100644
--- a/lib/lp/snappy/browser/widgets/storechannels.py
+++ b/lib/lp/snappy/browser/widgets/storechannels.py
@@ -28,21 +28,21 @@ 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.channels import (
+ channel_components_delimiter,
+ channel_list_to_string,
+ channel_string_to_list,
+ )
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")
- _separator = channel_components_delimiter
_default_track = 'latest'
_widgets_set_up = False
@@ -86,23 +86,6 @@ class StoreChannelsWidget(BrowserWidget, InputWidget):
risks_widget = getattr(self, 'risks_widget', None)
return risks_widget and bool(risks_widget.vocabulary)
- def buildChannelName(self, track, risk, branch):
- """Return channel name composed from given track, risk, and branch."""
- channel = risk
- if track and track != self._default_track:
- channel = self._separator.join((track, channel))
- if branch:
- channel = self._separator.join((channel, branch))
- return channel
-
- def splitChannelName(self, channel):
- """Return extracted track, risk, and branch from given channel name."""
- try:
- track, risk, branch = split_channel_name(channel)
- except ValueError:
- raise AssertionError("Not a valid value: %r" % channel)
- return track, risk, branch
-
def setRenderedValue(self, value):
"""See `IWidget`."""
self.setUpSubWidgets()
@@ -113,7 +96,10 @@ class StoreChannelsWidget(BrowserWidget, InputWidget):
branches = set()
risks = []
for channel in value:
- track, risk, branch = self.splitChannelName(channel)
+ try:
+ track, risk, branch = channel_string_to_list(channel)
+ except ValueError:
+ raise AssertionError("Not a valid value: %r" % channel)
tracks.add(track)
risks.append(risk)
branches.add(branch)
@@ -149,16 +135,18 @@ class StoreChannelsWidget(BrowserWidget, InputWidget):
track = self.track_widget.getInputValue()
risks = self.risks_widget.getInputValue()
branch = self.branch_widget.getInputValue()
- if track and self._separator in track:
- error_msg = "Track name cannot include '%s'." % self._separator
+ if track and channel_components_delimiter in track:
+ error_msg = "Track name cannot include '%s'." % (
+ channel_components_delimiter)
raise WidgetInputError(
self.name, self.label, LaunchpadValidationError(error_msg))
- if branch and self._separator in branch:
- error_msg = "Branch name cannot include '%s'." % self._separator
+ if branch and channel_components_delimiter in branch:
+ error_msg = "Branch name cannot include '%s'." % (
+ channel_components_delimiter)
raise WidgetInputError(
self.name, self.label, LaunchpadValidationError(error_msg))
channels = [
- self.buildChannelName(track, risk, branch) for risk in risks]
+ channel_list_to_string(track, risk, branch) for risk in risks]
return channels
def error(self):
diff --git a/lib/lp/snappy/browser/widgets/tests/test_storechannelswidget.py b/lib/lp/snappy/browser/widgets/tests/test_storechannelswidget.py
index 78d6a48..b776347 100644
--- a/lib/lp/snappy/browser/widgets/tests/test_storechannelswidget.py
+++ b/lib/lp/snappy/browser/widgets/tests/test_storechannelswidget.py
@@ -65,48 +65,6 @@ class TestStoreChannelsWidget(TestCaseWithFactory):
self.assertIsNone(getattr(self.widget, "branch_widget", None))
self.assertIsNone(self.widget.has_risks_vocabulary)
- def test_buildChannelName_no_track_or_branch(self):
- self.assertEqual(
- "edge", self.widget.buildChannelName(None, "edge", None))
-
- def test_buildChannelName_with_track(self):
- self.assertEqual(
- "track/edge", self.widget.buildChannelName("track", "edge", None))
-
- def test_buildChannelName_with_branch(self):
- self.assertEqual(
- "edge/fix-123",
- self.widget.buildChannelName(None, "edge", "fix-123"))
-
- def test_buildChannelName_with_track_and_branch(self):
- self.assertEqual(
- "track/edge/fix-123",
- self.widget.buildChannelName("track", "edge", "fix-123"))
-
- def test_splitChannelName_no_track_or_branch(self):
- self.assertEqual(
- (None, "edge", None), self.widget.splitChannelName("edge"))
-
- def test_splitChannelName_with_track(self):
- self.assertEqual(
- ("track", "edge", None),
- self.widget.splitChannelName("track/edge"))
-
- def test_splitChannelName_with_branch(self):
- self.assertEqual(
- (None, "edge", "fix-123"),
- self.widget.splitChannelName("edge/fix-123"))
-
- def test_splitChannelName_with_track_and_branch(self):
- self.assertEqual(
- ("track", "edge", "fix-123"),
- self.widget.splitChannelName("track/edge/fix-123"))
-
- def test_splitChannelName_invalid(self):
- self.assertRaises(
- AssertionError, self.widget.splitChannelName,
- "track/edge/invalid/too-long")
-
def test_setRenderedValue_empty(self):
self.widget.setRenderedValue([])
self.assertIsNone(self.widget.track_widget._getCurrentValue())
diff --git a/lib/lp/snappy/validators/channels.py b/lib/lp/snappy/validators/channels.py
index 77a210b..28a1fc0 100644
--- a/lib/lp/snappy/validators/channels.py
+++ b/lib/lp/snappy/validators/channels.py
@@ -5,49 +5,13 @@
from lp import _
from lp.app.validators import LaunchpadValidationError
-from lp.registry.enums import StoreRisk
+from lp.services.channels import channel_string_to_list
from lp.services.webapp.escaping import (
html_escape,
structured,
)
-# delimiter separating channel components
-channel_components_delimiter = '/'
-
-
-def _is_risk(component):
- """Does this channel component identify a risk?"""
- return component in {item.title for item in StoreRisk.items}
-
-
-def split_channel_name(channel):
- """Return extracted track, risk, and branch from given channel name."""
- components = channel.split(channel_components_delimiter)
- if len(components) == 3:
- track, risk, branch = components
- elif len(components) == 2:
- # Identify risk to determine if this is track/risk or risk/branch.
- if _is_risk(components[0]):
- if _is_risk(components[1]):
- raise ValueError(
- "Branch name cannot match a risk name: %r" % channel)
- track = None
- risk, branch = components
- elif _is_risk(components[1]):
- track, risk = components
- branch = None
- else:
- raise ValueError("No valid risk provided: %r" % channel)
- elif len(components) == 1:
- track = None
- risk = components[0]
- branch = None
- else:
- raise ValueError("Invalid channel name: %r" % channel)
- return track, risk, branch
-
-
def channels_validator(channels):
"""Return True if the channels in a list are valid, or raise a
LaunchpadValidationError.
@@ -56,7 +20,7 @@ def channels_validator(channels):
branches = set()
for name in channels:
try:
- track, risk, branch = split_channel_name(name)
+ track, risk, branch = channel_string_to_list(name)
except ValueError:
message = _(
"Invalid channel name '${name}'. Channel names must be of the "
diff --git a/lib/lp/snappy/validators/tests/test_channels.py b/lib/lp/snappy/validators/tests/test_channels.py
index 11a3d9b..6e3cda8 100644
--- a/lib/lp/snappy/validators/tests/test_channels.py
+++ b/lib/lp/snappy/validators/tests/test_channels.py
@@ -2,10 +2,7 @@
# GNU Affero General Public License version 3 (see the file LICENSE).
from lp.app.validators import LaunchpadValidationError
-from lp.snappy.validators.channels import (
- channels_validator,
- split_channel_name,
- )
+from lp.snappy.validators.channels import channels_validator
from lp.testing import TestCaseWithFactory
from lp.testing.layers import LaunchpadFunctionalLayer
@@ -14,32 +11,6 @@ class TestChannelsValidator(TestCaseWithFactory):
layer = LaunchpadFunctionalLayer
- def test_split_channel_name_no_track_or_branch(self):
- self.assertEqual((None, "edge", None), split_channel_name("edge"))
-
- def test_split_channel_name_with_track(self):
- self.assertEqual(
- ("track", "edge", None), split_channel_name("track/edge"))
-
- def test_split_channel_name_with_branch(self):
- self.assertEqual(
- (None, "edge", "fix-123"), split_channel_name("edge/fix-123"))
-
- def test_split_channel_name_with_track_and_branch(self):
- self.assertEqual(
- ("track", "edge", "fix-123"),
- split_channel_name("track/edge/fix-123"))
-
- def test_split_channel_name_no_risk(self):
- self.assertRaises(ValueError, split_channel_name, "track/fix-123")
-
- def test_split_channel_name_ambiguous_risk(self):
- self.assertRaises(ValueError, split_channel_name, "edge/stable")
-
- def test_split_channel_name_too_many_components(self):
- self.assertRaises(
- ValueError, split_channel_name, "track/edge/invalid/too-long")
-
def test_channels_validator_valid(self):
self.assertTrue(
channels_validator(['1.1/beta/fix-123', '1.1/edge/fix-123']))
