chore: cyndi replacement os display sort and dev data

common/peewee_model.py

@@ -6,6 +6,7 @@
 from datetime import datetime
 from enum import Enum
 
+from peewee import SQL
 from peewee import AutoField
 from peewee import BooleanField
 from peewee import CharField
@@ -16,6 +17,7 @@
 from peewee import Model
 from peewee import TextField
 from peewee import UUIDField
+from peewee import fn
 from playhouse.postgres_ext import ArrayField
 from playhouse.postgres_ext import BinaryJSONField
 
@@ -131,6 +133,23 @@ class OperatingSystem(BaseModel):
     cves_unpatched_moderate = IntegerField(null=False)
     cves_unpatched_low = IntegerField(null=False)
 
+    @classmethod
+    def display_os_format(cls):
+        """Expression for OS display string ('Name Major.Minor'), null handled as 'N/A'."""
+        # OS Name baseline
+        name_base = fn.COALESCE(cls.name, SQL("'N/A'"))
+        # Join Name and Major SPACE
+        name_major = fn.CONCAT_WS(" ", name_base, cls.major)
+        # Join result and Minor DOT
+        full_os_sql_string = fn.CONCAT_WS(".", name_major, cls.minor)
+
+        return full_os_sql_string
+
+    @classmethod
+    def sort_columns(cls):
+        """Columns for sorting by OS, first by name then by major last by minor)"""
+        return [cls.name, cls.major, cls.minor]
+
     class Meta:
         """operating_system table metadata"""
 

manager/cve_handler.py

@@ -36,6 +36,7 @@
 from common.peewee_model import CveMetadata
 from common.peewee_model import InsightsRule
 from common.peewee_model import InventoryHosts
+from common.peewee_model import OperatingSystem
 from common.peewee_model import RHAccount
 from common.peewee_model import SystemCveData
 from common.peewee_model import SystemGroupSet
@@ -203,7 +204,7 @@ def __init__(self, account_data, synopsis, list_args, parsed_args, uri, ids_only
             "status": SQL("status_id"),
             "updated": SQL("updated"),
             "first_reported": SQL("first_reported"),
-            "os": OS_INFO_SORT,
+            "os": ([SQL("os")] if UNLEASH.is_enabled(CYNDI_REPLICATION_REPLACEMENT) else OS_INFO_SORT),
             "advisory_available": SQL("advisory_available"),
             "remediation": SQL("remediation_type_id"),
             "advisories_list": SQL("advisories"),
@@ -313,6 +314,7 @@ def _full_query(rh_account_id, synopsis, parsed_args, filters, remediation_filte
                 on=((CveAccountData.rh_account_id == rh_account_id) & (CveMetadata.id == CveAccountData.cve_id)),
             )
             .join(InsightsRule, JOIN.LEFT_OUTER, on=(InsightsRule.id == SystemVulnerabilities.rule_id))
+            .join(OperatingSystem, JOIN.LEFT_OUTER, on=(SystemPlatform.operating_system_id == OperatingSystem.id))
             .where(CveMetadata.cve == synopsis)
             .where(SystemVulnerabilities.rh_account_id == rh_account_id)
             .where(system_is_active(image=None))
@@ -401,6 +403,7 @@ def _unpatched_full_query(self, rh_account_id, synopsis, parsed_args, filters) -
                 JOIN.LEFT_OUTER,
                 on=((CveAccountData.rh_account_id == rh_account_id) & (CveMetadata.id == CveAccountData.cve_id)),
             )
+            .join(OperatingSystem, JOIN.LEFT_OUTER, on=(SystemPlatform.operating_system_id == OperatingSystem.id))
             .where(CveMetadata.cve == synopsis)
             .where(system_is_active(image=None))
             .where(SystemVulnerablePackage.rh_account_id == rh_account_id)
@@ -469,6 +472,7 @@ def _id_query(rh_account_id, synopsis, parsed_args, filters, remediation_filter=
                 on=((CveAccountData.rh_account_id == rh_account_id) & (CveMetadata.id == CveAccountData.cve_id)),
             )
             .join(InsightsRule, JOIN.LEFT_OUTER, on=(InsightsRule.id == SystemVulnerabilities.rule_id))
+            .join(OperatingSystem, JOIN.LEFT_OUTER, on=(SystemPlatform.operating_system_id == OperatingSystem.id))
             .where(CveMetadata.cve == synopsis)
             .where(SystemVulnerabilities.rh_account_id == rh_account_id)
             .where(system_is_active(image=None))
@@ -539,6 +543,7 @@ def _unpatched_id_query(self, rh_account_id, synopsis, parsed_args, filters) ->
                 JOIN.LEFT_OUTER,
                 on=((CveAccountData.rh_account_id == rh_account_id) & (CveMetadata.id == CveAccountData.cve_id)),
             )
+            .join(OperatingSystem, JOIN.LEFT_OUTER, on=(SystemPlatform.operating_system_id == OperatingSystem.id))
             .where(CveMetadata.cve == synopsis)
             .where(SystemVulnerablePackage.rh_account_id == rh_account_id)
             .where(system_is_active(image=None))

manager/filters.py

@@ -464,15 +464,30 @@ def _filter_system_by_rhel_version(query, args, _kwargs):
     """
     if "rhel_version" in args and args["rhel_version"]:
         expr = None
-        for version in args["rhel_version"]:
-            if "." in version:
-                major, minor = version.split(".", 1)
-                expr = expr | \
-                    ((InventoryHosts.system_profile["operating_system"]["major"] == major) &
-                     (InventoryHosts.system_profile["operating_system"]["minor"] == minor))
-            else:
-                expr = expr | (InventoryHosts.system_profile["operating_system"]["major"] == version)
-        query = query.where((InventoryHosts.system_profile["operating_system"]["name"] == "RHEL") & (expr))
+        if UNLEASH.is_enabled(CYNDI_REPLICATION_REPLACEMENT):
+            for version in args["rhel_version"]:
+                try:
+                    if "." in version:
+                        major, minor = version.split(".", 1)
+                        major, minor = int(major), int(minor)  # Operating System is storing ints
+                        cond = (OperatingSystem.major == major) & (OperatingSystem.minor == minor)
+                    else:
+                        cond = OperatingSystem.major == int(version)
+                except (ValueError, TypeError):
+                    continue
+                expr = cond if expr is None else expr | cond
+            if expr is not None:
+                query = query.where((OperatingSystem.name == "RHEL") & (expr))
+        else:
+            for version in args["rhel_version"]:
+                if "." in version:
+                    major, minor = version.split(".", 1)
+                    expr = expr | \
+                        ((InventoryHosts.system_profile["operating_system"]["major"] == major) &
+                            (InventoryHosts.system_profile["operating_system"]["minor"] == minor))
+                else:
+                    expr = expr | (InventoryHosts.system_profile["operating_system"]["major"] == version)
+            query = query.where((InventoryHosts.system_profile["operating_system"]["name"] == "RHEL") & (expr))
     return query
 
 

manager/system_handler.py

@@ -33,6 +33,7 @@
 from common.peewee_model import CveMetadata
 from common.peewee_model import InsightsRule
 from common.peewee_model import InventoryHosts
+from common.peewee_model import OperatingSystem
 from common.peewee_model import RHAccount
 from common.peewee_model import SystemCveData
 from common.peewee_model import SystemGroupSet
@@ -559,6 +560,7 @@ def _full_query(rh_account_id):
         return (
             SystemPlatform.select(*selectables)
             .join(SystemTagSet, JOIN.LEFT_OUTER, on=(SystemPlatform.tag_set_id == SystemTagSet.id))
+            .join(OperatingSystem, JOIN.LEFT_OUTER, on=(SystemPlatform.operating_system_id == OperatingSystem.id))
             .where(SystemPlatform.rh_account_id == rh_account_id)
             .where(system_is_active(deleted=False, image=None, opt_out=None, stale=None))
             .where(SystemPlatform.last_evaluation.is_null(False) | SystemPlatform.advisor_evaluated.is_null(False))
@@ -576,6 +578,7 @@ def _id_query(rh_account_id, list_args):
         query = (
             SystemPlatform.select(*selectables)
             .join(SystemTagSet, JOIN.LEFT_OUTER, on=(SystemPlatform.tag_set_id == SystemTagSet.id))
+            .join(OperatingSystem, JOIN.LEFT_OUTER, on=(SystemPlatform.operating_system_id == OperatingSystem.id))
             .where(SystemPlatform.rh_account_id == rh_account_id)
             .where(system_is_active(deleted=False, image=None, opt_out=None, stale=None))
             .where(SystemPlatform.last_evaluation.is_null(False) | SystemPlatform.advisor_evaluated.is_null(False))
@@ -882,13 +885,15 @@ def handle_get(cls, **kwargs):
             query = (
                 SystemPlatform.select(*selectables)
                 .join(SystemTagSet, JOIN.LEFT_OUTER, on=(SystemPlatform.tag_set_id == SystemTagSet.id))
+                .join(OperatingSystem, JOIN.LEFT_OUTER, on=(SystemPlatform.operating_system_id == OperatingSystem.id))
                 .where(SystemPlatform.inventory_id == inventory_id)
                 .where(SystemPlatform.rh_account_id == rh_account_id)
                 .where(SystemPlatform.when_deleted.is_null(True))
                 .dicts()
             )
+            query = cyndi_join(query)
             query = apply_filters(query, args, [filter_types.SYSTEM_TAGS], {})
-            system = cyndi_join(query).get()
+            system = query.get()
         except DoesNotExist as exc:
             raise ApplicationException("inventory_id must exist and inventory_id must be visible to user", 404) from exc
         if context.context["user"]["identity_type"] == "System" and system["owner_id"] != context.context["user"]["system_cn"]:

manager/vulnerabilities_handler.py

@@ -226,6 +226,9 @@ def _count_subquery(rh_account_id, args, filters, remediation_filter=None):
                                                            & (CveAccountData.rh_account_id == rh_account_id)))
                 .join(SystemCveData, JOIN.LEFT_OUTER, on=((SystemPlatform.id == SystemCveData.system_id)
                                                           & (SystemVulnerabilities.cve_id == SystemCveData.cve_id)))
+                .join(OperatingSystem, JOIN.LEFT_OUTER,
+                      on=(SystemPlatform.operating_system_id == OperatingSystem.id))
+
                 .where(SystemVulnerabilities.rh_account_id == rh_account_id)
                 .where(system_is_vulnerable())
                 .group_by(SystemVulnerabilities.cve_id, SystemPlatform.operating_system_id))
@@ -260,6 +263,9 @@ def _unpatched_count_subquery(rh_account_id, args, filters):
                                                                    & (CveAccountData.rh_account_id == rh_account_id)))
                         .join(SystemCveData, JOIN.LEFT_OUTER, on=((SystemPlatform.id == SystemCveData.system_id)
                                                                   & (VulnerablePackageCVE.cve_id == SystemCveData.cve_id)))
+                        .join(OperatingSystem, JOIN.LEFT_OUTER,
+                              on=(SystemPlatform.operating_system_id == OperatingSystem.id))
+
                         .where(SystemVulnerablePackage.rh_account_id == rh_account_id)
                         .group_by(VulnerablePackageCVE.cve_id, SystemPlatform.operating_system_id))
         unfixed_subq = cyndi_join(unfixed_subq)

tests/manager_tests/test_cve_handler.py

@@ -301,6 +301,37 @@ def test_cve_sort_check(self):
             self.vfetch(f"cves/CVE-2023-1/affected_systems?advisory_available=true,false&sort={sort}").check_response()
             self.vfetch(f"cves/CVE-2023-1/affected_systems?advisory_available=true&sort={sort}").check_response()
 
+    def test_cves_affected_systems_rhel_version(self):
+        """Test affected_systems rhel_version filter"""
+        response = self.vfetch("cves/CVE-2023-1/affected_systems?rhel_version=7.4").check_response()
+        assert response.body.meta.total_items == len(response.body.data)
+        for system in response.body.data:
+            assert system.attributes.get("os") is not None
+            assert "7.4" in str(
+                system.attributes["os"]
+            ), f"Filter rhel_version=7.4 should only return systems with 7.4, got os={system.attributes['os']}"
+
+    def test_cves_affected_systems_rhel_version_non_existing(self):
+        # RHEL 10 is not in test data, filter should return no systems
+        response_10 = self.vfetch("cves/CVE-2023-1/affected_systems?rhel_version=10").check_response()
+        assert response_10.body.meta.total_items == 0
+        assert len(response_10.body.data) == 0
+        # As well as RHEL 8.1 and 8.2
+        response_8dot1 = self.vfetch("cves/CVE-2023-1/affected_systems?rhel_version=8.1").check_response()
+        assert response_8dot1.body.meta.total_items == 0
+        assert len(response_8dot1.body.data) == 0
+        response_8dot2 = self.vfetch("cves/CVE-2023-1/affected_systems?rhel_version=8.2").check_response()
+        assert response_8dot2.body.meta.total_items == 0
+        assert len(response_8dot2.body.data) == 0
+        # 8.3 should return the one affected system
+        response_8dot3 = self.vfetch("cves/CVE-2023-1/affected_systems?rhel_version=8.3").check_response()
+        assert response_8dot3.body.meta.total_items == len(response_8dot3.body.data) == 1
+        for system in response_8dot3.body.data:
+            assert system.attributes.get("os") is not None
+            assert "8.3" in str(
+                system.attributes["os"]
+            ), f"Filter rhel_version=8.3 should only return systems with 8.3, got os={system.attributes['os']}"
+
     def test_cve_filter_check(self):
         filters = {
             "rule": ["true", "false", "affected_not_vulnerable"],

tests/manager_tests/test_system_handler.py

@@ -105,11 +105,22 @@ def get_workspace_ids_for_permission(self, permission, principal_id):
 
         return calls
 
-    @pytest.mark.parametrize("value,expected", RHEL_VERSION_FILTER, ids=[x[0] for x in RHEL_VERSION_FILTER])
+    @pytest.mark.parametrize(
+        "value,expected",
+        RHEL_VERSION_FILTER,
+        ids=[v for v, _ in RHEL_VERSION_FILTER],
+    )
     def test_system_rhel_version(self, value, expected):
         """Test SYSTEM_RHEL_VERSION filter"""
         response = self.vfetch("systems?rhel_version={}".format(value)).check_response()
         assert len(response.body.data) == expected
+        if expected > 0:
+            allowed = [p.strip() for p in value.split(",")]
+            for system in response.body.data:
+                os_str = str(system.attributes.get("os") or "")
+                assert any(
+                    part in os_str for part in allowed
+                ), "Filter rhel_version={} should only return systems with os in {}, got os={}".format(value, allowed, os_str)
 
     def test_systems(self):
         response = self.vfetch("systems?excluded=false").check_response()
@@ -274,6 +285,67 @@ def test_system_details_invalid_id(self):
         response = self.vfetch("systems/b1f21450-0000-1111-2222-000000000000").check_response(status_code=404)
         assert response.body.errors[0].detail == "inventory_id must exist and inventory_id must be visible to user"
 
+    def test_system_details_response_structure_known_system(self):
+        """Check whether GetSystemDetails returns all expected fields with correct types on KNOWN system"""
+        response = self.vfetch("systems/00000000-0000-0000-0000-000000000005").check_response()
+        data = response.body.data
+
+        expected_fields = (
+            "last_evaluation",
+            "rules_evaluation",
+            "opt_out",
+            "last_upload",
+            "stale",
+            "host_type",
+            "os",
+            "rhsm_lock",
+            "tags",
+            "updated",
+            "insights_id",
+            "inventory_group",
+        )
+        for field in expected_fields:  # All expected fields needs to be present...
+            assert hasattr(data, field), "Missing field: {}".format(field)
+
+        # ...as well as data types match
+        assert isinstance(data.opt_out, bool)
+        assert isinstance(data.stale, bool)
+        assert isinstance(data.tags, list)
+        assert isinstance(data.inventory_group, list)
+        assert isinstance(data.os, str)
+        assert isinstance(data.host_type, str)
+
+    def test_system_details_response_structure_any_system(self):
+        """GetSystemDetails structure should be consistent for ANY system from the list"""
+        list_response = self.vfetch("systems?excluded=false").check_response()
+        assert len(list_response.body.data) > 0
+        inventory_id = list_response.body.data[0].id
+
+        response = self.vfetch("systems/{}".format(inventory_id)).check_response()
+        data = response.body.data
+
+        expected_fields = (
+            "last_evaluation",
+            "rules_evaluation",
+            "opt_out",
+            "last_upload",
+            "stale",
+            "host_type",
+            "os",
+            "rhsm_lock",
+            "tags",
+            "updated",
+            "insights_id",
+            "inventory_group",
+        )
+        for field in expected_fields:
+            assert hasattr(data, field), "Missing field: {}".format(field)
+
+        assert isinstance(data.opt_out, bool)
+        assert isinstance(data.stale, bool)
+        assert isinstance(data.tags, list)
+        assert isinstance(data.inventory_group, list)
+
     def test_system_business_risk_filtering(self):
         response = self.vfetch("systems/00000000-0000-0000-0000-000000000005/cves?business_risk_id=0").check_response()
         assert len(response.body.data) == 9