diff --git a/plugins/module_utils/k8s/client.py b/plugins/module_utils/k8s/client.py
index 0cef6c50..39f751cc 100644
--- a/plugins/module_utils/k8s/client.py
+++ b/plugins/module_utils/k8s/client.py
@@ -27,9 +27,15 @@ except ImportError:
 
 try:
     import kubernetes
+    from kubernetes.dynamic.exceptions import (
+        ResourceNotFoundError,
+        ResourceNotUniqueError,
+    )
+    from kubernetes.dynamic.resource import Resource
 except ImportError:
-    # Handled in module setup
-    pass
+    # kubernetes import error is handled in module setup
+    # This is defined only for the sake of Ansible's checked import requirement
+    Resource = Any  # type: ignore
 
 try:
     import urllib3
@@ -198,6 +204,21 @@ class K8SClient:
     def resources(self) -> List[Any]:
         return self.client.resources
 
+    def resource(self, kind: str, api_version: str) -> Resource:
+        """Fetch a kubernetes client resource.
+
+        This will attempt to find a kubernetes resource trying, in order, kind,
+        name, singular_name and short_names.
+        """
+        for attribute in ["kind", "name", "singular_name"]:
+            try:
+                return self.client.resources.get(
+                    **{"api_version": api_version, attribute: kind}
+                )
+            except (ResourceNotFoundError, ResourceNotUniqueError):
+                pass
+        return self.client.resources.get(api_version=api_version, short_names=[kind])
+
     def _ensure_dry_run(self, params: Dict) -> Dict:
         if self.dry_run:
             params["dry_run"] = True
diff --git a/plugins/module_utils/k8s/exceptions.py b/plugins/module_utils/k8s/exceptions.py
index 38c52b8c..f3a82c6b 100644
--- a/plugins/module_utils/k8s/exceptions.py
+++ b/plugins/module_utils/k8s/exceptions.py
@@ -7,4 +7,6 @@ class CoreException(Exception):
 
 
 class ResourceTimeout(CoreException):
-    pass
+    def __init__(self, message="", result=None):
+        self.result = result or {}
+        super().__init__(message)
diff --git a/plugins/module_utils/k8s/runner.py b/plugins/module_utils/k8s/runner.py
index bce2f7ff..c1fe24c4 100644
--- a/plugins/module_utils/k8s/runner.py
+++ b/plugins/module_utils/k8s/runner.py
@@ -13,10 +13,13 @@ from ansible_collections.kubernetes.core.plugins.module_utils.k8s.resource impor
 )
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import (
     K8sService,
+    diff_objects,
 )
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.exceptions import (
     CoreException,
+    ResourceTimeout,
 )
+from ansible_collections.kubernetes.core.plugins.module_utils.k8s.waiter import exists
 from ansible_collections.kubernetes.core.plugins.module_utils.selector import (
     LabelSelectorFilter,
 )
@@ -49,7 +52,7 @@ def run_module(module) -> None:
     definitions = create_definitions(module.params)
 
     for definition in definitions:
-        result = {"changed": False, "result": {}, "warnings": []}
+        result = {"changed": False, "result": {}}
         warnings = []
 
         if module.params.get("validate") is not None:
@@ -58,18 +61,21 @@ def run_module(module) -> None:
         try:
             result = perform_action(svc, definition, module.params)
         except CoreException as e:
-            msg = to_native(e)
+            try:
+                error = e.result
+            except AttributeError:
+                error = {}
+            error["msg"] = to_native(e)
             if warnings:
-                msg += "\n" + "\n    ".join(warnings)
+                error.setdefault("warnings", []).extend(warnings)
+
             if module.params.get("continue_on_error"):
-                result["error"] = {"msg": msg}
+                result["error"] = error
             else:
-                module.fail_json(msg=msg)
+                module.fail_json(**error)
 
         if warnings:
-            result.setdefault("warnings", [])
-            result["warnings"] += warnings
-
+            result.setdefault("warnings", []).extend(warnings)
         changed |= result["changed"]
         results.append(result)
 
@@ -86,7 +92,9 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
     state = params.get("state", None)
     kind = definition.get("kind")
     api_version = definition.get("apiVersion")
+
     result = {"changed": False, "result": {}}
+    instance = {}
 
     resource = svc.find_resource(kind, api_version, fail=True)
     definition["kind"] = resource.kind
@@ -94,8 +102,10 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
     existing = svc.retrieve(resource, definition)
 
     if state == "absent":
-        result = svc.delete(resource, definition, existing)
+        instance = svc.delete(resource, definition, existing)
         result["method"] = "delete"
+        if exists(existing):
+            result["changed"] = True
     else:
         if label_selectors:
             filter_selector = LabelSelectorFilter(label_selectors)
@@ -110,7 +120,7 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
                 return result
 
         if params.get("apply"):
-            result = svc.apply(resource, definition, existing)
+            instance = svc.apply(resource, definition, existing)
             result["method"] = "apply"
         elif not existing:
             if state == "patched":
@@ -121,13 +131,51 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
                     )
                 )
                 return result
-            result = svc.create(resource, definition)
+            instance = svc.create(resource, definition)
             result["method"] = "create"
+            result["changed"] = True
         elif params.get("force", False):
-            result = svc.replace(resource, definition, existing)
+            instance = svc.replace(resource, definition, existing)
             result["method"] = "replace"
         else:
-            result = svc.update(resource, definition, existing)
+            instance = svc.update(resource, definition, existing)
             result["method"] = "update"
 
+    # If needed, wait and/or create diff
+    success = True
+
+    if result["method"] == "delete":
+        # wait logic is a bit different for delete as `instance` may be a status object
+        if params.get("wait") and not svc.module.check_mode:
+            success, waited, duration = svc.wait(resource, definition)
+            result["duration"] = duration
+    else:
+        if params.get("wait") and not svc.module.check_mode:
+            success, instance, duration = svc.wait(resource, instance)
+            result["duration"] = duration
+
+    if result["method"] not in ("create", "delete"):
+        if existing:
+            existing = existing.to_dict()
+        else:
+            existing = {}
+        match, diffs = diff_objects(existing, instance)
+        if match and diffs:
+            result.setdefault("warnings", []).append(
+                "No meaningful diff was generated, but the API may not be idempotent "
+                "(only metadata.generation or metadata.resourceVersion were changed)"
+            )
+        result["changed"] = not match
+        if svc.module._diff:
+            result["diff"] = diffs
+
+    result["result"] = instance
+    if not success:
+        raise ResourceTimeout(
+            '"{0}" "{1}": Timed out waiting on resource'.format(
+                definition["kind"], origin_name
+            ),
+            result,
+        )
+
     return result
diff --git a/plugins/module_utils/k8s/service.py b/plugins/module_utils/k8s/service.py
index 6a23d770..c2f51fc8 100644
--- a/plugins/module_utils/k8s/service.py
+++ b/plugins/module_utils/k8s/service.py
@@ -16,7 +16,6 @@ from ansible_collections.kubernetes.core.plugins.module_utils.k8s.waiter import
 
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.exceptions import (
     CoreException,
-    ResourceTimeout,
 )
 
 from ansible.module_utils.common.dict_transformations import dict_merge
@@ -78,18 +77,9 @@ class K8sService:
 
     def find_resource(
         self, kind: str, api_version: str, fail: bool = False
-    ) -> Optional[ResourceInstance]:
-        for attribute in ["kind", "name", "singular_name"]:
-            try:
-                return self.client.resources.get(
-                    **{"api_version": api_version, attribute: kind}
-                )
-            except (ResourceNotFoundError, ResourceNotUniqueError):
-                pass
+    ) -> Optional[Resource]:
         try:
-            return self.client.resources.get(
-                api_version=api_version, short_names=[kind]
-            )
+            return self.client.resource(kind, api_version)
         except (ResourceNotFoundError, ResourceNotUniqueError):
             if fail:
                 raise CoreException(
@@ -97,6 +87,32 @@ class K8sService:
                     % (api_version, kind)
                 )
 
+    def wait(
+        self, resource: Resource, instance: Dict
+    ) -> Tuple[bool, Optional[Dict], int]:
+        wait_sleep = self.module.params.get("wait_sleep")
+        wait_timeout = self.module.params.get("wait_timeout")
+        wait_condition = None
+        if self.module.params.get("wait_condition") and self.module.params[
+            "wait_condition"
+        ].get("type"):
+            wait_condition = self.module.params["wait_condition"]
+        state = "present"
+        if self.module.params.get("state") == "absent":
+            state = "absent"
+        label_selectors = self.module.params.get("label_selectors")
+
+        waiter = get_waiter(
+            self.client, resource, condition=wait_condition, state=state
+        )
+        return waiter.wait(
+            timeout=wait_timeout,
+            sleep=wait_sleep,
+            name=instance["metadata"].get("name"),
+            namespace=instance["metadata"].get("namespace"),
+            label_selectors=label_selectors,
+        )
+
     def create_project_request(self, definition: Dict) -> Dict:
         definition["kind"] = "ProjectRequest"
         results = {"changed": False, "result": {}}
@@ -116,35 +132,6 @@ class K8sService:
 
         return results
 
-    def diff_objects(self, existing: Dict, new: Dict) -> Tuple[bool, Dict]:
-        result: Dict = dict()
-        diff = recursive_diff(existing, new)
-        if not diff:
-            return True, result
-
-        result["before"] = diff[0]
-        result["after"] = diff[1]
-
-        # If only metadata.generation and metadata.resourceVersion changed, ignore it
-        ignored_keys = set(["generation", "resourceVersion"])
-
-        if list(result["after"].keys()) != ["metadata"] or list(
-            result["before"].keys()
-        ) != ["metadata"]:
-            return False, result
-
-        if not set(result["after"]["metadata"].keys()).issubset(ignored_keys):
-            return False, result
-        if not set(result["before"]["metadata"].keys()).issubset(ignored_keys):
-            return False, result
-
-        if hasattr(self.module, "warn"):
-            self.module.warn(
-                "No meaningful diff was generated, but the API may not be idempotent (only metadata.generation or metadata.resourceVersion were changed)"
-            )
-
-        return True, result
-
     def patch_resource(
         self,
         resource: Resource,
@@ -305,17 +292,8 @@ class K8sService:
         return result
 
     def create(self, resource: Resource, definition: Dict) -> Dict:
-        origin_name = definition["metadata"].get("name")
         namespace = definition["metadata"].get("namespace")
         name = definition["metadata"].get("name")
-        wait = self.module.params.get("wait")
-        wait_sleep = self.module.params.get("wait_sleep")
-        wait_timeout = self.module.params.get("wait_timeout")
-        wait_condition = None
-        if self.module.params.get("wait_condition") and self.module.params[
-            "wait_condition"
-        ].get("type"):
-            wait_condition = self.module.params["wait_condition"]
         results = {"changed": False, "result": {}}
 
         if self.module.check_mode and not self.client.dry_run:
@@ -343,30 +321,7 @@ class K8sService:
                 reason = e.body if hasattr(e, "body") else e
                 msg = "Failed to create object: {0}".format(reason)
                 raise CoreException(msg) from e
-
-        success = True
-        results["result"] = k8s_obj
-
-        if wait and not self.module.check_mode:
-            waiter = get_waiter(self.client, resource, condition=wait_condition)
-            success, results["result"], results["duration"] = waiter.wait(
-                timeout=wait_timeout,
-                sleep=wait_sleep,
-                name=k8s_obj["metadata"]["name"],
-                namespace=namespace,
-            )
-
-        results["changed"] = True
-
-        if not success:
-            raise ResourceTimeout(
-                '"{0}" "{1}": Resource creation timed out'.format(
-                    definition["kind"], origin_name
-                ),
-                results,
-            )
-
-        return results
+        return k8s_obj
 
     def apply(
         self,
@@ -374,89 +329,34 @@ class K8sService:
         definition: Dict,
         existing: Optional[ResourceInstance] = None,
     ) -> Dict:
-        apply = self.module.params.get("apply", False)
-        origin_name = definition["metadata"].get("name")
-        name = definition["metadata"].get("name")
         namespace = definition["metadata"].get("namespace")
-        wait = self.module.params.get("wait")
-        wait_sleep = self.module.params.get("wait_sleep")
-        wait_condition = None
-        if self.module.params.get("wait_condition") and self.module.params[
-            "wait_condition"
-        ].get("type"):
-            wait_condition = self.module.params["wait_condition"]
-        wait_timeout = self.module.params.get("wait_timeout")
-        results = {"changed": False, "result": {}}
-
-        if apply:
-            if self.module.check_mode and not self.client.dry_run:
-                ignored, patch = apply_object(resource, _encode_stringdata(definition))
-                if existing:
-                    k8s_obj = dict_merge(existing.to_dict(), patch)
-                else:
-                    k8s_obj = patch
-            else:
-                try:
-                    params = {}
-                    if self.module.check_mode:
-                        params["dry_run"] = "All"
-                    k8s_obj = self.client.apply(
-                        resource, definition, namespace=namespace, **params
-                    ).to_dict()
-                except Exception as e:
-                    reason = e.body if hasattr(e, "body") else e
-                    msg = "Failed to apply object: {0}".format(reason)
-                    raise CoreException(msg) from e
-
-            success = True
-            results["result"] = k8s_obj
-
-            if wait and not self.module.check_mode:
-                waiter = get_waiter(self.client, resource, condition=wait_condition)
-                success, results["result"], results["duration"] = waiter.wait(
-                    timeout=wait_timeout,
-                    sleep=wait_sleep,
-                    name=name,
-                    namespace=namespace,
-                )
 
+        if self.module.check_mode and not self.client.dry_run:
+            ignored, patch = apply_object(resource, _encode_stringdata(definition))
             if existing:
-                existing = existing.to_dict()
+                k8s_obj = dict_merge(existing.to_dict(), patch)
             else:
-                existing = {}
-
-            match, diffs = self.diff_objects(existing, results["result"])
-            results["changed"] = not match
-
-            if self.module._diff:
-                results["diff"] = diffs
-
-            if not success:
-                raise ResourceTimeout(
-                    '"{0}" "{1}": Resource apply timed out'.format(
-                        definition["kind"], origin_name
-                    ),
-                    results,
-                )
-
-        return results
+                k8s_obj = patch
+        else:
+            try:
+                params = {}
+                if self.module.check_mode:
+                    params["dry_run"] = "All"
+                k8s_obj = self.client.apply(
+                    resource, definition, namespace=namespace, **params
+                ).to_dict()
+            except Exception as e:
+                reason = e.body if hasattr(e, "body") else e
+                msg = "Failed to apply object: {0}".format(reason)
+                raise CoreException(msg) from e
+        return k8s_obj
 
     def replace(
         self, resource: Resource, definition: Dict, existing: ResourceInstance,
     ) -> Dict:
         append_hash = self.module.params.get("append_hash", False)
         name = definition["metadata"].get("name")
-        origin_name = definition["metadata"].get("name")
         namespace = definition["metadata"].get("namespace")
-        wait = self.module.params.get("wait")
-        wait_sleep = self.module.params.get("wait_sleep")
-        wait_timeout = self.module.params.get("wait_timeout")
-        wait_condition = None
-        if self.module.params.get("wait_condition") and self.module.params[
-            "wait_condition"
-        ].get("type"):
-            wait_condition = self.module.params["wait_condition"]
-        results = {"changed": False, "result": {}}
 
         if self.module.check_mode and not self.module.client.dry_run:
             k8s_obj = _encode_stringdata(definition)
@@ -477,47 +377,13 @@ class K8sService:
                 reason = e.body if hasattr(e, "body") else e
                 msg = "Failed to replace object: {0}".format(reason)
                 raise CoreException(msg) from e
-
-        match, diffs = self.diff_objects(existing.to_dict(), k8s_obj)
-        success = True
-        results["result"] = k8s_obj
-
-        if wait and not self.module.check_mode:
-            waiter = get_waiter(self.client, resource, condition=wait_condition)
-            success, results["result"], results["duration"] = waiter.wait(
-                timeout=wait_timeout, sleep=wait_sleep, name=name, namespace=namespace,
-            )
-        match, diffs = self.diff_objects(existing.to_dict(), results["result"])
-        results["changed"] = not match
-
-        if self.module._diff:
-            results["diff"] = diffs
-
-        if not success:
-            raise ResourceTimeout(
-                '"{0}" "{1}": Resource replacement timed out'.format(
-                    definition["kind"], origin_name
-                ),
-                results,
-            )
-
-        return results
+        return k8s_obj
 
     def update(
         self, resource: Resource, definition: Dict, existing: ResourceInstance
     ) -> Dict:
         name = definition["metadata"].get("name")
-        origin_name = definition["metadata"].get("name")
         namespace = definition["metadata"].get("namespace")
-        wait = self.module.params.get("wait")
-        wait_sleep = self.module.params.get("wait_sleep")
-        wait_timeout = self.module.params.get("wait_timeout")
-        wait_condition = None
-        if self.module.params.get("wait_condition") and self.module.params[
-            "wait_condition"
-        ].get("type"):
-            wait_condition = self.module.params["wait_condition"]
-        results = {"changed": False, "result": {}}
 
         if self.module.check_mode and not self.client.dry_run:
             k8s_obj = dict_merge(existing.to_dict(), _encode_stringdata(definition))
@@ -538,31 +404,7 @@ class K8sService:
                 break
             if exception:
                 raise exception
-
-        success = True
-        results["result"] = k8s_obj
-
-        if wait and not self.module.check_mode:
-            waiter = get_waiter(self.client, resource, condition=wait_condition)
-            success, results["result"], results["duration"] = waiter.wait(
-                timeout=wait_timeout, sleep=wait_sleep, name=name, namespace=namespace,
-            )
-
-        match, diffs = self.diff_objects(existing.to_dict(), results["result"])
-        results["changed"] = not match
-
-        if self.module._diff:
-            results["diff"] = diffs
-
-        if not success:
-            raise ResourceTimeout(
-                '"{0}" "{1}": Resource update timed out'.format(
-                    definition["kind"], origin_name
-                ),
-                results,
-            )
-
-        return results
+        return k8s_obj
 
     def delete(
         self,
@@ -572,72 +414,65 @@ class K8sService:
     ) -> Dict:
         delete_options = self.module.params.get("delete_options")
         label_selectors = self.module.params.get("label_selectors")
-        origin_name = definition["metadata"].get("name")
         name = definition["metadata"].get("name")
         namespace = definition["metadata"].get("namespace")
-        wait = self.module.params.get("wait")
-        wait_sleep = self.module.params.get("wait_sleep")
-        wait_timeout = self.module.params.get("wait_timeout")
-        results = {"changed": False, "result": {}}
         params = {}
 
-        def _empty_resource_list() -> bool:
-            if existing and existing.kind.endswith("List"):
-                return existing.items == []
-            return False
+        if not exists(existing):
+            return {}
 
-        if not existing or _empty_resource_list():
-            # The object already does not exist
-            return results
-        else:
-            # Delete the object
-            results["changed"] = True
-            if self.module.check_mode and not self.client.dry_run:
-                return results
-            else:
-                if name:
-                    params["name"] = name
+        # Delete the object
+        if self.module.check_mode and not self.client.dry_run:
+            return {}
 
-                if namespace:
-                    params["namespace"] = namespace
+        if name:
+            params["name"] = name
 
-                if label_selectors:
-                    params["label_selector"] = ",".join(label_selectors)
+        if namespace:
+            params["namespace"] = namespace
 
-                if delete_options:
-                    body = {
-                        "apiVersion": "v1",
-                        "kind": "DeleteOptions",
-                    }
-                    body.update(delete_options)
-                    params["body"] = body
+        if label_selectors:
+            params["label_selector"] = ",".join(label_selectors)
 
-                if self.module.check_mode:
-                    params["dry_run"] = "All"
-                try:
-                    k8s_obj = self.client.delete(resource, **params)
-                    results["result"] = k8s_obj.to_dict()
-                except Exception as e:
-                    reason = e.body if hasattr(e, "body") else e
-                    msg = "Failed to delete object: {0}".format(reason)
-                    raise CoreException(msg) from e
+        if delete_options:
+            body = {
+                "apiVersion": "v1",
+                "kind": "DeleteOptions",
+            }
+            body.update(delete_options)
+            params["body"] = body
 
-                if wait and not self.module.check_mode:
-                    waiter = get_waiter(self.client, resource, state="absent")
-                    success, resource, duration = waiter.wait(
-                        timeout=wait_timeout,
-                        sleep=wait_sleep,
-                        name=name,
-                        namespace=namespace,
-                        label_selectors=label_selectors,
-                    )
-                    results["duration"] = duration
-                    if not success:
-                        raise ResourceTimeout(
-                            '"{0}" "{1}": Resource deletion timed out'.format(
-                                definition["kind"], origin_name
-                            ),
-                            results,
-                        )
+        if self.module.check_mode:
+            params["dry_run"] = "All"
+        try:
+            k8s_obj = self.client.delete(resource, **params).to_dict()
+        except Exception as e:
+            reason = e.body if hasattr(e, "body") else e
+            msg = "Failed to delete object: {0}".format(reason)
+            raise CoreException(msg) from e
+        return k8s_obj
 
-                return results
+
+def diff_objects(existing: Dict, new: Dict) -> Tuple[bool, Dict]:
+    result = {}
+    diff = recursive_diff(existing, new)
+    if not diff:
+        return True, result
+
+    result["before"] = diff[0]
+    result["after"] = diff[1]
+
+    if list(result["after"].keys()) != ["metadata"] or list(
+        result["before"].keys()
+    ) != ["metadata"]:
+        return False, result
+
+    # If only metadata.generation and metadata.resourceVersion changed, ignore it
+    ignored_keys = set(["generation", "resourceVersion"])
+
+    if not set(result["after"]["metadata"].keys()).issubset(ignored_keys):
+        return False, result
+    if not set(result["before"]["metadata"].keys()).issubset(ignored_keys):
+        return False, result
+
+    return True, result
diff --git a/plugins/module_utils/k8s/waiter.py b/plugins/module_utils/k8s/waiter.py
index d52491f4..88013527 100644
--- a/plugins/module_utils/k8s/waiter.py
+++ b/plugins/module_utils/k8s/waiter.py
@@ -107,7 +107,7 @@ RESOURCE_PREDICATES = {
 
 
 def empty_list(resource: ResourceInstance) -> bool:
-    return resource.kind.endswith("List") and not resource.get("items")
+    return resource["kind"].endswith("List") and not resource.get("items")
 
 
 def clock(total: int, interval: int) -> Iterator[int]:
@@ -134,7 +134,7 @@ class Waiter:
         namespace: Optional[str] = None,
         label_selectors: Optional[List[str]] = None,
         field_selectors: Optional[List[str]] = None,
-    ) -> Tuple[bool, Optional[Dict], int]:
+    ) -> Tuple[bool, Dict, int]:
         params = {}
 
         if name:
@@ -149,7 +149,7 @@ class Waiter:
         if field_selectors:
             params["field_selector"] = ",".join(field_selectors)
 
-        instance: Optional[Dict] = None
+        instance = {}
         response = None
         elapsed = 0
         for i in clock(timeout, sleep):
diff --git a/plugins/modules/k8s_json_patch.py b/plugins/modules/k8s_json_patch.py
index eb2303ef..61595454 100644
--- a/plugins/modules/k8s_json_patch.py
+++ b/plugins/modules/k8s_json_patch.py
@@ -143,7 +143,7 @@ from ansible_collections.kubernetes.core.plugins.module_utils.k8s.core import (
     AnsibleK8SModule,
 )
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import (
-    K8sService,
+    diff_objects,
 )
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.waiter import (
     get_waiter,
@@ -194,7 +194,7 @@ def json_patch(existing, patch):
         return None, error
 
 
-def execute_module(module, svc):
+def execute_module(module, client):
     kind = module.params.get("kind")
     api_version = module.params.get("api_version")
     name = module.params.get("name")
@@ -213,8 +213,7 @@ def execute_module(module, svc):
     def build_error_msg(kind, name, msg):
         return "%s %s: %s" % (kind, name, msg)
 
-    client = svc.client
-    resource = svc.find_resource(kind, api_version, fail=True)
+    resource = client.resource(kind, api_version)
 
     try:
         existing = client.get(resource, name=name, namespace=namespace)
@@ -265,7 +264,7 @@ def execute_module(module, svc):
         success, result["result"], result["duration"] = waiter.wait(
             wait_timeout, wait_sleep, name, namespace
         )
-    match, diffs = svc.diff_objects(existing.to_dict(), obj)
+    match, diffs = diff_objects(existing.to_dict(), obj)
     result["changed"] = not match
     if module._diff:
         result["diff"] = diffs
@@ -285,8 +284,7 @@ def main():
         module_class=AnsibleModule, argument_spec=args, supports_check_mode=True
     )
     client = get_api_client(module)
-    svc = K8sService(client, module)
-    execute_module(module, svc)
+    execute_module(module, client)
 
 
 if __name__ == "__main__":
diff --git a/plugins/modules/k8s_scale.py b/plugins/modules/k8s_scale.py
index 5e5cc92e..fb370d06 100644
--- a/plugins/modules/k8s_scale.py
+++ b/plugins/modules/k8s_scale.py
@@ -170,7 +170,7 @@ from ansible_collections.kubernetes.core.plugins.module_utils.k8s.exceptions imp
     ResourceTimeout,
 )
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import (
-    K8sService,
+    diff_objects,
 )
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.waiter import (
     get_waiter,
@@ -214,8 +214,7 @@ def execute_module(client, module):
     if wait:
         return_attributes["duration"] = 0
 
-    svc = K8sService(client, module)
-    resource = svc.find_resource(kind, api_version, fail=True)
+    resource = client.resource(kind, api_version)
     multiple_scale = False
     try:
         existing = resource.get(
@@ -297,7 +296,7 @@ def execute_module(client, module):
                 try:
                     result = scale(
                         client,
-                        svc,
+                        module,
                         resource,
                         existing,
                         replicas,
@@ -337,9 +336,8 @@ def argspec():
 
 
 def scale(
-    client, svc, resource, existing_object, replicas, wait, wait_time, wait_sleep,
+    client, module, resource, existing_object, replicas, wait, wait_time, wait_sleep,
 ):
-    module = svc.module
     name = existing_object.metadata.name
     namespace = existing_object.metadata.namespace
     kind = existing_object.kind
@@ -361,7 +359,6 @@ def scale(
     if module.check_mode:
         k8s_obj = copy.deepcopy(existing.to_dict())
         k8s_obj["spec"]["replicas"] = replicas
-        match, diffs = svc.diff_objects(existing.to_dict(), k8s_obj)
         if wait:
             result["duration"] = 0
         result["result"] = k8s_obj
@@ -383,7 +380,7 @@ def scale(
             if not success:
                 raise ResourceTimeout("Resource scaling timed out", **result)
 
-    match, diffs = svc.diff_objects(existing.to_dict(), k8s_obj)
+    match, diffs = diff_objects(existing.to_dict(), result["result"])
     result["changed"] = not match
     if module._diff:
         result["diff"] = diffs
diff --git a/tests/unit/module_utils/test_client.py b/tests/unit/module_utils/test_client.py
index cab9c937..8265b768 100644
--- a/tests/unit/module_utils/test_client.py
+++ b/tests/unit/module_utils/test_client.py
@@ -43,7 +43,10 @@ _temp_files = []
 
 def _remove_temp_file():
     for f in _temp_files:
-        os.remove(f)
+        try:
+            os.remove(f)
+        except FileNotFoundError:
+            pass
 
 
 def _create_temp_file(content=""):
diff --git a/tests/unit/module_utils/test_runner.py b/tests/unit/module_utils/test_runner.py
index 21579c2c..45c6f29a 100644
--- a/tests/unit/module_utils/test_runner.py
+++ b/tests/unit/module_utils/test_runner.py
@@ -1,5 +1,8 @@
 import pytest
-from unittest.mock import MagicMock, call
+from copy import deepcopy
+from unittest.mock import Mock
+
+from kubernetes.dynamic.resource import ResourceInstance
 
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.runner import (
     perform_action,
@@ -24,30 +27,109 @@ definition = {
     },
 }
 
+modified_def = deepcopy(definition)
+modified_def["metadata"]["labels"]["environment"] = "testing"
+
 
 @pytest.mark.parametrize(
-    "params, expected",
+    "action, params, existing, instance, expected",
     [
-        ({"state": "absent"}, call.__setitem__("method", "delete")),
-        ({"apply": True}, call.__setitem__("method", "apply")),
-        ({"force": True}, call.__setitem__("method", "replace")),
-        ({"apply": False}, call.__setitem__("method", "update")),
-        ({}, call.__setitem__("method", "update")),
+        (
+            "delete",
+            {"state": "absent"},
+            {},
+            {},
+            {"changed": False, "method": "delete", "result": {}},
+        ),
+        (
+            "delete",
+            {"state": "absent"},
+            definition,
+            {"kind": "Status"},
+            {"changed": True, "method": "delete", "result": {"kind": "Status"}},
+        ),
+        (
+            "apply",
+            {"apply": "yes"},
+            {},
+            definition,
+            {"changed": True, "method": "apply", "result": definition},
+        ),
+        (
+            "create",
+            {"state": "patched"},
+            {},
+            {},
+            {
+                "changed": False,
+                "result": {},
+                "warnings": [
+                    "resource 'kind=Pod,name=foo' was not found but will not be created as 'state' parameter has been set to 'patched'"
+                ],
+            },
+        ),
+        (
+            "create",
+            {},
+            {},
+            definition,
+            {"changed": True, "method": "create", "result": definition},
+        ),
+        (
+            "replace",
+            {"force": "yes"},
+            definition,
+            definition,
+            {"changed": False, "method": "replace", "result": definition},
+        ),
+        (
+            "replace",
+            {"force": "yes"},
+            definition,
+            modified_def,
+            {"changed": True, "method": "replace", "result": modified_def},
+        ),
+        (
+            "update",
+            {},
+            definition,
+            definition,
+            {"changed": False, "method": "update", "result": definition},
+        ),
+        (
+            "update",
+            {},
+            definition,
+            modified_def,
+            {"changed": True, "method": "update", "result": modified_def},
+        ),
+        (
+            "create",
+            {"label_selectors": ["app=foo"]},
+            {},
+            definition,
+            {
+                "changed": False,
+                "msg": "resource 'kind=Pod,name=foo,namespace=foo' filtered by label_selectors.",
+            },
+        ),
+        (
+            "create",
+            {"label_selectors": ["app=nginx"]},
+            {},
+            definition,
+            {"changed": True, "method": "create", "result": definition},
+        ),
     ],
 )
-def test_perform_action(params, expected):
-    module = MagicMock()
-    module.params = params
+def test_perform_action(action, params, existing, instance, expected):
+    svc = Mock()
+    svc.find_resource.return_value = Mock(
+        kind=definition["kind"], group_version=definition["apiVersion"]
+    )
+    svc.retrieve.return_value = ResourceInstance(None, existing) if existing else None
+    spec = {action + ".return_value": instance}
+    svc.configure_mock(**spec)
 
-    result = perform_action(MagicMock(), definition, module.params)
-    result.assert_has_calls([expected], any_order=True)
-
-
-def test_perform_action_create():
-    spec = {"retrieve.side_effect": [{}]}
-    svc = MagicMock(**spec)
-    module = MagicMock()
-    module.params = {}
-
-    result = perform_action(svc, definition, module.params)
-    result.assert_has_calls([call.__setitem__("method", "create")], any_order=True)
+    result = perform_action(svc, definition, params)
+    assert expected.items() <= result.items()
diff --git a/tests/unit/module_utils/test_service.py b/tests/unit/module_utils/test_service.py
index 8ef836d8..a1822de6 100644
--- a/tests/unit/module_utils/test_service.py
+++ b/tests/unit/module_utils/test_service.py
@@ -5,6 +5,7 @@ from kubernetes.dynamic.resource import ResourceInstance, Resource
 
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import (
     K8sService,
+    diff_objects,
 )
 
 from kubernetes.dynamic.exceptions import NotFoundError
@@ -59,16 +60,14 @@ def mock_pod_updated_resource_instance():
 
 
 def test_diff_objects_no_diff():
-    svc = K8sService(Mock(), Mock())
-    match, diff = svc.diff_objects(pod_definition, pod_definition)
+    match, diff = diff_objects(pod_definition, pod_definition)
 
     assert match is True
     assert diff == {}
 
 
 def test_diff_objects_meta_diff():
-    svc = K8sService(Mock(), Mock())
-    match, diff = svc.diff_objects(pod_definition, pod_definition_updated)
+    match, diff = diff_objects(pod_definition, pod_definition_updated)
 
     assert match is False
     assert diff["before"] == {
@@ -98,8 +97,7 @@ def test_diff_objects_spec_diff():
             ]
         },
     }
-    svc = K8sService(Mock(), Mock())
-    match, diff = svc.diff_objects(pod_definition, pod_definition_updated)
+    match, diff = diff_objects(pod_definition, pod_definition_updated)
 
     assert match is False
     assert diff["before"]["spec"] == pod_definition["spec"]
@@ -110,7 +108,7 @@ def test_find_resource():
     mock_pod_resource = Resource(
         api_version="v1", kind="Pod", namespaced=False, preferred=True, prefix="api"
     )
-    spec = {"resources.get.side_effect": [mock_pod_resource]}
+    spec = {"resource.return_value": mock_pod_resource}
     client = Mock(**spec)
     svc = K8sService(client, Mock())
     resource = svc.find_resource("Pod", "v1")
@@ -122,38 +120,45 @@ def test_find_resource():
 def test_service_delete_existing_resource(mock_pod_resource_instance):
     spec = {"delete.side_effect": [mock_pod_resource_instance]}
     client = Mock(**spec)
-    module = Mock()
-    module.params = {}
-    module.check_mode = False
+    module = Mock(
+        params={"delete_options": {"gracePeriodSeconds": 2}}, check_mode=False
+    )
+    resource = Mock()
     svc = K8sService(client, module)
-    results = svc.delete(Mock(), pod_definition, mock_pod_resource_instance)
+    result = svc.delete(resource, pod_definition, mock_pod_resource_instance)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is True
-    assert results["result"] == pod_definition
+    assert isinstance(result, dict)
+    assert result == mock_pod_resource_instance.to_dict()
+    client.delete.assert_called_with(
+        resource,
+        name=pod_definition["metadata"]["name"],
+        namespace=pod_definition["metadata"]["namespace"],
+        body={"apiVersion": "v1", "kind": "DeleteOptions", "gracePeriodSeconds": 2},
+    )
 
 
 def test_service_delete_no_existing_resource():
     module = Mock()
     module.params = {}
     module.check_mode = False
-    svc = K8sService(Mock(), module)
-    results = svc.delete(Mock(), pod_definition)
+    client = Mock()
+    client.delete.return_value = mock_pod_resource_instance
+    svc = K8sService(client, module)
+    result = svc.delete(Mock(), pod_definition)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is False
-    assert results["result"] == {}
+    assert result == {}
+    client.delete.assert_not_called()
 
 
 def test_service_delete_existing_resource_check_mode(mock_pod_resource_instance):
-    module = Mock()
-    module.params = {"wait": False}
-    module.check_mode = True
-    svc = K8sService(Mock(), module)
-    results = svc.delete(Mock(), pod_definition, mock_pod_resource_instance)
+    module = Mock(params={}, check_mode=True)
+    client = Mock(dry_run=False)
+    client.delete.return_value = mock_pod_resource_instance
+    svc = K8sService(client, module)
+    result = svc.delete(Mock(), pod_definition, mock_pod_resource_instance)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is True
+    assert result == {}
+    client.delete.assert_not_called()
 
 
 def test_service_create_resource(mock_pod_resource_instance):
@@ -163,25 +168,20 @@ def test_service_create_resource(mock_pod_resource_instance):
     module.params = {}
     module.check_mode = False
     svc = K8sService(client, module)
-    results = svc.create(Mock(), pod_definition)
+    result = svc.create(Mock(), pod_definition)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is True
-    assert results["result"] == pod_definition
+    assert result == mock_pod_resource_instance.to_dict()
 
 
 def test_service_create_resource_check_mode():
-    client = Mock()
-    client.dry_run = False
-    module = Mock()
-    module.params = {}
-    module.check_mode = True
+    client = Mock(dry_run=False)
+    client.create.return_value = mock_pod_resource_instance
+    module = Mock(params={}, check_mode=True)
     svc = K8sService(client, module)
-    results = svc.create(Mock(), pod_definition)
+    result = svc.create(Mock(), pod_definition)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is True
-    assert results["result"] == pod_definition
+    assert result == pod_definition
+    client.create.assert_not_called()
 
 
 def test_service_retrieve_existing_resource(mock_pod_resource_instance):
@@ -227,117 +227,39 @@ def test_create_project_request():
 
 
 def test_service_apply_existing_resource(mock_pod_resource_instance):
-    spec = {"apply.side_effect": [ResourceInstance(None, pod_definition_updated)]}
-    client = Mock(**spec)
-    module = Mock()
-    module.params = {"apply": True}
-    module.check_mode = False
-    svc = K8sService(client, module)
-    results = svc.apply(Mock(), pod_definition_updated, mock_pod_resource_instance)
-
-    assert isinstance(results, dict)
-    assert results["changed"] is True
-    assert results["diff"] is not {}
-    assert results["result"] == pod_definition_updated
-
-
-def test_service_apply_existing_resource_no_diff(mock_pod_resource_instance):
     spec = {"apply.side_effect": [mock_pod_resource_instance]}
     client = Mock(**spec)
     module = Mock()
     module.params = {"apply": True}
     module.check_mode = False
     svc = K8sService(client, module)
-    results = svc.apply(Mock(), pod_definition, mock_pod_resource_instance)
+    result = svc.apply(Mock(), pod_definition_updated, mock_pod_resource_instance)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is False
-    assert results["diff"] == {}
-    assert results["result"] == pod_definition
+    assert result == mock_pod_resource_instance.to_dict()
 
 
-def test_service_apply_existing_resource_no_apply(mock_pod_resource_instance):
-    spec = {"apply.side_effect": [mock_pod_resource_instance]}
-    client = Mock(**spec)
-    module = Mock()
-    module.params = {"apply": False}
-    module.check_mode = False
-    svc = K8sService(client, module)
-    results = svc.apply(Mock(), pod_definition, mock_pod_resource_instance)
-
-    assert isinstance(results, dict)
-    assert results["changed"] is False
-    assert results["result"] == {}
-
-
-def test_service_replace_existing_resource_no_diff(mock_pod_resource_instance):
+def test_service_replace_existing_resource(mock_pod_resource_instance):
     spec = {"replace.side_effect": [mock_pod_resource_instance]}
     client = Mock(**spec)
     module = Mock()
     module.params = {}
     module.check_mode = False
     svc = K8sService(client, module)
-    results = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
+    result = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is False
-    assert results["diff"] == {}
-    assert results["result"] == pod_definition
+    assert result == mock_pod_resource_instance.to_dict()
 
 
-def test_service_replace_existing_resource(
-    mock_pod_resource_instance, mock_pod_updated_resource_instance
-):
-    spec = {"replace.side_effect": [mock_pod_updated_resource_instance]}
+def test_service_update_existing_resource(mock_pod_resource_instance):
+    spec = {"replace.side_effect": [mock_pod_resource_instance]}
     client = Mock(**spec)
     module = Mock()
     module.params = {}
     module.check_mode = False
     svc = K8sService(client, module)
-    results = svc.replace(Mock(), pod_definition_updated, mock_pod_resource_instance)
+    result = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is True
-    assert results["result"] == pod_definition_updated
-    assert results["diff"] != {}
-    assert results["diff"]["before"] is not {}
-    assert results["diff"]["after"] is not {}
-
-
-def test_service_update_existing_resource(
-    mock_pod_resource_instance, mock_pod_updated_resource_instance
-):
-    spec = {"replace.side_effect": [mock_pod_updated_resource_instance]}
-    client = Mock(**spec)
-    module = Mock()
-    module.params = {}
-    module.check_mode = False
-    svc = K8sService(client, module)
-    results = svc.replace(Mock(), pod_definition_updated, mock_pod_resource_instance)
-
-    assert isinstance(results, dict)
-    assert results["changed"] is True
-    assert results["result"] == pod_definition_updated
-    assert results["diff"] != {}
-    assert results["diff"]["before"] is not {}
-    assert results["diff"]["after"] is not {}
-
-
-def test_service_update_existing_resource_no_diff(mock_pod_updated_resource_instance):
-    spec = {"replace.side_effect": [mock_pod_updated_resource_instance]}
-    client = Mock(**spec)
-    module = Mock()
-    module.params = {}
-    module.check_mode = False
-    svc = K8sService(client, module)
-    results = svc.replace(
-        Mock(), pod_definition_updated, mock_pod_updated_resource_instance
-    )
-
-    assert isinstance(results, dict)
-    assert results["changed"] is False
-    assert results["result"] == pod_definition_updated
-    assert results["diff"] == {}
+    assert result == mock_pod_resource_instance.to_dict()
 
 
 def test_service_find(mock_pod_resource_instance):
diff --git a/tests/unit/module_utils/test_waiter.py b/tests/unit/module_utils/test_waiter.py
index b2118895..b5ce10a5 100644
--- a/tests/unit/module_utils/test_waiter.py
+++ b/tests/unit/module_utils/test_waiter.py
@@ -89,7 +89,7 @@ def test_waiter_waits_for_missing_resource():
         namespace=RESOURCES[0]["metadata"].get("namespace"),
     )
     assert result is False
-    assert instance is None
+    assert instance == {}
     assert abs(elapsed - 3) <= 1