Changed status tweaks for API validation and encryption

API validation topic:
 - do not set changed=True if the object did not actually change
 - deals with cases where API manipulates data before saving

Warn if encrypted data prevent accurate changed status

Handle false changed case of tower_user password
  password field not present in data

Test changed=True warning with JT/WFJT survey spec defaults
  case for list data in JSON

awx_collection/test/awx/conftest.py

@@ -241,12 +241,12 @@ def silence_deprecation():
     """The deprecation warnings are stored in a global variable
     they will create cross-test interference. Use this to turn them off.
     """
-    with mock.patch('ansible.module_utils.basic.AnsibleModule.deprecate'):
-        yield
+    with mock.patch('ansible.module_utils.basic.AnsibleModule.deprecate') as this_mock:
+        yield this_mock
 
 
 @pytest.fixture
 def silence_warning():
     """Warnings use global variable, same as deprecations."""
-    with mock.patch('ansible.module_utils.basic.AnsibleModule.warn'):
-        yield
+    with mock.patch('ansible.module_utils.basic.AnsibleModule.warn') as this_mock:
+        yield this_mock

awx_collection/test/awx/test_credential.py

@@ -81,30 +81,6 @@ def test_create_vault_credential(run_module, admin_user, organization, silence_d
     assert result['id'] == cred.pk
 
 
-@pytest.mark.django_db
-def test_create_custom_credential_type(run_module, admin_user, silence_deprecation):
-    # Example from docs
-    result = run_module('tower_credential_type', dict(
-        name='Nexus',
-        description='Credentials type for Nexus',
-        kind='cloud',
-        inputs={"fields": [{"id": "server", "type": "string", "default": "", "label": ""}], "required": []},
-        injectors={'extra_vars': {'nexus_credential': 'test'}},
-        state='present',
-        validate_certs='false'
-    ), admin_user)
-    assert not result.get('failed', False), result.get('msg', result)
-    assert result.get('changed'), result
-
-    ct = CredentialType.objects.get(name='Nexus')
-
-    assert result['name'] == 'Nexus'
-    assert result['id'] == ct.pk
-
-    assert ct.inputs == {"fields": [{"id": "server", "type": "string", "default": "", "label": ""}], "required": []}
-    assert ct.injectors == {'extra_vars': {'nexus_credential': 'test'}}
-
-
 @pytest.mark.django_db
 def test_ct_precedence_over_kind(run_module, admin_user, organization, cred_type, silence_deprecation):
     result = run_module('tower_credential', dict(
@@ -155,16 +131,15 @@ def test_missing_credential_type(run_module, admin_user, organization):
 
 
 @pytest.mark.django_db
-def test_make_use_of_custom_credential_type(run_module, admin_user, cred_type):
-    Organization.objects.create(name='test-org')
-
+def test_make_use_of_custom_credential_type(run_module, organization, admin_user, cred_type):
     result = run_module('tower_credential', dict(
         name='Galaxy Token for Steve',
-        organization='test-org',
+        organization=organization.name,
         credential_type=cred_type.name,
-        inputs={'token': '7rEZK38DJl58A7RxA6EC7lLvUHbBQ1'},
-        state='present'
+        inputs={'token': '7rEZK38DJl58A7RxA6EC7lLvUHbBQ1'}
     ), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+    assert result.get('changed', False), result
 
     cred = Credential.objects.get(name='Galaxy Token for Steve')
     assert cred.credential_type_id == cred_type.id
@@ -174,3 +149,30 @@ def test_make_use_of_custom_credential_type(run_module, admin_user, cred_type):
 
     assert result['name'] == "Galaxy Token for Steve"
     assert result['id'] == cred.pk
+
+
+@pytest.mark.django_db
+def test_secret_field_write_twice(run_module, organization, admin_user, cred_type, silence_warning):
+    val1 = '7rEZK38DJl58A7RxA6EC7lLvUHbBQ1'
+    result = run_module('tower_credential', dict(
+        name='Galaxy Token for Steve',
+        organization=organization.name,
+        credential_type=cred_type.name,
+        inputs={'token': val1}
+    ), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+
+    Credential.objects.get(id=result['id']).inputs['token'] == val1
+
+    val2 = '7rEZ238DJl5837rxA6xxxlLvUHbBQ1'
+
+    result = run_module('tower_credential', dict(
+        name='Galaxy Token for Steve',
+        organization=organization.name,
+        credential_type=cred_type.name,
+        inputs={'token': val2}
+    ), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+
+    Credential.objects.get(id=result['id']).inputs['token'] == val2
+    assert result.get('changed'), result

awx_collection/test/awx/test_credential_type.py

@@ -0,0 +1,49 @@
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import pytest
+
+from awx.main.models import CredentialType
+
+
+@pytest.mark.django_db
+def test_create_custom_credential_type(run_module, admin_user, silence_deprecation):
+    # Example from docs
+    result = run_module('tower_credential_type', dict(
+        name='Nexus',
+        description='Credentials type for Nexus',
+        kind='cloud',
+        inputs={"fields": [{"id": "server", "type": "string", "default": "", "label": ""}], "required": []},
+        injectors={'extra_vars': {'nexus_credential': 'test'}},
+        state='present',
+    ), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+    assert result.get('changed'), result
+
+    ct = CredentialType.objects.get(name='Nexus')
+
+    assert result['name'] == 'Nexus'
+    assert result['id'] == ct.pk
+
+    assert ct.inputs == {"fields": [{"id": "server", "type": "string", "default": "", "label": ""}], "required": []}
+    assert ct.injectors == {'extra_vars': {'nexus_credential': 'test'}}
+
+
+@pytest.mark.django_db
+def test_changed_false_with_api_changes(run_module, admin_user):
+    result = run_module('tower_credential_type', dict(
+        name='foo',
+        kind='cloud',
+        inputs={"fields": [{"id": "env_value", "label": "foo", "default": "foo"}]},
+        injectors={'env': {'TEST_ENV_VAR': '{{ env_value }}'}},
+    ), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+    assert result.get('changed'), result
+
+    result = run_module('tower_credential_type', dict(
+        name='foo',
+        inputs={"fields": [{"id": "env_value", "label": "foo", "default": "foo"}]},
+        injectors={'env': {'TEST_ENV_VAR': '{{ env_value }}'}},
+    ), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+    assert not result.get('changed'), result

awx_collection/test/awx/test_job_template.py

@@ -112,7 +112,7 @@ def test_job_template_with_survey_spec(run_module, admin_user, project, inventor
     assert result.get('changed', False), result
     jt = JobTemplate.objects.get(pk=result['id'])
 
-    # assert jt.survey_spec == survey_spec
+    assert jt.survey_spec == survey_spec
 
     prior_ct = ActivityStream.objects.count()
     result = run_module('tower_job_template', dict(
@@ -130,3 +130,38 @@ def test_job_template_with_survey_spec(run_module, admin_user, project, inventor
 
     assert jt.survey_spec == survey_spec
     assert ActivityStream.objects.count() == prior_ct
+
+
+@pytest.mark.django_db
+def test_job_template_with_survey_encrypted_default(run_module, admin_user, project, inventory, silence_warning):
+    spec = {
+        "spec": [
+            {
+                "index": 0,
+                "question_name": "my question?",
+                "default": "very_secret_value",
+                "variable": "myvar",
+                "type": "password",
+                "required": False
+            }
+        ],
+        "description": "test",
+        "name": "test"
+    }
+    for i in range(2):
+        result = run_module('tower_job_template', dict(
+            name='foo',
+            playbook='helloworld.yml',
+            project=project.name,
+            inventory=inventory.name,
+            survey_spec=spec,
+            survey_enabled=True
+        ), admin_user)
+        assert not result.get('failed', False), result.get('msg', result)
+
+    assert result.get('changed', False), result  # not actually desired, but assert for sanity
+
+    silence_warning.assert_called_once_with(
+        "The field survey_spec of job_template {0} has encrypted data and "
+        "may inaccurately report task is changed.".format(result['id'])
+    )

awx_collection/test/awx/test_notification.py

@@ -85,7 +85,6 @@ def test_invalid_notification_configuration(run_module, admin_user, organization
 
 
 @pytest.mark.django_db
-@pytest.mark.xfail(reason='Handling API validation changes w.r.t. changed status is an open item')
 def test_deprecated_to_modern_no_op(run_module, admin_user, organization):
     nt_config = {
         'url': 'http://www.example.com/hook',

awx_collection/test/awx/test_user.py

@@ -0,0 +1,47 @@
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import pytest
+
+from unittest import mock
+
+from awx.main.models import User
+
+
+@pytest.fixture
+def mock_auth_stuff():
+    """Some really specific session-related stuff is done for changing or setting
+    passwords, so we will just avoid that here.
+    """
+    with mock.patch('awx.api.serializers.update_session_auth_hash'):
+        yield
+
+
+@pytest.mark.django_db
+def test_create_user(run_module, admin_user, mock_auth_stuff):
+    result = run_module('tower_user', dict(
+        username='Bob',
+        password='pass4word'
+    ), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+    assert result.get('changed'), result
+
+    user = User.objects.get(id=result['id'])
+    assert user.username == 'Bob'
+
+
+@pytest.mark.django_db
+def test_password_no_op_warning(run_module, admin_user, mock_auth_stuff, silence_warning):
+    for i in range(2):
+        result = run_module('tower_user', dict(
+            username='Bob',
+            password='pass4word'
+        ), admin_user)
+        assert not result.get('failed', False), result.get('msg', result)
+
+    assert result.get('changed')  # not actually desired, but assert for sanity
+
+    silence_warning.assert_called_once_with(
+        "The field password of user {0} has encrypted data and "
+        "may inaccurately report task is changed.".format(result['id'])
+    )