Filter unreadable notices when listing notices

Added behave and unit tests

Fixes: #2898

features/unattached_status.feature

@@ -599,3 +599,26 @@ Feature: Unattached status
         Examples: ubuntu release
            | release | machine_type  |
            | jammy   | lxd-container |
+
+    Scenario Outline: Check notice file read permission
+        Given a `<release>` `<machine_type>` machine with ubuntu-advantage-tools installed
+        When I run `mkdir -p /run/ubuntu-advantage/notices` with sudo
+        When I run `touch /run/ubuntu-advantage/notices/crasher` with sudo
+        When I run `chmod 0 /run/ubuntu-advantage/notices/crasher` with sudo
+        When I run `mkdir -p /var/lib/ubuntu-advantage/notices` with sudo
+        When I run `touch /var/lib/ubuntu-advantage/notices/crasher` with sudo
+        When I run `chmod 0 /var/lib/ubuntu-advantage/notices/crasher` with sudo
+        When I run `touch /run/ubuntu-advantage/notices/10-reboot_required` with sudo
+        When I run `pro status` as non-root
+        Then stdout matches regexp:
+        """
+        NOTICES
+        System reboot required
+        """
+        Examples: ubuntu release
+          | release | machine_type  |
+          | xenial  | lxd-container |
+          | bionic  | lxd-container |
+          | focal   | lxd-container |
+          | jammy   | lxd-container |
+          | mantic  | lxd-container |

uaclient/cli/tests/test_cli_status.py

@@ -372,8 +372,8 @@ def test_status_help(
             ([], ""),
             (
                 [
-                    [FakeNotice.a, "adesc"],
-                    [FakeNotice.b, "bdesc"],
+                    [FakeNotice.reboot_required, "adesc"],
+                    [FakeNotice.enable_reboot_required, "bdesc"],
                 ],
                 "\nNOTICES\nadesc\nbdesc\n",
             ),

uaclient/conftest.py

@@ -285,9 +285,15 @@ def event():
 
 
 class FakeNotice(NoticeFileDetails, Enum):
-    a = NoticeFileDetails("01", "a", True, "notice_a")
-    a2 = NoticeFileDetails("03", "a2", True, "notice_a2")
-    b = NoticeFileDetails("02", "b2", False, "notice_b")
+    reboot_required = NoticeFileDetails(
+        "10", "reboot_required", False, "notice_a"
+    )
+    reboot_script_failed = NoticeFileDetails(
+        "12", "reboot_script_failed", True, "notice_a2"
+    )
+    enable_reboot_required = NoticeFileDetails(
+        "11", "enable_reboot_required", False, "notice_b"
+    )
 
 
 @pytest.yield_fixture(autouse=True)

uaclient/files/notices.py

@@ -160,6 +160,45 @@ def remove(self, notice_details: Notice):
         )
         system.ensure_file_absent(os.path.join(directory, filename))
 
+    def _get_notice_file_names(self, directory: str) -> List[str]:
+        """Gets the list of notice file names in the given directory.
+
+        :param directory: The directory to search for notice files.
+        :returns: List of notice file names.
+        """
+        return [
+            file_name
+            for file_name in os.listdir(directory)
+            if os.path.isfile(os.path.join(directory, file_name))
+            and self._is_valid_notice_file(directory, file_name)
+        ]
+
+    def _is_valid_notice_file(self, directory: str, file_name: str) -> bool:
+        """Checks if the notice file is valid.
+
+        :param file_name: The name of the notice file.
+        :returns: True if the file is valid, False otherwise.
+        """
+        is_permanent_dir = directory == defaults.NOTICES_PERMANENT_DIRECTORY
+        valid_file_names = {
+            "{}-{}".format(n.order_id, n.label)
+            for n in Notice
+            if n.is_permanent == is_permanent_dir
+        }
+        return file_name in valid_file_names
+
+    def _get_default_message(self, file_name: str) -> str:
+        """Gets the default message for a notice file.
+
+        :param file_name: The name of the notice file.
+        :returns: The default message defined in the enum.
+        """
+        order_id, label = file_name.split("-")
+        for notice in Notice:
+            if notice.order_id == order_id and notice.label == label:
+                return notice.value.message
+        return ""
+
     def list(self) -> List[str]:
         """Gets all the notice files currently saved.
 
@@ -173,35 +212,24 @@ def list(self) -> List[str]:
         for notice_directory in notice_directories:
             if not os.path.exists(notice_directory):
                 continue
-            notice_file_names = [
-                file_name
-                for file_name in os.listdir(notice_directory)
-                if os.path.isfile(os.path.join(notice_directory, file_name))
-            ]
+            notice_file_names = self._get_notice_file_names(notice_directory)
             for notice_file_name in notice_file_names:
-                notice_file_contents = system.load_file(
-                    os.path.join(notice_directory, notice_file_name)
-                )
+                try:
+                    notice_file_contents = system.load_file(
+                        os.path.join(notice_directory, notice_file_name)
+                    )
+                except PermissionError:
+                    LOG.warning(
+                        "Permission error while reading " + notice_file_name
+                    )
+                    continue
                 if notice_file_contents:
                     notices.append(notice_file_contents)
                 else:
-                    # if no contents of file, default to message
-                    # defined in the enum
-                    try:
-                        order_id, label = notice_file_name.split("-")
-                        notice = None
-                        for n in Notice:
-                            if n.order_id == order_id and n.label == label:
-                                notice = n
-                        if notice is None:
-                            raise Exception()
-                        notices.append(notice.value.message)
-                    except Exception:
-                        LOG.warning(
-                            "Something went wrong while processing"
-                            " notice: %s.",
-                            notice_file_name,
-                        )
+                    default_message = self._get_default_message(
+                        notice_file_name
+                    )
+                    notices.append(default_message)
         notices.sort()
         return notices
 

uaclient/files/tests/test_notices.py

@@ -14,8 +14,8 @@ class TestNotices:
         "label,content",
         (
             (
-                FakeNotice.a,
-                "notice_a",
+                FakeNotice.reboot_script_failed,
+                "notice_a2",
             ),
         ),
     )
@@ -30,7 +30,10 @@ def test_add(
         notice.add(label, content)
         assert [
             mock.call(
-                os.path.join(defaults.NOTICES_PERMANENT_DIRECTORY, "01-a"),
+                os.path.join(
+                    defaults.NOTICES_PERMANENT_DIRECTORY,
+                    "12-reboot_script_failed",
+                ),
                 content,
             )
         ] == sys_write_file.call_args_list
@@ -44,15 +47,18 @@ def test_add_non_root(
         caplog_text,
     ):
         notice = NoticesManager()
-        notice.add(FakeNotice.a, "content")
+        notice.add(FakeNotice.reboot_required, "content")
         assert [] == m_sys_write_file.call_args_list
-        assert "NoticesManager.add(a) called as non-root user" in caplog_text()
+        assert (
+            "NoticesManager.add(reboot_required) called as non-root user"
+            in caplog_text()
+        )
 
     @pytest.mark.parametrize(
         "label,content",
         (
             (
-                FakeNotice.a,
+                FakeNotice.reboot_required,
                 "notice_a",
             ),
         ),
@@ -74,8 +80,8 @@ def test_add_duplicate_label(
         "label,content",
         (
             (
-                FakeNotice.a,
-                "notice_a",
+                FakeNotice.reboot_script_failed,
+                "notice_a2",
             ),
         ),
     )
@@ -91,7 +97,10 @@ def test_remove(
         notice.remove(label)
         assert [
             mock.call(
-                os.path.join(defaults.NOTICES_PERMANENT_DIRECTORY, "01-a"),
+                os.path.join(
+                    defaults.NOTICES_PERMANENT_DIRECTORY,
+                    "12-reboot_script_failed",
+                ),
             )
         ] == sys_file_absent.call_args_list
 
@@ -104,10 +113,11 @@ def test_remove_non_root(
         caplog_text,
     ):
         notice = NoticesManager()
-        notice.remove(FakeNotice.a)
+        notice.remove(FakeNotice.reboot_required)
         assert [] == m_sys_file_absent.call_args_list
         assert (
-            "NoticesManager.remove(a) called as non-root user" in caplog_text()
+            "NoticesManager.remove(reboot_required) called as non-root user"
+            in caplog_text()
         )
 
     @mock.patch("uaclient.files.notices.NoticesManager.list")
@@ -116,11 +126,37 @@ def test_remove_non_root(
     def test_notice_module(
         self, notice_cls_add, notice_cls_remove, notice_cls_read
     ):
-        notices.add(FakeNotice.a)
+        notices.add(FakeNotice.reboot_required)
         assert [
-            mock.call(FakeNotice.a, "notice_a"),
+            mock.call(FakeNotice.reboot_required, "notice_a"),
         ] == notice_cls_add.call_args_list
-        notices.remove(FakeNotice.a)
-        assert [mock.call(FakeNotice.a)] == notice_cls_remove.call_args_list
+        notices.remove(FakeNotice.reboot_required)
+        assert [
+            mock.call(FakeNotice.reboot_required)
+        ] == notice_cls_remove.call_args_list
         notices.list()
         assert 1 == notice_cls_read.call_count
+
+    @mock.patch("uaclient.files.notices.NoticesManager._get_notice_file_names")
+    def test_get_notice_file_names(self, m_get_notice_file_names):
+        notice = NoticesManager()
+        m_get_notice_file_names.return_value = []
+        assert [] == notice._get_notice_file_names("directory")
+        m_get_notice_file_names.return_value = ["file1", "file2", "file3"]
+        assert ["file1", "file2", "file3"] == notice._get_notice_file_names(
+            "directory"
+        )
+
+    @mock.patch("uaclient.files.notices.NoticesManager._get_notice_file_names")
+    @mock.patch("uaclient.system.load_file")
+    def test_list(self, m_load_file, m_get_notice_file_names):
+        notice = NoticesManager()
+
+        m_get_notice_file_names.side_effect = (
+            lambda directory: []
+            if directory == defaults.NOTICES_TEMPORARY_DIRECTORY
+            else ["fakeNotice1"]
+        )
+        m_load_file.return_value = "test"
+
+        assert ["test"] == notice.list()

uaclient/tests/test_config.py

@@ -74,13 +74,13 @@ class TestNotices:
         (
             ([], ()),
             (
-                [[FakeNotice.a2, "a1"]],
+                [[FakeNotice.reboot_script_failed, "a1"]],
                 ["a1"],
             ),
             (
                 [
-                    [FakeNotice.a, "a1"],
-                    [FakeNotice.a2, "a2"],
+                    [FakeNotice.reboot_required, "a1"],
+                    [FakeNotice.reboot_script_failed, "a2"],
                 ],
                 [
                     "a1",
@@ -89,8 +89,8 @@ class TestNotices:
             ),
             (
                 [
-                    [FakeNotice.a, "a1"],
-                    [FakeNotice.a, "a1"],
+                    [FakeNotice.reboot_required, "a1"],
+                    [FakeNotice.reboot_required, "a1"],
                 ],
                 [
                     "a1",
@@ -116,11 +116,11 @@ def test_add_notice_avoids_duplicates(
         "_notices",
         (
             ([]),
-            ([[FakeNotice.a]]),
+            ([[FakeNotice.reboot_required]]),
             (
                 [
-                    [FakeNotice.a],
-                    [FakeNotice.a2],
+                    [FakeNotice.reboot_required],
+                    [FakeNotice.reboot_script_failed],
                 ]
             ),
         ),
@@ -139,29 +139,29 @@ def test_add_notice_fails_as_nonroot(
     @pytest.mark.parametrize(
         "notices_,removes,expected",
         (
-            ([], [FakeNotice.a], []),
+            ([], [FakeNotice.reboot_required], []),
             (
-                [[FakeNotice.a2]],
-                [FakeNotice.a2],
+                [[FakeNotice.reboot_script_failed]],
+                [FakeNotice.reboot_script_failed],
                 [],
             ),
             (
                 [
-                    [FakeNotice.a],
-                    [FakeNotice.a2],
+                    [FakeNotice.reboot_required],
+                    [FakeNotice.reboot_script_failed],
                 ],
-                [FakeNotice.a],
+                [FakeNotice.reboot_required],
                 ["notice_a2"],
             ),
             (
                 [
-                    [FakeNotice.a],
-                    [FakeNotice.a2],
-                    [FakeNotice.b],
+                    [FakeNotice.reboot_required],
+                    [FakeNotice.reboot_script_failed],
+                    [FakeNotice.enable_reboot_required],
                 ],
                 [
-                    FakeNotice.a,
-                    FakeNotice.a2,
+                    FakeNotice.reboot_required,
+                    FakeNotice.reboot_script_failed,
                 ],
                 ["notice_b"],
             ),