Apt pinned packages

features/schemas/ua_security_status.json

@@ -93,6 +93,7 @@
             "type": "string",
             "enum": [
               "upgrade_available",
+              "upgrade_available_not_preferred",
               "pending_attach",
               "pending_enable",
               "upgrade_unavailable"

uaclient/security_status.py

@@ -17,6 +17,7 @@
     get_apt_cache_datetime,
     get_apt_pkg_cache,
     get_esm_apt_pkg_cache,
+    get_pkg_candidate_version,
 )
 from uaclient.config import UAConfig
 from uaclient.entitlements import ESMAppsEntitlement, ESMInfraEntitlement
@@ -39,6 +40,7 @@
 class UpdateStatus(Enum):
     "Represents the availability of a security package."
     AVAILABLE = "upgrade_available"
+    AVAILABLE_NOT_PREFERRED = "upgrade_available_not_preferred"
     UNATTACHED = "pending_attach"
     NOT_ENABLED = "pending_enable"
     UNAVAILABLE = "upgrade_unavailable"
@@ -118,7 +120,20 @@ def get_origin_for_installed_package(
     return "third-party"
 
 
-def get_update_status(service_name: str, ua_info: Dict[str, Any]) -> str:
+@lru_cache(maxsize=None)
+def _is_candidate_version(pkg: str, version: str) -> bool:
+    """Returns True if the package version is a candidate version."""
+    candidate_version = get_pkg_candidate_version(pkg, check_esm_cache=False)
+    if candidate_version:
+        return version == candidate_version
+    return False
+
+
+def get_update_status(
+    service_name: str,
+    ua_info: Dict[str, Any],
+    version: apt_pkg.Version,
+) -> str:
     """Defines the update status for a package based on the service name.
 
     For ESM-[Infra|Apps] packages, first checks if Pro is attached. If this is
@@ -127,7 +142,14 @@ def get_update_status(service_name: str, ua_info: Dict[str, Any]) -> str:
     if service_name in ("standard-security", "standard-updates") or (
         ua_info["attached"] and service_name in ua_info["enabled_services"]
     ):
-        return UpdateStatus.AVAILABLE.value
+        is_candidate = _is_candidate_version(
+            version.parent_pkg.name, version.ver_str
+        )
+        return (
+            UpdateStatus.AVAILABLE.value
+            if is_candidate
+            else UpdateStatus.AVAILABLE_NOT_PREFERRED.value
+        )
     if not ua_info["attached"]:
         return UpdateStatus.UNATTACHED.value
     if service_name in ua_info["entitled_services"]:
@@ -262,8 +284,8 @@ def create_updates_list(
 ) -> List[Dict[str, Any]]:
     updates = []
     for service, version_list in upgradable_versions.items():
-        status = get_update_status(service, ua_info)
         for version, origin in version_list:
+            status = get_update_status(service, ua_info, version)
             updates.append(
                 {
                     "package": version.parent_pkg.name,

uaclient/tests/test_security_status.py

@@ -64,22 +64,46 @@ def get_candidate_ver(self, package: mock.MagicMock):
 
 class TestSecurityStatus:
     @pytest.mark.parametrize(
-        "service_name,ua_info,expected_result",
+        "service_name,ua_info,version,expected_result",
         (
-            ("standard-security", {}, UpdateStatus.AVAILABLE.value),
+            (
+                "standard-security",
+                {},
+                mock_version("1.0"),
+                UpdateStatus.AVAILABLE.value,
+            ),
+            (
+                "standard-security",
+                {},
+                mock_version("0.8"),
+                UpdateStatus.AVAILABLE_NOT_PREFERRED.value,
+            ),
             (
                 "esm",
                 {"attached": True, "enabled_services": ["esm"]},
+                mock_version("1.0"),
                 UpdateStatus.AVAILABLE.value,
             ),
-            ("esm", {"attached": False}, UpdateStatus.UNATTACHED.value),
+            (
+                "esm",
+                {"attached": True, "enabled_services": ["esm"]},
+                mock_version("1.1"),
+                UpdateStatus.AVAILABLE_NOT_PREFERRED.value,
+            ),
+            (
+                "esm",
+                {"attached": False},
+                mock_version("1.0"),
+                UpdateStatus.UNATTACHED.value,
+            ),
             (
                 "esm",
                 {
                     "attached": True,
                     "enabled_services": ["not-esm"],
                     "entitled_services": ["esm"],
                 },
+                mock_version("1.0"),
                 UpdateStatus.NOT_ENABLED.value,
             ),
             (
@@ -89,12 +113,24 @@ class TestSecurityStatus:
                     "enabled_services": ["not-esm"],
                     "entitled_services": ["not-esm-again"],
                 },
+                mock_version("1.0"),
                 UpdateStatus.UNAVAILABLE.value,
             ),
         ),
     )
-    def test_get_update_status(self, service_name, ua_info, expected_result):
-        assert get_update_status(service_name, ua_info) == expected_result
+    @mock.patch(M_PATH + "get_pkg_candidate_version", return_value="1.0")
+    def test_get_update_status(
+        self,
+        m_pkg_candidate_version,
+        service_name,
+        ua_info,
+        version,
+        expected_result,
+    ):
+        assert (
+            get_update_status(service_name, ua_info, version)
+            == expected_result
+        )
 
     @pytest.mark.parametrize("is_attached", (True, False))
     @mock.patch(M_PATH + "_is_attached")
@@ -508,8 +544,10 @@ def test_filter_updates_when_esm_disabled(self, m_esm_cache):
     )
     @mock.patch(M_PATH + "filter_updates")
     @mock.patch(M_PATH + "get_apt_pkg_cache")
+    @mock.patch(M_PATH + "get_pkg_candidate_version", return_value=None)
     def test_security_status_dict(
         self,
+        m_pkg_candidate_version,
         m_cache,
         m_filter_sec_updates,
         _m_get_origin,
@@ -524,11 +562,27 @@ def test_security_status_dict(
         m_version = mock_version("1.0", size=123456)
         m_package = mock_package("example_package", m_version)
 
-        m_cache.return_value.packages = [m_package] * 10
+        # Different package with version that is not candidate
+        m_version_2 = mock_version("2.0", size=123456)
+        m_package_2 = mock_package(
+            "example_pkg_diff_candidate_version", m_version_2
+        )
+
+        m_package_list = [m_package] * 10
+        m_package_list.append(m_package_2)
+        m_cache.return_value.packages = m_package_list
+
+        m_pkg_candidate_version.return_value = "1.0"
+
         m_filter_sec_updates.return_value = {
-            "esm-infra": [(m_version, "some.url.for.esm")] * 2,
+            "esm-infra": [
+                (m_version, "some.url.for.esm"),
+                (m_version, "some.url.for.esm"),
+            ],
             "esm-apps": [],
-            "standard-security": [],
+            "standard-security": [
+                (m_version_2, "security.ubuntu.com"),
+            ],
         }
         m_reboot_status.return_value = mock.MagicMock(
             reboot_required=RebootStatus.REBOOT_NOT_REQUIRED.value
@@ -553,15 +607,23 @@ def test_security_status_dict(
                     "origin": "some.url.for.esm",
                     "download_size": 123456,
                 },
+                {
+                    "package": "example_pkg_diff_candidate_version",
+                    "version": "2.0",
+                    "service_name": "standard-security",
+                    "status": "upgrade_available_not_preferred",
+                    "origin": "security.ubuntu.com",
+                    "download_size": 123456,
+                },
             ],
             "summary": {
                 "ua": {
                     "attached": False,
                     "enabled_services": [],
                     "entitled_services": [],
                 },
-                "num_installed_packages": 10,
-                "num_main_packages": 10,
+                "num_installed_packages": 11,
+                "num_main_packages": 11,
                 "num_restricted_packages": 0,
                 "num_universe_packages": 0,
                 "num_multiverse_packages": 0,
@@ -571,7 +633,7 @@ def test_security_status_dict(
                 "num_esm_apps_packages": 0,
                 "num_esm_infra_updates": 2,
                 "num_esm_apps_updates": 0,
-                "num_standard_security_updates": 0,
+                "num_standard_security_updates": 1,
                 "reboot_required": "no",
             },
             "livepatch": {"fixed_cves": []},