tests: Add/update tests for warnings returned by k8s and K8sService

Add/update tests that verify k8s and K8sService return the expected
warnings.

Signed-off-by: Felix Matouschek <[email protected]>

tests/unit/module_utils/test_runner.py

@@ -31,7 +31,7 @@
 
 
 @pytest.mark.parametrize(
-    "action, params, existing, instance, expected",
+    "action, params, existing, instance_warnings, expected",
     [
         (
             "delete",
@@ -51,14 +51,26 @@
             "apply",
             {"apply": "yes"},
             {},
-            definition,
+            (definition, []),
             {"changed": True, "method": "apply", "result": definition},
         ),
+        (
+            "apply",
+            {"apply": "yes"},
+            {},
+            (definition, ["test warning"]),
+            {
+                "changed": True,
+                "method": "apply",
+                "result": definition,
+                "warnings": ["test warning"],
+            },
+        ),
         (
             "create",
             {"state": "patched"},
             {},
-            {},
+            ({}, []),
             {
                 "changed": False,
                 "result": {},
@@ -71,42 +83,78 @@
             "create",
             {},
             {},
-            definition,
+            (definition, []),
             {"changed": True, "method": "create", "result": definition},
         ),
+        (
+            "create",
+            {},
+            {},
+            (definition, ["test warning"]),
+            {
+                "changed": True,
+                "method": "create",
+                "result": definition,
+                "warnings": ["test warning"],
+            },
+        ),
         (
             "replace",
             {"force": "yes"},
             definition,
-            definition,
+            (definition, []),
             {"changed": False, "method": "replace", "result": definition},
         ),
         (
             "replace",
             {"force": "yes"},
             definition,
-            modified_def,
+            (modified_def, []),
             {"changed": True, "method": "replace", "result": modified_def},
         ),
+        (
+            "replace",
+            {"force": "yes"},
+            definition,
+            (modified_def, ["test warning"]),
+            {
+                "changed": True,
+                "method": "replace",
+                "result": modified_def,
+                "warnings": ["test warning"],
+            },
+        ),
         (
             "update",
             {},
             definition,
-            definition,
+            (definition, []),
             {"changed": False, "method": "update", "result": definition},
         ),
         (
             "update",
             {},
             definition,
-            modified_def,
+            (modified_def, []),
             {"changed": True, "method": "update", "result": modified_def},
         ),
+        (
+            "update",
+            {},
+            definition,
+            (modified_def, ["test warning"]),
+            {
+                "changed": True,
+                "method": "update",
+                "result": modified_def,
+                "warnings": ["test warning"],
+            },
+        ),
         (
             "create",
             {"label_selectors": ["app=foo"]},
             {},
-            definition,
+            (definition, []),
             {
                 "changed": False,
                 "msg": "resource 'kind=Pod,name=foo,namespace=foo' filtered by label_selectors.",
@@ -116,18 +164,18 @@
             "create",
             {"label_selectors": ["app=nginx"]},
             {},
-            definition,
+            (definition, []),
             {"changed": True, "method": "create", "result": definition},
         ),
     ],
 )
-def test_perform_action(action, params, existing, instance, expected):
+def test_perform_action(action, params, existing, instance_warnings, 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}
+    spec = {action + ".return_value": instance_warnings}
     svc.configure_mock(**spec)
 
     result = perform_action(svc, definition, params)

tests/unit/module_utils/test_service.py

@@ -1,9 +1,11 @@
+from json import dumps
 from unittest.mock import Mock
 
 import pytest
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import (
     K8sService,
     diff_objects,
+    parse_quoted_string,
 )
 from kubernetes.dynamic.exceptions import NotFoundError
 from kubernetes.dynamic.resource import Resource, ResourceInstance
@@ -57,6 +59,22 @@ def mock_pod_updated_resource_instance():
     return ResourceInstance(None, pod_definition_updated)
 
 
+@pytest.fixture(scope="module")
+def mock_pod_response():
+    resp = Mock()
+    resp.data.decode.return_value = dumps(pod_definition)
+    resp.headers = {}
+    return resp
+
+
+@pytest.fixture(scope="module")
+def mock_pod_warnings_response():
+    resp = Mock()
+    resp.data.decode.return_value = dumps(pod_definition)
+    resp.headers = {"warning": '299 - "test warning 1", 299 - "test warning 2"'}
+    return resp
+
+
 def test_diff_objects_no_diff():
     match, diff = diff_objects(pod_definition, pod_definition)
 
@@ -159,26 +177,44 @@ def test_service_delete_existing_resource_check_mode(mock_pod_resource_instance)
     client.delete.assert_not_called()
 
 
-def test_service_create_resource(mock_pod_resource_instance):
-    spec = {"create.side_effect": [mock_pod_resource_instance]}
+def test_service_create_resource(mock_pod_response, mock_pod_resource_instance):
+    spec = {"create.side_effect": [mock_pod_response]}
+    client = Mock(**spec)
+    module = Mock()
+    module.params = {}
+    module.check_mode = False
+    svc = K8sService(client, module)
+    result, warnings = svc.create(Mock(), pod_definition)
+
+    assert result == mock_pod_resource_instance.to_dict()
+    assert not warnings
+
+
+def test_service_create_resource_warnings(
+    mock_pod_warnings_response, mock_pod_resource_instance
+):
+    spec = {"create.side_effect": [mock_pod_warnings_response]}
     client = Mock(**spec)
     module = Mock()
     module.params = {}
     module.check_mode = False
     svc = K8sService(client, module)
-    result = svc.create(Mock(), pod_definition)
+    result, warnings = svc.create(Mock(), pod_definition)
 
     assert result == mock_pod_resource_instance.to_dict()
+    assert warnings[0] == "test warning 1"
+    assert warnings[1] == "test warning 2"
 
 
 def test_service_create_resource_check_mode():
     client = Mock(dry_run=False)
     client.create.return_value = mock_pod_resource_instance
     module = Mock(params={}, check_mode=True)
     svc = K8sService(client, module)
-    result = svc.create(Mock(), pod_definition)
+    result, warnings = svc.create(Mock(), pod_definition)
 
     assert result == pod_definition
+    assert not warnings
     client.create.assert_not_called()
 
 
@@ -224,40 +260,99 @@ def test_create_project_request():
     assert results["result"] == project_definition
 
 
-def test_service_apply_existing_resource(mock_pod_resource_instance):
-    spec = {"apply.side_effect": [mock_pod_resource_instance]}
+def test_service_apply_existing_resource(mock_pod_response, mock_pod_resource_instance):
+    spec = {"apply.side_effect": [mock_pod_response]}
     client = Mock(**spec)
     module = Mock()
     module.params = {"apply": True}
     module.check_mode = False
     svc = K8sService(client, module)
-    result = svc.apply(Mock(), pod_definition_updated, mock_pod_resource_instance)
+    result, warnings = svc.apply(
+        Mock(), pod_definition_updated, mock_pod_resource_instance
+    )
 
     assert result == mock_pod_resource_instance.to_dict()
+    assert not warnings
 
 
-def test_service_replace_existing_resource(mock_pod_resource_instance):
-    spec = {"replace.side_effect": [mock_pod_resource_instance]}
+def test_service_apply_existing_resource_warnings(
+    mock_pod_warnings_response, mock_pod_resource_instance
+):
+    spec = {"apply.side_effect": [mock_pod_warnings_response]}
+    client = Mock(**spec)
+    module = Mock()
+    module.params = {"apply": True}
+    module.check_mode = False
+    svc = K8sService(client, module)
+    result, warnings = svc.apply(
+        Mock(), pod_definition_updated, mock_pod_resource_instance
+    )
+
+    assert result == mock_pod_resource_instance.to_dict()
+    assert warnings[0] == "test warning 1"
+    assert warnings[1] == "test warning 2"
+
+
+def test_service_replace_existing_resource(
+    mock_pod_response, mock_pod_resource_instance
+):
+    spec = {"replace.side_effect": [mock_pod_response]}
+    client = Mock(**spec)
+    module = Mock()
+    module.params = {}
+    module.check_mode = False
+    svc = K8sService(client, module)
+    result, warnings = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
+
+    assert result == mock_pod_resource_instance.to_dict()
+    assert not warnings
+
+
+def test_service_replace_existing_resource_warnings(
+    mock_pod_warnings_response, mock_pod_resource_instance
+):
+    spec = {"replace.side_effect": [mock_pod_warnings_response]}
     client = Mock(**spec)
     module = Mock()
     module.params = {}
     module.check_mode = False
     svc = K8sService(client, module)
-    result = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
+    result, warnings = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
 
     assert result == mock_pod_resource_instance.to_dict()
+    assert warnings[0] == "test warning 1"
+    assert warnings[1] == "test warning 2"
 
 
-def test_service_update_existing_resource(mock_pod_resource_instance):
-    spec = {"replace.side_effect": [mock_pod_resource_instance]}
+def test_service_update_existing_resource(
+    mock_pod_response, mock_pod_resource_instance
+):
+    spec = {"replace.side_effect": [mock_pod_response]}
     client = Mock(**spec)
     module = Mock()
     module.params = {}
     module.check_mode = False
     svc = K8sService(client, module)
-    result = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
+    result, warnings = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
 
     assert result == mock_pod_resource_instance.to_dict()
+    assert not warnings
+
+
+def test_service_update_existing_resource_warnings(
+    mock_pod_warnings_response, mock_pod_resource_instance
+):
+    spec = {"replace.side_effect": [mock_pod_warnings_response]}
+    client = Mock(**spec)
+    module = Mock()
+    module.params = {}
+    module.check_mode = False
+    svc = K8sService(client, module)
+    result, warnings = svc.replace(Mock(), pod_definition, mock_pod_resource_instance)
+
+    assert result == mock_pod_resource_instance.to_dict()
+    assert warnings[0] == "test warning 1"
+    assert warnings[1] == "test warning 2"
 
 
 def test_service_find(mock_pod_resource_instance):
@@ -288,3 +383,24 @@ def test_service_find_error():
     assert isinstance(results, dict)
     assert results["api_found"] is True
     assert results["resources"] == []
+
+
+@pytest.mark.parametrize(
+    "quoted_string,expected_val,expected_remainder",
+    [
+        (
+            '"Response is stale" Tue, 15 Nov 1994 12:45:26 GMT',
+            "Response is stale",
+            "Tue, 15 Nov 1994 12:45:26 GMT",
+        ),
+        (
+            '"unknown field \\"spec.template.spec.disk\\""',
+            'unknown field "spec.template.spec.disk"',
+            "",
+        ),
+    ],
+)
+def test_parse_quoted_string(quoted_string, expected_val, expected_remainder):
+    val, remainder = parse_quoted_string(quoted_string)
+    assert val == expected_val
+    assert remainder == expected_remainder