api: u.pro.status.notices.v1

Adds an api endpoint which returns list of active notices
in the machine

features/api.feature

@@ -19,6 +19,7 @@ Feature: Client behaviour for the API endpoints
     When I run `python3 -c "from uaclient.api.u.pro.services.dependencies.v1 import dependencies"` as non-root
     When I run `python3 -c "from uaclient.api.u.pro.status.enabled_services.v1 import enabled_services"` as non-root
     When I run `python3 -c "from uaclient.api.u.pro.status.is_attached.v1 import is_attached"` as non-root
+    When I run `python3 -c "from uaclient.api.u.pro.status.notices.v1 import notice_list"` as non-root
     When I run `python3 -c "from uaclient.api.u.pro.version.v1 import version"` as non-root
     When I run `python3 -c "from uaclient.api.u.security.package_manifest.v1 import package_manifest"` as non-root
     When I run `python3 -c "from uaclient.api.u.unattended_upgrades.status.v1 import status"` as non-root

features/api_notices.feature

@@ -0,0 +1,92 @@
+Feature: Status notices api
+
+  @uses.config.contract_token
+  Scenario Outline: Check notices returned by status api
+    Given a `<release>` `<machine_type>` machine with ubuntu-advantage-tools installed
+    When I create the file `/tmp/response-overlay.json` with the following:
+      """
+      {
+          "https://contracts.canonical.com/v1/context/machines/token": [
+          {
+            "code": 200,
+            "response": {
+              "machineTokenInfo": {
+                "accountInfo": {
+                  "name": "testName",
+                  "id": "testAccID"
+                },
+                "contractInfo": {
+                  "id": "testCID",
+                  "name": "testName",
+                  "resourceEntitlements": [
+                    {
+                      "type": "support",
+                      "affordances": {
+                        "onlySeries": "jammy"
+                      }
+                    }
+                  ]
+                },
+                "machineId": "testMID"
+              }
+            }
+        }],
+        "https://contracts.canonical.com/v1/contracts/testCID/context/machines/testMID": [
+          {
+            "code": 200,
+            "response": {
+              "activityToken": "test-activity-token",
+              "activityID": "test-activity-id",
+              "activityPingInterval": 123456789
+            }
+          }],
+          "https://contracts.canonical.com/v1/contracts/testCID/machine-activity/testMID": [
+          {
+            "code": 200,
+            "response": {
+              "activityToken": "test-activity-token",
+              "activityID": "test-activity-id",
+              "activityPingInterval": 123456789
+            }
+          }]
+      }
+      """
+    And I append the following on uaclient config:
+      """
+      features:
+        serviceclient_url_responses: "/tmp/response-overlay.json"
+      """
+    When I attach `contract_token` with sudo
+    Then the machine is attached
+    When I run `pro api u.pro.status.notices.v1` with sudo
+    Then API data field output matches regexp:
+      """
+      {
+        "attributes": {
+          "notices": [
+            {
+              "label": "contract_expired",
+              "message": ".*",
+              "order_id": "5"
+            },
+            {
+              "label": "limited_to_release",
+              "message": ".*",
+              "order_id": "80"
+            }
+          ]
+        },
+        "meta": {
+          "environment_vars": []
+        },
+        "type": "NoticesList"
+      }
+      """
+
+    Examples: ubuntu release
+      | release | machine_type  |
+      | xenial  | lxd-container |
+      | bionic  | lxd-container |
+      | focal   | lxd-container |
+      | jammy   | lxd-container |
+      | noble   | lxd-container |

features/subscription_attach_restrictions.feature

@@ -106,7 +106,7 @@ Feature: One time pro subscription related tests
     When I run `pro status` with sudo
     Then stdout contains substring:
       """
-      Limited to release: Ubuntu <onlyrelease> (<onlyseries_codename>)
+      #Limited to release: Ubuntu <onlyrelease> (<onlyseries_codename>)
       """
 
     Examples: ubuntu release

uaclient/api/api.py

@@ -32,6 +32,7 @@
     "u.pro.services.enable.v1",
     "u.pro.status.enabled_services.v1",
     "u.pro.status.is_attached.v1",
+    "u.pro.status.notices.v1",
     "u.apt_news.current_news.v1",
     "u.security.package_manifest.v1",
     "u.unattended_upgrades.status.v1",

uaclient/api/u/pro/status/notices/v1.py

@@ -0,0 +1,70 @@
+import logging
+from typing import List
+
+from uaclient.api.api import APIEndpoint
+from uaclient.api.data_types import AdditionalInfo
+from uaclient.config import UAConfig
+from uaclient.data_types import DataObject, Field, StringDataValue, data_list
+from uaclient.files.notices import Notice, NoticesManager
+
+LOG = logging.getLogger("ubuntupro.lib.auto_attach")
+
+
+class NoticeInfo(DataObject):
+    fields = [
+        Field("order_id", StringDataValue, doc="Notice order id"),
+        Field(
+            "message",
+            StringDataValue,
+            doc="Message to be displayed by the notice",
+        ),
+        Field("label", StringDataValue, doc="Notice label"),
+    ]
+
+    def __init__(self, order_id: str, message: str, label: str):
+        self.order_id = order_id
+        self.message = message
+        self.label = label
+
+
+class NoticeListResult(DataObject, AdditionalInfo):
+    fields = [
+        Field(
+            "notices",
+            data_list(NoticeInfo),
+            doc="A list of ``Notice`` objects",
+        ),
+    ]
+
+    def __init__(self, notices: List[NoticeInfo]):
+        self.notices = notices
+
+
+def notice_list() -> NoticeListResult:
+    return _get_notice_list(cfg=UAConfig())
+
+
+def _get_notice_list(cfg: UAConfig) -> NoticeListResult:
+    _notice_cls = NoticesManager()
+    noticeList = _notice_cls.get_active_notices()
+    notices = []
+    for notice in noticeList:
+        for n in Notice:
+            if notice.order_id == n.order_id:
+                notices.append(
+                    NoticeInfo(
+                        order_id=n.order_id,
+                        message=notice.message,
+                        label=n.label,
+                    )
+                )
+    notices.sort(key=lambda x: x.order_id)
+    return NoticeListResult(notices=notices)
+
+
+endpoint = APIEndpoint(
+    version="v1",
+    name="NoticesList",
+    fn=_get_notice_list,
+    options_cls=None,
+)

uaclient/files/notices.py

@@ -11,6 +11,7 @@
 NoticeFileDetails = namedtuple(
     "NoticeFileDetails", ["order_id", "label", "is_permanent", "message"]
 )
+ActiveNotice = namedtuple("ActiveNotice", ["order_id", "label", "message"])
 
 
 class Notice(NoticeFileDetails, Enum):
@@ -205,21 +206,19 @@ def _get_default_message(self, file_name: str) -> str:
                 return notice.value.message
         return ""
 
-    def list(self) -> List[str]:
-        """Gets all the notice files currently saved.
-
-        :returns: List of notice file contents.
-        """
+    def get_active_notices(self) -> List[ActiveNotice]:
+        """Gets the list of active notices."""
         notice_directories = (
             defaults.NOTICES_PERMANENT_DIRECTORY,
             defaults.NOTICES_TEMPORARY_DIRECTORY,
         )
-        notices = []
+        active_notices = []
         for notice_directory in notice_directories:
             if not os.path.exists(notice_directory):
                 continue
             notice_file_names = self._get_notice_file_names(notice_directory)
             for notice_file_name in notice_file_names:
+                notice_order_id, notice_label = notice_file_name.split("-")
                 try:
                     notice_file_contents = system.load_file(
                         os.path.join(notice_directory, notice_file_name)
@@ -230,12 +229,28 @@ def list(self) -> List[str]:
                     )
                     continue
                 if notice_file_contents:
-                    notices.append(notice_file_contents)
+                    message = notice_file_contents
                 else:
-                    default_message = self._get_default_message(
-                        notice_file_name
+                    message = self._get_default_message(notice_file_name)
+                active_notices.append(
+                    ActiveNotice(
+                        order_id=notice_order_id,
+                        label=notice_label,
+                        message=message,
                     )
-                    notices.append(default_message)
+                )
+        active_notices.sort(key=lambda x: x.order_id)
+        return active_notices
+
+    def list(self) -> List[str]:
+        """Gets all the notice messages currently saved.
+
+        :returns: List of notice file contents.
+        """
+        notices = []
+        active_notices = self.get_active_notices()
+        for notice in active_notices:
+            notices.append(notice.message)
         notices.sort()
         return notices
 

uaclient/files/tests/test_notices.py

@@ -155,7 +155,7 @@ def test_list(self, m_load_file, m_get_notice_file_names):
         m_get_notice_file_names.side_effect = lambda directory: (
             []
             if directory == defaults.NOTICES_TEMPORARY_DIRECTORY
-            else ["fakeNotice1"]
+            else ["1-fakeNotice"]
         )
         m_load_file.return_value = "test"