feat: remove content access mode cache

This patch aims to remove references to the
'ContentAccessModeCache' and refactor/simplify related
code that checks this cache since entitlement is being phased
out and SCA will be the only content access mode.

Resolves: CCT-619

Signed-off-by: Jason Jerome <jajerome@redhat.com>

src/rhsmlib/services/entitlement.py

@@ -612,11 +612,6 @@ def refresh(self, remove_cache: bool = False, force: bool = False) -> None:
             content_access = inj.require(inj.CONTENT_ACCESS_CACHE)
             if content_access.exists():
                 content_access.remove()
-            # Also remove the content access mode cache to be sure we display
-            # SCA or regular mode correctly
-            content_access_mode = inj.require(inj.CONTENT_ACCESS_MODE_CACHE)
-            if content_access_mode.exists():
-                content_access_mode.delete_cache()
 
         if force is True:
             # Force a regen of the entitlement certs for this consumer

src/rhsmlib/services/refresh.py

@@ -54,12 +54,6 @@ def refresh(self, force: bool = False) -> None:
         :return: None
         """
 
-        # First remove the content access mode cache to be sure we display
-        # SCA or regular mode correctly
-        content_access_mode = inj.require(inj.CONTENT_ACCESS_MODE_CACHE)
-        if content_access_mode.exists():
-            content_access_mode.delete_cache()
-
         # Remove the release status cache, in case it was changed
         # on the server; it will be fetched when needed again
         inj.require(inj.RELEASE_STATUS_CACHE).delete_cache()

src/rhsmlib/services/register.py

@@ -185,26 +185,6 @@ def register(
         # Save syspurpose attributes from consumer to cache file
         syspurposelib.write_syspurpose_cache(syspurpose_dict)
 
-        content_access_mode_cache = inj.require(inj.CONTENT_ACCESS_MODE_CACHE)
-
-        # Is information about content access mode included in consumer
-        if "owner" not in consumer:
-            log.warning("Consumer does not contain any information about owner.")
-        elif "contentAccessMode" in consumer["owner"]:
-            log.debug("Saving content access mode from consumer object to cache file.")
-            # When we know content access mode from consumer, then write it to cache file
-            content_access_mode = consumer["owner"]["contentAccessMode"]
-            content_access_mode_cache.set_data(content_access_mode, self.identity)
-            content_access_mode_cache.write_cache()
-        else:
-            # If not, then we have to do another REST API call to get this information
-            # It will not be included in cache file. When cache file is empty, then
-            # it will trigger accessing REST API and saving result in cache file.
-            log.debug("Information about content access mode is not included in consumer")
-            content_access_mode = content_access_mode_cache.read_data()
-            # Add information about content access mode to consumer
-            consumer["owner"]["contentAccessMode"] = content_access_mode
-
         return consumer
 
     def validate_options(self, options: dict) -> None:

src/subscription_manager/cache.py

@@ -1023,72 +1023,6 @@ def _is_cache_obsoleted(self, uep: connection.UEPConnection, identity: "Identity
             return True
 
 
-class ContentAccessModeCache(ConsumerCache):
-    """
-    Cache information about current owner (organization), specifically, the content access mode.
-    This value is used independently.
-    """
-
-    # Grab the current owner (and hence the content_access_mode of that owner) at most, once per
-    # 4 hours
-    TIMEOUT = 60 * 60 * 4
-
-    CACHE_FILE = "/var/lib/rhsm/cache/content_access_mode.json"
-
-    def __init__(self, data: Any = None):
-        super(ContentAccessModeCache, self).__init__(data=data)
-
-    def _sync_with_server(
-        self, uep: connection.UEPConnection, consumer_uuid: str, _: Optional[datetime.datetime] = None
-    ) -> str:
-        try:
-            current_owner: Dict = uep.getOwner(consumer_uuid)
-        except Exception:
-            log.debug(
-                "Error checking for content access mode,"
-                "defaulting to assuming not in Simple Content Access mode"
-            )
-        else:
-            if "contentAccessMode" in current_owner:
-                return current_owner["contentAccessMode"]
-            else:
-                log.debug(
-                    "The owner returned from the server did not contain a "
-                    "'content_access_mode'. Perhaps the connected Entitlement Server doesn't"
-                    "support 'content_access_mode'?"
-                )
-        return "unknown"
-
-    def _is_cache_obsoleted(self, uep: connection.UEPConnection, identity: "Identity"):
-        """
-        We don't know if the cache is valid until we get valid response
-        :param uep: object representing connection to candlepin server
-        :param identity: consumer identity
-        :return: True, when cache is obsoleted or validity of cache is unknown.
-        """
-        if uep is None:
-            cp_provider: CPProvider = inj.require(inj.CP_PROVIDER)
-            uep: connection.UEPConnection = cp_provider.get_consumer_auth_cp()
-
-        if hasattr(uep.conn, "is_consumer_cert_key_valid"):
-            if uep.conn.is_consumer_cert_key_valid is None:
-                log.debug(
-                    f"Cache file {self.CACHE_FILE} cannot be considered as valid, because no connection has "
-                    "been created yet"
-                )
-                return True
-            elif uep.conn.is_consumer_cert_key_valid is True:
-                return False
-            else:
-                log.debug(
-                    f"Cache file {self.CACHE_FILE} cannot be considered as valid, "
-                    "because consumer certificate probably is not valid"
-                )
-                return True
-        else:
-            return True
-
-
 class SupportedResourcesCache(ConsumerCache):
     """
     Cache supported resources of candlepin server for current identity

src/subscription_manager/injection.py

@@ -19,7 +19,6 @@
 PRODUCT_DATE_RANGE_CALCULATOR = "PRODUCT_DATE_RANGE_CALCULATOR"
 ENT_DIR = "ENT_DIR"
 PROD_DIR = "PROD_DIR"
-CONTENT_ACCESS_MODE_CACHE = "CONTENT_ACCESS_MODE_CACHE"
 CURRENT_OWNER_CACHE = "CURRENT_OWNER_CACHE"
 SYSPURPOSE_VALID_FIELDS_CACHE = "SYSPURPOSE_VALID_FIELDS_CACHE"
 SUPPORTED_RESOURCES_CACHE = "SUPPORTED_RESOURCES_CACHE"

src/subscription_manager/injectioninit.py

@@ -30,7 +30,6 @@
     AvailableEntitlementsCache,
     CurrentOwnerCache,
     SyspurposeValidFieldsCache,
-    ContentAccessModeCache,
 )
 
 from subscription_manager.cert_sorter import CertSorter
@@ -65,7 +64,6 @@ def init_dep_injection():
     #        but runs a new version of injectioninit...)
     inj.provide(inj.ENTITLEMENT_STATUS_CACHE, EntitlementStatusCache, singleton=True)
     inj.provide(inj.SYSTEMPURPOSE_COMPLIANCE_STATUS_CACHE, SyspurposeComplianceStatusCache, singleton=True)
-    inj.provide(inj.CONTENT_ACCESS_MODE_CACHE, ContentAccessModeCache, singleton=True)
     inj.provide(inj.CURRENT_OWNER_CACHE, CurrentOwnerCache, singleton=True)
     inj.provide(inj.SYSPURPOSE_VALID_FIELDS_CACHE, SyspurposeValidFieldsCache)
     inj.provide(inj.SUPPORTED_RESOURCES_CACHE, SupportedResourcesCache, singleton=True)

src/subscription_manager/utils.py

@@ -202,8 +202,7 @@ def is_simple_content_access(
     # When identity is not known, then system is not registered
     if identity.uuid is None:
         return False
-    content_access_mode = inj.require(inj.CONTENT_ACCESS_MODE_CACHE).read_data(uep=uep)
-    return content_access_mode == "org_environment"
+    return True
 
 
 def get_current_owner(uep: Optional["UEPConnection"] = None, identity: "Identity" = None) -> dict:

subscription-manager.spec

@@ -740,6 +740,7 @@ rmdir %{python_sitearch}/subscription_manager-*-*.egg-info --ignore-fail-on-non-
 # Remove old cache files
 # The -f flag ensures that exit code 0 will be returned even if the file does not exist.
 rm -f /var/lib/rhsm/cache/rhsm_icon.json
+rm -f /var/lib/rhsm/cache/content_access_mode.json
 
 %changelog
 * Thu Sep 26 2024 Pino Toscano <ptoscano@redhat.com> 1.30.2-1

test/cli_command/test_refresh.py

@@ -2,8 +2,8 @@
 from ..test_managercli import TestCliProxyCommand
 from unittest.mock import Mock
 from subscription_manager import managercli
-from subscription_manager.cache import ContentAccessCache, ContentAccessModeCache
-from subscription_manager.injection import provide, CONTENT_ACCESS_CACHE, CONTENT_ACCESS_MODE_CACHE
+from subscription_manager.cache import ContentAccessCache
+from subscription_manager.injection import provide, CONTENT_ACCESS_CACHE
 
 
 class TestRefreshCommand(TestCliProxyCommand):
@@ -28,15 +28,8 @@ def test_cache_removed(self):
         mock_content_access_cache = Mock(spec=ContentAccessCache)
         mock_content_access_cache.return_value.exists.return_value = True
         provide(CONTENT_ACCESS_CACHE, mock_content_access_cache)
-        mock_content_access_mode_cache = Mock(spec=ContentAccessModeCache)
-        mock_content_access_mode_cache.return_value.exists.return_value = True
-        provide(CONTENT_ACCESS_MODE_CACHE, mock_content_access_mode_cache)
 
         self.cc.main([])
 
         # This cache should not be deleted to be able to use HTTP header 'If-Modified-Since'
         mock_content_access_cache.return_value.remove.assert_not_called()
-        # Cache about content access mode should be deleted, because content access mode
-        # can be changed from SCA to entitlement and vice versa
-        mock_content_access_mode_cache.return_value.exists.assert_called_once()
-        mock_content_access_mode_cache.return_value.delete_cache.assert_called_once()

test/fixture.py

@@ -203,7 +203,6 @@ def unstub_conf():
         inj.provide(inj.SUPPORTED_RESOURCES_CACHE, stubs.StubSupportedResourcesCache())
         inj.provide(inj.SYSPURPOSE_VALID_FIELDS_CACHE, stubs.StubSyspurposeValidFieldsCache())
         inj.provide(inj.CURRENT_OWNER_CACHE, stubs.StubCurrentOwnerCache)
-        inj.provide(inj.CONTENT_ACCESS_MODE_CACHE, stubs.StubContentAccessModeCache())
         inj.provide(inj.OVERRIDE_STATUS_CACHE, stubs.StubOverrideStatusCache())
         inj.provide(inj.RELEASE_STATUS_CACHE, stubs.StubReleaseStatusCache())
         inj.provide(inj.AVAILABLE_ENTITLEMENT_CACHE, stubs.StubAvailableEntitlementsCache())

test/rhsmlib/services/test_register.py

@@ -16,7 +16,7 @@
 
 import subscription_manager.injection as inj
 
-from subscription_manager.cache import InstalledProductsManager, ContentAccessModeCache
+from subscription_manager.cache import InstalledProductsManager
 from subscription_manager.cp_provider import CPProvider
 from subscription_manager.facts import Facts
 from subscription_manager.identity import Identity
@@ -180,12 +180,6 @@ def setUp(self):
         # Add a mock cp_provider
         self.mock_cp_provider = mock.Mock(spec=CPProvider, name="CPProvider")
 
-        # Add a mock for content access mode cache
-        self.mock_content_access_mode_cache = mock.Mock(
-            spec=ContentAccessModeCache, name="ContentAccessModeCache"
-        )
-        self.mock_content_access_mode_cache.read_data = mock.Mock(return_value="entitlement")
-
         # For the tests in which it's used, the consumer_auth cp and basic_auth cp can be the same
         self.mock_cp_provider.get_consumer_auth_cp.return_value = self.mock_cp
         self.mock_cp_provider.get_basic_auth_cp.return_value = self.mock_cp
@@ -215,8 +209,6 @@ def injection_definitions(self, *args, **kwargs):
             return self.mock_facts
         elif args[0] == inj.CP_PROVIDER:
             return self.mock_cp_provider
-        elif args[0] == inj.CONTENT_ACCESS_MODE_CACHE:
-            return self.mock_content_access_mode_cache
         else:
             return None
 

test/stubs.py

@@ -32,7 +32,6 @@
     AvailableEntitlementsCache,
     SyspurposeValidFieldsCache,
     CurrentOwnerCache,
-    ContentAccessModeCache,
     SyspurposeComplianceStatusCache,
 )
 from subscription_manager.facts import Facts
@@ -785,14 +784,6 @@ def delete_cache(self):
         self.server_status = None
 
 
-class StubContentAccessModeCache(ContentAccessModeCache):
-    def write_cache(self, debug=False):
-        pass
-
-    def delete_cache(self):
-        self.server_status = None
-
-
 class StubSupportedResourcesCache(SupportedResourcesCache):
     def write_cache(self, debug=False):
         pass

test/test_cache.py

@@ -47,7 +47,6 @@
     SupportedResourcesCache,
     AvailableEntitlementsCache,
     CurrentOwnerCache,
-    ContentAccessModeCache,
     SyspurposeComplianceStatusCache,
 )
 
@@ -1327,28 +1326,6 @@ def test_max_timeout(self):
         self.assertEqual(timeout, self.cache.UBOUND)
 
 
-class TestContentAccessModeCache(SubManFixture):
-    MOCK_CACHE_FILE_CONTENT = '{"7f85da06-5c35-44ba-931d-f11f6e581f89": "entitlement"}'
-
-    def setUp(self):
-        super(TestContentAccessModeCache, self).setUp()
-        self.cache = ContentAccessModeCache()
-
-    def test_reading_nonexisting_cache(self):
-        data = self.cache.read_cache_only()
-        self.assertIsNone(data)
-
-    def test_reading_existing_cache(self):
-        temp_cache_dir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, temp_cache_dir)
-        self.cache.CACHE_FILE = os.path.join(temp_cache_dir, "content_access_mode.json")
-        with open(self.cache.CACHE_FILE, "w") as cache_file:
-            cache_file.write(self.MOCK_CACHE_FILE_CONTENT)
-        data = self.cache.read_cache_only()
-        self.assertTrue("7f85da06-5c35-44ba-931d-f11f6e581f89" in data)
-        self.assertEqual(data["7f85da06-5c35-44ba-931d-f11f6e581f89"], "entitlement")
-
-
 class TestSyspurposeComplianceStatusCache(SubManFixture):
     def setUp(self):
         super(TestSyspurposeComplianceStatusCache, self).setUp()

test/test_cert_sorter.py

@@ -63,8 +63,9 @@ def stub_prod_cert(pid):
 
 
 class CertSorterTests(SubManFixture):
+    @patch("subscription_manager.cert_sorter.utils.is_simple_content_access")
     @patch("subscription_manager.cache.InstalledProductsManager.update_check")
-    def setUp(self, mock_update):
+    def setUp(self, mock_update, mock_is_simple_content_access):
         SubManFixture.setUp(self)
         # Setup mock product and entitlement certs:
         self.prod_dir = StubProductDirectory(pids=[INST_PID_1, INST_PID_2, INST_PID_3, INST_PID_4])
@@ -79,6 +80,7 @@ def setUp(self, mock_update):
                 ),
             ]
         )
+        mock_is_simple_content_access.return_value = False
 
         self.mock_uep = StubUEP()
 
@@ -170,11 +172,13 @@ def test_installed_mismatch_unentitled(self, mock_update):
         # server reported it here:
         self.assertFalse(INST_PID_3 in sorter.unentitled_products)
 
+    @patch("subscription_manager.cert_sorter.utils.is_simple_content_access")
     @patch("subscription_manager.cache.InstalledProductsManager.update_check")
-    def test_missing_installed_product(self, mock_update):
+    def test_missing_installed_product(self, mock_update, mock_is_simple_content_access):
         # Add a new installed product server doesn't know about:
         prod_dir = StubProductDirectory(pids=[INST_PID_1, INST_PID_2, INST_PID_3, "product4"])
         inj.provide(inj.PROD_DIR, prod_dir)
+        mock_is_simple_content_access.return_value = False
         sorter = CertSorter()
         self.assertTrue("product4" in sorter.unentitled_products)
 

test/test_utils.py

@@ -798,17 +798,12 @@ def setUp(self):
         super(TestIsOwnerUsingSimpleContentAccess, self).setUp()
         self.cp_provider = Mock()
         self.mock_uep = Mock()
-        self.mock_uep.getOwner = Mock(return_value=self.MOCK_ENTITLEMENT_OWNER)
+        self.mock_uep.getOwner = Mock(return_value=self.MOCK_ORG_ENVIRONMENT_OWNER)
         self.cp_provider.get_consumer_auth_cp = Mock(return_value=self.mock_uep)
         self.identity = Mock()
         self.identity.uuid = Mock(return_value="7f85da06-5c35-44ba-931d-f11f6e581f89")
 
-    def test_get_entitlement_owner(self):
-        ret = is_simple_content_access(uep=self.mock_uep, identity=self.identity)
-        self.assertFalse(ret)
-
     def test_get_org_environment_owner(self):
-        self.mock_uep.getOwner = Mock(return_value=self.MOCK_ORG_ENVIRONMENT_OWNER)
         ret = is_simple_content_access(uep=self.mock_uep, identity=self.identity)
         self.assertTrue(ret)