Don't list UA services that are not entitled

Instead of only checking if a given UA service is available, we now also
check if it is entitled.

 - the available field for a service refers to its availability on the
   current machine (e.g. on Focal running on a amd64 CPU) ; whereas
 - the entitled field tells us if the contract covers the service.

Therefore, we need to make sure that we only list the services that are
both "available" and "entitled".

Signed-off-by: Olivier Gayot <olivier.gayot@canonical.com>
This commit is contained in:
Olivier Gayot 2022-01-25 18:46:48 +01:00
parent 229e364956
commit 7dfe722686
5 changed files with 31 additions and 24 deletions

View File

@ -6,7 +6,7 @@
{ {
"name": "cis", "name": "cis",
"description": "Center for Internet Security Audit Tools", "description": "Center for Internet Security Audit Tools",
"entitled": "yes", "entitled": "no",
"auto_enabled": "no", "auto_enabled": "no",
"available": "yes" "available": "yes"
}, },
@ -15,7 +15,7 @@
"description": "UA Apps: Extended Security Maintenance (ESM)", "description": "UA Apps: Extended Security Maintenance (ESM)",
"entitled": "yes", "entitled": "yes",
"auto_enabled": "yes", "auto_enabled": "yes",
"available": "yes" "available": "no"
}, },
{ {
"name": "esm-infra", "name": "esm-infra",

View File

@ -78,7 +78,8 @@ class UbuntuAdvantageController(SubiquityTuiController):
""" Asynchronously check the token passed as an argument. """ """ Asynchronously check the token passed as an argument. """
async def inner() -> None: async def inner() -> None:
try: try:
svcs = await self.ua_interface.get_avail_services(token=token) svcs = await \
self.ua_interface.get_activable_services(token=token)
except InvalidUATokenError: except InvalidUATokenError:
if isinstance(self.ui.body, UbuntuAdvantageView): if isinstance(self.ui.body, UbuntuAdvantageView):
self.ui.body.show_invalid_token() self.ui.body.show_invalid_token()
@ -90,7 +91,7 @@ class UbuntuAdvantageController(SubiquityTuiController):
self.ui.body.show_unknown_error() self.ui.body.show_unknown_error()
else: else:
if isinstance(self.ui.body, UbuntuAdvantageView): if isinstance(self.ui.body, UbuntuAdvantageView):
self.ui.body.show_available_services(svcs) self.ui.body.show_activable_services(svcs)
self._check_task = schedule_task(inner()) self._check_task = schedule_task(inner())

View File

@ -125,24 +125,26 @@ class TestUAClientUAInterfaceStrategy(unittest.TestCase):
class TestUAInterface(unittest.TestCase): class TestUAInterface(unittest.TestCase):
def test_mocked_get_avail_services(self): def test_mocked_get_activable_services(self):
strategy = MockedUAInterfaceStrategy(scale_factor=1_000_000) strategy = MockedUAInterfaceStrategy(scale_factor=1_000_000)
interface = UAInterface(strategy) interface = UAInterface(strategy)
with self.assertRaises(InvalidUATokenError): with self.assertRaises(InvalidUATokenError):
run_coro(interface.get_avail_services(token="invalidToken")) run_coro(interface.get_activable_services(token="invalidToken"))
# Tokens starting with "f" in dry-run mode simulate an "internal" # Tokens starting with "f" in dry-run mode simulate an "internal"
# error. # error.
with self.assertRaises(CheckSubscriptionError): with self.assertRaises(CheckSubscriptionError):
run_coro(interface.get_avail_services(token="failure")) run_coro(interface.get_activable_services(token="failure"))
# Tokens starting with "x" is dry-run mode simulate an expired token. # Tokens starting with "x" is dry-run mode simulate an expired token.
with self.assertRaises(ExpiredUATokenError): with self.assertRaises(ExpiredUATokenError):
run_coro(interface.get_avail_services(token="xpiredToken")) run_coro(interface.get_activable_services(token="xpiredToken"))
# Other tokens are considered valid in dry-run mode. # Other tokens are considered valid in dry-run mode.
services = run_coro(interface.get_avail_services(token="validToken")) services = run_coro(
interface.get_activable_services(token="validToken"))
for service in services: for service in services:
self.assertIn("name", service) self.assertIn("name", service)
self.assertIn("description", service) self.assertIn("description", service)
self.assertTrue(service["available"]) self.assertEqual(service["available"], "yes")
self.assertEqual(service["entitled"], "yes")

View File

@ -147,9 +147,9 @@ class UAInterface:
""" Return a dictionary containing the subscription information. """ """ Return a dictionary containing the subscription information. """
return await self.strategy.query_info(token) return await self.strategy.query_info(token)
async def get_avail_services(self, token: str) -> List[dict]: async def get_activable_services(self, token: str) -> List[dict]:
""" Return a list of available services for the subscription """ Return a list of activable services (i.e. services that are
associated with the token provided. entitled to the subscription and available on the current hardware).
""" """
info = await self.get_subscription(token) info = await self.get_subscription(token)
@ -157,8 +157,12 @@ class UAInterface:
if expiration.timestamp() <= dt.utcnow().timestamp(): if expiration.timestamp() <= dt.utcnow().timestamp():
raise ExpiredUATokenError(token, expires=info["expires"]) raise ExpiredUATokenError(token, expires=info["expires"])
def is_avail_service(service: dict) -> bool: def is_activable_service(service: dict) -> bool:
# TODO do we need to check for service["entitled"] as well? # - the available field for a service refers to its availability on
return service["available"] == "yes" # the current machine (e.g. on Focal running on a amd64 CPU) ;
# whereas
# - the entitled field tells us if the contract covers the service.
return service["available"] == "yes" \
and service["entitled"] == "yes"
return [svc for svc in info["services"] if is_avail_service(svc)] return [svc for svc in info["services"] if is_activable_service(svc)]

View File

@ -189,16 +189,16 @@ class UbuntuAdvantageView(BaseView):
self.remove_overlay() self.remove_overlay()
self.show_stretchy_overlay(ContinueAnywayWidget(self)) self.show_stretchy_overlay(ContinueAnywayWidget(self))
def show_available_services(self, services: List[dict]) -> None: def show_activable_services(self, services: List[dict]) -> None:
""" Display an overlay with the list of services that will be enabled """ Display an overlay with the list of services that can be enabled
via Ubuntu Advantage subscription. After the user confirms, the next we via Ubuntu Advantage subscription. After the user confirms, we will
will quit the current view and move on. """ quit the current view and move on. """
self.remove_overlay() self.remove_overlay()
self.show_stretchy_overlay(ShowServicesWidget(self, services)) self.show_stretchy_overlay(ShowServicesWidget(self, services))
class ShowServicesWidget(Stretchy): class ShowServicesWidget(Stretchy):
""" Widget to show the available services for UA subscription. """ """ Widget to show the activable services for UA subscription. """
def __init__(self, parent: UbuntuAdvantageView, def __init__(self, parent: UbuntuAdvantageView,
services: List[dict]) -> None: services: List[dict]) -> None:
""" Initializes the widget by including the list of services as a """ Initializes the widget by including the list of services as a
@ -207,8 +207,8 @@ class ShowServicesWidget(Stretchy):
ok = ok_btn(label=_("OK"), on_press=self.ok) ok = ok_btn(label=_("OK"), on_press=self.ok)
title = _("Available Services") title = _("Activable Services")
header = _("List of services that are available through your " header = _("List of services that are activable through your "
"Ubuntu Advantage subscription:") "Ubuntu Advantage subscription:")
widgets: List[Widget] = [ widgets: List[Widget] = [