Added support for syspurpose to redhat_subscribtion module (#59850)

* Added several unit tests
* Added documentation for new syspurpose option and suboptions
* Simplified specification of module arguments
* Added new changelog file with fragments

changelogs/fragments/redhat_subscribtion.yml

@@ -2,4 +2,4 @@ bugfixes:
   - redhat_subscription - made code compatible with Python3 (https://github.com/ansible/ansible/pull/58665)
 minor_changes:
   - redhat_subscription - do not call ``subscribtion-manager`` command, when it is not necessary (https://github.com/ansible/ansible/pull/58665)
-  - redhat_subscription - made code more testable (https://github.com/ansible/ansible/pull/58665)
+  - redhat_subscription - made code more testable (https://github.com/ansible/ansible/pull/58665)

changelogs/fragments/redhat_subscribtion_syspurpose.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - redhat_subscription - allow to set syspurpose attributes (https://github.com/ansible/ansible/pull/59850)

lib/ansible/modules/packaging/os/redhat_subscription.py

@@ -131,6 +131,34 @@ options:
         description:
             - Set a release version
         version_added: "2.8"
+    syspurpose:
+        description:
+            - Set syspurpose attributes in file C(/etc/rhsm/syspurpose/syspurpose.json)
+              and synchronize these attributes with RHSM server. Syspurpose attributes help attach
+              the most appropriate subscriptions to the system automatically. When C(syspurpose.json) file
+              already contains some attributes, then new attributes overwrite existing attributes.
+              When some attribute is not listed in the new list of attributes, the existing
+              attribute will be removed from C(syspurpose.json) file. Unknown attributes are ignored.
+        type: dict
+        default: {}
+        version_added: "2.9"
+        suboptions:
+            usage:
+                description: Syspurpose attribute usage
+            role:
+                description: Syspurpose attribute role
+            service_level_agreement:
+                description: Syspurpose attribute service_level_agreement
+            addons:
+                description: Syspurpose attribute addons
+                type: list
+            sync:
+                description:
+                    - When this option is true, then syspurpose attributes are synchronized with
+                      RHSM server immediately. When this option is false, then syspurpose attributes
+                      will be synchronized with RHSM server by rhsmcertd daemon.
+                type: bool
+                default: False
 '''
 
 EXAMPLES = '''
@@ -201,6 +229,21 @@ EXAMPLES = '''
     username: joe_user
     password: somepass
     release: 7.4
+
+- name: Register as user (joe_user) with password (somepass), set syspurpose attributes and synchronize them with server
+  redhat_subscription:
+    state: present
+    username: joe_user
+    password: somepass
+    auto_attach: true
+    syspurpose:
+      usage: "Production"
+      role: "Red Hat Enterprise Server"
+      service_level_agreement: "Premium"
+      addons:
+        - addon1
+        - addon2
+      sync: true
 '''
 
 RETURN = '''
@@ -213,10 +256,12 @@ subscribed_pool_ids:
     }
 '''
 
-import os
+from os.path import isfile
+from os import unlink
 import re
 import shutil
 import tempfile
+import json
 
 from ansible.module_utils.basic import AnsibleModule
 from ansible.module_utils._text import to_native
@@ -240,8 +285,8 @@ class RegistrationBase(object):
 
     def enable(self):
         # Remove any existing redhat.repo
-        if os.path.isfile(self.REDHAT_REPO):
+        if isfile(self.REDHAT_REPO):
-            os.unlink(self.REDHAT_REPO)
+            unlink(self.REDHAT_REPO)
 
     def register(self):
         raise NotImplementedError("Must be implemented by a sub-class")
@@ -255,7 +300,7 @@ class RegistrationBase(object):
     def update_plugin_conf(self, plugin, enabled=True):
         plugin_conf = '/etc/yum/pluginconf.d/%s.conf' % plugin
 
-        if os.path.isfile(plugin_conf):
+        if isfile(plugin_conf):
             tmpfd, tmpfile = tempfile.mkstemp()
             shutil.copy2(plugin_conf, tmpfile)
             cfg = configparser.ConfigParser()
@@ -549,6 +594,13 @@ class Rhsm(RegistrationBase):
         return {'changed': changed, 'subscribed_pool_ids': missing_pools.keys(),
                 'unsubscribed_serials': serials}
 
+    def sync_syspurpose(self):
+        """
+        Try to synchronize syspurpose attributes with server
+        """
+        args = [SUBMAN_CMD, 'status']
+        rc, stdout, stderr = self.module.run_command(args, check_rc=False)
+
 
 class RhsmPool(object):
     '''
@@ -644,71 +696,124 @@ class RhsmPools(object):
                 yield product
 
 
+class SysPurpose(object):
+    """
+    This class is used for reading and writing to syspurpose.json file
+    """
+
+    SYSPURPOSE_FILE_PATH = "/etc/rhsm/syspurpose/syspurpose.json"
+
+    ALLOWED_ATTRIBUTES = ['role', 'usage', 'service_level_agreement', 'addons']
+
+    def __init__(self, path=None):
+        """
+        Initialize class used for reading syspurpose json file
+        """
+        self.path = path or self.SYSPURPOSE_FILE_PATH
+
+    def update_syspurpose(self, new_syspurpose):
+        """
+        Try to update current syspurpose with new attributes from new_syspurpose
+        """
+        syspurpose = {}
+        syspurpose_changed = False
+        for key, value in new_syspurpose.items():
+            if key in self.ALLOWED_ATTRIBUTES:
+                if value is not None:
+                    syspurpose[key] = value
+            elif key == 'sync':
+                pass
+            else:
+                raise KeyError("Attribute: %s not in list of allowed attributes: %s" %
+                               (key, self.ALLOWED_ATTRIBUTES))
+        current_syspurpose = self._read_syspurpose()
+        if current_syspurpose != syspurpose:
+            syspurpose_changed = True
+        # Update current syspurpose with new values
+        current_syspurpose.update(syspurpose)
+        # When some key is not listed in new syspurpose, then delete it from current syspurpose
+        # and ignore custom attributes created by user (e.g. "foo": "bar")
+        for key in list(current_syspurpose):
+            if key in self.ALLOWED_ATTRIBUTES and key not in syspurpose:
+                del current_syspurpose[key]
+        self._write_syspurpose(current_syspurpose)
+        return syspurpose_changed
+
+    def _write_syspurpose(self, new_syspurpose):
+        """
+        This function tries to update current new_syspurpose attributes to
+        json file.
+        """
+        with open(self.path, "w") as fp:
+            fp.write(json.dumps(new_syspurpose, indent=2, ensure_ascii=False, sort_keys=True))
+
+    def _read_syspurpose(self):
+        """
+        Read current syspurpuse from json file.
+        """
+        current_syspurpose = {}
+        try:
+            with open(self.path, "r") as fp:
+                content = fp.read()
+        except IOError:
+            pass
+        else:
+            current_syspurpose = json.loads(content)
+        return current_syspurpose
+
+
 def main():
 
     # Load RHSM configuration from file
     rhsm = Rhsm(None)
 
+    # Note: the default values for parameters are:
+    # 'type': 'str', 'default': None, 'required': False
+    # So there is no need to repeat these values for each parameter.
     module = AnsibleModule(
-        argument_spec=dict(
+        argument_spec={
-            state=dict(default='present',
+            'state': {'default': 'present', 'choices': ['present', 'absent']},
-                       choices=['present', 'absent']),
+            'username': {},
-            username=dict(default=None,
+            'password': {'no_log': True},
-                          required=False),
+            'server_hostname': {},
-            password=dict(default=None,
+            'server_insecure': {},
-                          required=False,
+            'rhsm_baseurl': {},
-                          no_log=True),
+            'rhsm_repo_ca_cert': {},
-            server_hostname=dict(default=None,
+            'auto_attach': {'aliases': ['autosubscribe'], 'type': 'bool'},
-                                 required=False),
+            'activationkey': {'no_log': True},
-            server_insecure=dict(default=None,
+            'org_id': {},
-                                 required=False),
+            'environment': {},
-            rhsm_baseurl=dict(default=None,
+            'pool': {'default': '^$'},
-                              required=False),
+            'pool_ids': {'default': [], 'type': 'list'},
-            rhsm_repo_ca_cert=dict(default=None, required=False),
+            'consumer_type': {},
-            auto_attach=dict(aliases=['autosubscribe'], default=False, type='bool'),
+            'consumer_name': {},
-            activationkey=dict(default=None,
+            'consumer_id': {},
-                               required=False,
+            'force_register': {'default': False, 'type': 'bool'},
-                               no_log=True),
+            'server_proxy_hostname': {},
-            org_id=dict(default=None,
+            'server_proxy_port': {},
-                        required=False),
+            'server_proxy_user': {},
-            environment=dict(default=None,
+            'server_proxy_password': {'no_log': True},
-                             required=False, type='str'),
+            'release': {},
-            pool=dict(default='^$',
+            'syspurpose': {
-                      required=False,
+                'type': 'dict',
-                      type='str'),
+                'options': {
-            pool_ids=dict(default=[],
+                    'role': {},
-                          required=False,
+                    'usage': {},
-                          type='list'),
+                    'service_level_agreement': {},
-            consumer_type=dict(default=None,
+                    'addons': {'type': 'list'},
-                               required=False),
+                    'sync': {'type': 'bool', 'default': False}
-            consumer_name=dict(default=None,
+                }
-                               required=False),
+            }
-            consumer_id=dict(default=None,
+        },
-                             required=False),
-            force_register=dict(default=False,
-                                type='bool'),
-            server_proxy_hostname=dict(default=None,
-                                       required=False),
-            server_proxy_port=dict(default=None,
-                                   required=False),
-            server_proxy_user=dict(default=None,
-                                   required=False),
-            server_proxy_password=dict(default=None,
-                                       required=False,
-                                       no_log=True),
-            release=dict(default=None, required=False)
-        ),
         required_together=[['username', 'password'],
                            ['server_proxy_hostname', 'server_proxy_port'],
                            ['server_proxy_user', 'server_proxy_password']],
-
         mutually_exclusive=[['activationkey', 'username'],
                             ['activationkey', 'consumer_id'],
                             ['activationkey', 'environment'],
                             ['activationkey', 'autosubscribe'],
                             ['force', 'consumer_id'],
                             ['pool', 'pool_ids']],
-
         required_if=[['state', 'present', ['username', 'activationkey'], True]],
     )
 
@@ -745,15 +850,28 @@ def main():
     server_proxy_user = module.params['server_proxy_user']
     server_proxy_password = module.params['server_proxy_password']
     release = module.params['release']
+    syspurpose = module.params['syspurpose']
 
     global SUBMAN_CMD
     SUBMAN_CMD = module.get_bin_path('subscription-manager', True)
 
+    syspurpose_changed = False
+    if syspurpose is not None:
+        try:
+            syspurpose_changed = SysPurpose().update_syspurpose(syspurpose)
+        except Exception as err:
+            module.fail_json(msg="Failed to update syspurpose attributes: %s" % to_native(err))
+
     # Ensure system is registered
     if state == 'present':
 
         # Register system
         if rhsm.is_registered and not force_register:
+            if syspurpose and 'sync' in syspurpose and syspurpose['sync'] is True:
+                try:
+                    rhsm.sync_syspurpose()
+                except Exception as e:
+                    module.fail_json(msg="Failed to synchronize syspurpose attributes: %s" % to_native(e))
             if pool != '^$' or pool_ids:
                 try:
                     if pool_ids:
@@ -765,7 +883,10 @@ def main():
                 else:
                     module.exit_json(**result)
             else:
-                module.exit_json(changed=False, msg="System already registered.")
+                if syspurpose_changed is True:
+                    module.exit_json(changed=True, msg="Syspurpose attributes changed.")
+                else:
+                    module.exit_json(changed=False, msg="System already registered.")
         else:
             try:
                 rhsm.enable()
@@ -774,6 +895,8 @@ def main():
                               consumer_type, consumer_name, consumer_id, force_register,
                               environment, rhsm_baseurl, server_insecure, server_hostname,
                               server_proxy_hostname, server_proxy_port, server_proxy_user, server_proxy_password, release)
+                if syspurpose and 'sync' in syspurpose and syspurpose['sync'] is True:
+                    rhsm.sync_syspurpose()
                 if pool_ids:
                     subscribed_pool_ids = rhsm.subscribe_by_pool_ids(pool_ids)
                 elif pool != '^$':
@@ -786,6 +909,7 @@ def main():
                 module.exit_json(changed=True,
                                  msg="System successfully registered to '%s'." % server_hostname,
                                  subscribed_pool_ids=subscribed_pool_ids)
+
     # Ensure system is *not* registered
     if state == 'absent':
         if not rhsm.is_registered:

test/units/modules/packaging/os/test_redhat_subscription.py

@@ -12,6 +12,9 @@ from ansible.modules.packaging.os import redhat_subscription
 
 import pytest
 
+import os
+import tempfile
+
 TESTED_MODULE = redhat_subscription.__name__
 
 
@@ -21,8 +24,8 @@ def patch_redhat_subscription(mocker):
     Function used for mocking some parts of redhat_subscribtion module
     """
     mocker.patch('ansible.modules.packaging.os.redhat_subscription.RegistrationBase.REDHAT_REPO')
-    mocker.patch('os.path.isfile', return_value=False)
+    mocker.patch('ansible.modules.packaging.os.redhat_subscription.isfile', return_value=False)
-    mocker.patch('os.unlink', return_value=True)
+    mocker.patch('ansible.modules.packaging.os.redhat_subscription.unlink', return_value=True)
     mocker.patch('ansible.modules.packaging.os.redhat_subscription.AnsibleModule.get_bin_path',
                  return_value='/testbin/subscription-manager')
 
@@ -811,7 +814,7 @@ Entitlement Type:    Physical
             ],
             'changed': True,
         }
-    ],
+    ]
 ]
 
 
@@ -849,3 +852,376 @@ def test_redhat_subscribtion(mocker, capfd, patch_redhat_subscription, testcase)
         call_args_list = [(item[0][0], item[1]) for item in basic.AnsibleModule.run_command.call_args_list]
         expected_call_args_list = [(item[0], item[1]) for item in testcase['run_command.calls']]
         assert call_args_list == expected_call_args_list
+
+
+SYSPURPOSE_TEST_CASES = [
+    # Test setting syspurpose attributes (system is already registered)
+    # and synchronization with candlepin server
+    [
+        {
+            'state': 'present',
+            'server_hostname': 'subscription.rhsm.redhat.com',
+            'username': 'admin',
+            'password': 'admin',
+            'org_id': 'admin',
+            'syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Production',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+                'sync': True
+            }
+        },
+        {
+            'id': 'test_setting_syspurpose_attributes',
+            'existing_syspurpose': {},
+            'expected_syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Production',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+            },
+            'run_command.calls': [
+                (
+                    ['/testbin/subscription-manager', 'identity'],
+                    {'check_rc': False},
+                    (0, 'system identity: b26df632-25ed-4452-8f89-0308bfd167cb', '')
+                ),
+                (
+                    ['/testbin/subscription-manager', 'status'],
+                    {'check_rc': False},
+                    (0, '''
++-------------------------------------------+
+   System Status Details
++-------------------------------------------+
+Overall Status: Current
+
+System Purpose Status: Matched
+''', '')
+                )
+            ],
+            'changed': True,
+            'msg': 'Syspurpose attributes changed.'
+        }
+    ],
+    # Test setting unspupported attributes
+    [
+        {
+            'state': 'present',
+            'server_hostname': 'subscription.rhsm.redhat.com',
+            'username': 'admin',
+            'password': 'admin',
+            'org_id': 'admin',
+            'syspurpose': {
+                'foo': 'Bar',
+                'role': 'AwesomeOS',
+                'usage': 'Production',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+                'sync': True
+            }
+        },
+        {
+            'id': 'test_setting_syspurpose_wrong_attributes',
+            'existing_syspurpose': {},
+            'expected_syspurpose': {},
+            'run_command.calls': [],
+            'failed': True
+        }
+    ],
+    # Test setting addons not a list
+    [
+        {
+            'state': 'present',
+            'server_hostname': 'subscription.rhsm.redhat.com',
+            'username': 'admin',
+            'password': 'admin',
+            'org_id': 'admin',
+            'syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Production',
+                'service_level_agreement': 'Premium',
+                'addons': 'ADDON1',
+                'sync': True
+            }
+        },
+        {
+            'id': 'test_setting_syspurpose_addons_not_list',
+            'existing_syspurpose': {},
+            'expected_syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Production',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1']
+            },
+            'run_command.calls': [
+                (
+                    ['/testbin/subscription-manager', 'identity'],
+                    {'check_rc': False},
+                    (0, 'system identity: b26df632-25ed-4452-8f89-0308bfd167cb', '')
+                ),
+                (
+                    ['/testbin/subscription-manager', 'status'],
+                    {'check_rc': False},
+                    (0, '''
++-------------------------------------------+
+   System Status Details
++-------------------------------------------+
+Overall Status: Current
+
+System Purpose Status: Matched
+''', '')
+                )
+            ],
+            'changed': True,
+            'msg': 'Syspurpose attributes changed.'
+        }
+    ],
+    # Test setting syspurpose attributes (system is already registered)
+    # without synchronization with candlepin server. Some syspurpose attributes were set
+    # in the past
+    [
+        {
+            'state': 'present',
+            'server_hostname': 'subscription.rhsm.redhat.com',
+            'username': 'admin',
+            'password': 'admin',
+            'org_id': 'admin',
+            'syspurpose': {
+                'role': 'AwesomeOS',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+                'sync': False
+            }
+        },
+        {
+            'id': 'test_changing_syspurpose_attributes',
+            'existing_syspurpose': {
+                'role': 'CoolOS',
+                'usage': 'Production',
+                'service_level_agreement': 'Super',
+                'addons': [],
+                'foo': 'bar'
+            },
+            'expected_syspurpose': {
+                'role': 'AwesomeOS',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+                'foo': 'bar'
+            },
+            'run_command.calls': [
+                (
+                    ['/testbin/subscription-manager', 'identity'],
+                    {'check_rc': False},
+                    (0, 'system identity: b26df632-25ed-4452-8f89-0308bfd167cb', '')
+                ),
+            ],
+            'changed': True,
+            'msg': 'Syspurpose attributes changed.'
+        }
+    ],
+    # Test trying to set syspurpose attributes (system is already registered)
+    # without synchronization with candlepin server. Some syspurpose attributes were set
+    # in the past. Syspurpose attributes are same as before
+    [
+        {
+            'state': 'present',
+            'server_hostname': 'subscription.rhsm.redhat.com',
+            'username': 'admin',
+            'password': 'admin',
+            'org_id': 'admin',
+            'syspurpose': {
+                'role': 'AwesomeOS',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+                'sync': False
+            }
+        },
+        {
+            'id': 'test_not_changing_syspurpose_attributes',
+            'existing_syspurpose': {
+                'role': 'AwesomeOS',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+            },
+            'expected_syspurpose': {
+                'role': 'AwesomeOS',
+                'service_level_agreement': 'Premium',
+                'addons': ['ADDON1', 'ADDON2'],
+            },
+            'run_command.calls': [
+                (
+                    ['/testbin/subscription-manager', 'identity'],
+                    {'check_rc': False},
+                    (0, 'system identity: b26df632-25ed-4452-8f89-0308bfd167cb', '')
+                ),
+            ],
+            'changed': False,
+            'msg': 'System already registered.'
+        }
+    ],
+    # Test of registration using username and password with auto-attach option, when
+    # syspurpose attributes are set
+    [
+        {
+            'state': 'present',
+            'username': 'admin',
+            'password': 'admin',
+            'org_id': 'admin',
+            'auto_attach': 'true',
+            'syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Testing',
+                'service_level_agreement': 'Super',
+                'addons': ['ADDON1'],
+                'sync': False
+            },
+        },
+        {
+            'id': 'test_registeration_username_password_auto_attach_syspurpose',
+            'existing_syspurpose': None,
+            'expected_syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Testing',
+                'service_level_agreement': 'Super',
+                'addons': ['ADDON1'],
+            },
+            'run_command.calls': [
+                (
+                    ['/testbin/subscription-manager', 'identity'],
+                    {'check_rc': False},
+                    (1, 'This system is not yet registered.', '')
+                ),
+                (
+                    [
+                        '/testbin/subscription-manager',
+                        'register',
+                        '--org', 'admin',
+                        '--auto-attach',
+                        '--username', 'admin',
+                        '--password', 'admin'
+                    ],
+                    {'check_rc': True, 'expand_user_and_vars': False},
+                    (0, '', '')
+                )
+            ],
+            'changed': True,
+            'msg': "System successfully registered to 'None'."
+        }
+    ],
+    # Test of registration using username and password with auto-attach option, when
+    # syspurpose attributes are set. Syspurpose attributes are also synchronized
+    # in this case
+    [
+        {
+            'state': 'present',
+            'username': 'admin',
+            'password': 'admin',
+            'org_id': 'admin',
+            'auto_attach': 'true',
+            'syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Testing',
+                'service_level_agreement': 'Super',
+                'addons': ['ADDON1'],
+                'sync': True
+            },
+        },
+        {
+            'id': 'test_registeration_username_password_auto_attach_syspurpose_sync',
+            'existing_syspurpose': None,
+            'expected_syspurpose': {
+                'role': 'AwesomeOS',
+                'usage': 'Testing',
+                'service_level_agreement': 'Super',
+                'addons': ['ADDON1'],
+            },
+            'run_command.calls': [
+                (
+                    ['/testbin/subscription-manager', 'identity'],
+                    {'check_rc': False},
+                    (1, 'This system is not yet registered.', '')
+                ),
+                (
+                    [
+                        '/testbin/subscription-manager',
+                        'register',
+                        '--org', 'admin',
+                        '--auto-attach',
+                        '--username', 'admin',
+                        '--password', 'admin'
+                    ],
+                    {'check_rc': True, 'expand_user_and_vars': False},
+                    (0, '', '')
+                ),
+                (
+                    ['/testbin/subscription-manager', 'status'],
+                    {'check_rc': False},
+                    (0, '''
++-------------------------------------------+
+   System Status Details
++-------------------------------------------+
+Overall Status: Current
+
+System Purpose Status: Matched
+''', '')
+                )
+            ],
+            'changed': True,
+            'msg': "System successfully registered to 'None'."
+        }
+    ],
+]
+
+
+SYSPURPOSE_TEST_CASES_IDS = [item[1]['id'] for item in SYSPURPOSE_TEST_CASES]
+
+
+@pytest.mark.parametrize('patch_ansible_module, testcase', SYSPURPOSE_TEST_CASES, ids=SYSPURPOSE_TEST_CASES_IDS, indirect=['patch_ansible_module'])
+@pytest.mark.usefixtures('patch_ansible_module')
+def test_redhat_subscribtion_syspurpose(mocker, capfd, patch_redhat_subscription, patch_ansible_module, testcase, tmpdir):
+    """
+    Run unit tests for test cases listen in SYSPURPOSE_TEST_CASES (syspurpose specific cases)
+    """
+
+    # Mock function used for running commands first
+    call_results = [item[2] for item in testcase['run_command.calls']]
+    mock_run_command = mocker.patch.object(
+        basic.AnsibleModule,
+        'run_command',
+        side_effect=call_results)
+
+    mock_syspurpose_file = tmpdir.mkdir("syspurpose").join("syspurpose.json")
+    # When there there are some existing syspurpose attributes specified, then
+    # write them to the file first
+    if testcase['existing_syspurpose'] is not None:
+        mock_syspurpose_file.write(json.dumps(testcase['existing_syspurpose']))
+    else:
+        mock_syspurpose_file.write("{}")
+
+    redhat_subscription.SysPurpose.SYSPURPOSE_FILE_PATH = str(mock_syspurpose_file)
+
+    # Try to run test case
+    with pytest.raises(SystemExit):
+        redhat_subscription.main()
+
+    out, err = capfd.readouterr()
+    results = json.loads(out)
+
+    if 'failed' in testcase:
+        assert results['failed'] == testcase['failed']
+    else:
+        assert 'changed' in results
+        assert results['changed'] == testcase['changed']
+        if 'msg' in results:
+            assert results['msg'] == testcase['msg']
+
+    mock_file_content = mock_syspurpose_file.read_text("utf-8")
+    current_syspurpose = json.loads(mock_file_content)
+    assert current_syspurpose == testcase['expected_syspurpose']
+
+    assert basic.AnsibleModule.run_command.call_count == len(testcase['run_command.calls'])
+    if basic.AnsibleModule.run_command.call_count:
+        call_args_list = [(item[0][0], item[1]) for item in basic.AnsibleModule.run_command.call_args_list]
+        expected_call_args_list = [(item[0], item[1]) for item in testcase['run_command.calls']]
+        assert call_args_list == expected_call_args_list