add awx collection export tests

* Basic export tests
* Added test that highlights a problem with running Schedule exports as
  non-root user. We rely on the POST key in the OPTIONS response to
  determine the fields to export for a resource. The POST key is not
  present if a user does NOT have create privileges.
* Fixed up forwarding all headers from the API server back to the test
  code. This was causing a problem in awxkit code that checks for
  allowed HTTP Verbs in the headers.

awx_collection/test/awx/conftest.py

@@ -18,7 +18,20 @@ import pytest
 from ansible.module_utils.six import raise_from
 
 from awx.main.tests.functional.conftest import _request
-from awx.main.models import Organization, Project, Inventory, JobTemplate, Credential, CredentialType, ExecutionEnvironment, UnifiedJob
+from awx.main.tests.functional.conftest import credentialtype_scm, credentialtype_ssh  # noqa: F401; pylint: disable=unused-variable
+from awx.main.models import (
+    Organization,
+    Project,
+    Inventory,
+    JobTemplate,
+    Credential,
+    CredentialType,
+    ExecutionEnvironment,
+    UnifiedJob,
+    WorkflowJobTemplate,
+    NotificationTemplate,
+    Schedule,
+)
 
 from django.db import transaction
 
@@ -123,7 +136,7 @@ def run_module(request, collection_import):
             sanitize_dict(py_data)
             resp._content = bytes(json.dumps(django_response.data), encoding='utf8')
             resp.status_code = django_response.status_code
-            resp.headers = {'X-API-Product-Name': 'AWX', 'X-API-Product-Version': '0.0.1-devel'}
+            resp.headers = dict(django_response.headers)
 
             if request.config.getoption('verbose') > 0:
                 logger.info('%s %s by %s, code:%s', method, '/api/' + url.split('/api/')[1], request_user.username, resp.status_code)
@@ -236,10 +249,8 @@ def job_template(project, inventory):
 
 
 @pytest.fixture
-def machine_credential(organization):
-    ssh_type = CredentialType.defaults['ssh']()
-    ssh_type.save()
-    return Credential.objects.create(credential_type=ssh_type, name='machine-cred', inputs={'username': 'test_user', 'password': 'pas4word'})
+def machine_credential(credentialtype_ssh, organization):  # noqa: F811
+    return Credential.objects.create(credential_type=credentialtype_ssh, name='machine-cred', inputs={'username': 'test_user', 'password': 'pas4word'})
 
 
 @pytest.fixture
@@ -253,9 +264,7 @@ def vault_credential(organization):
 def kube_credential():
     ct = CredentialType.defaults['kubernetes_bearer_token']()
     ct.save()
-    return Credential.objects.create(
-        credential_type=ct, name='kube-cred', inputs={'host': 'my.cluster', 'bearer_token': 'my-token', 'verify_ssl': False}
-    )
+    return Credential.objects.create(credential_type=ct, name='kube-cred', inputs={'host': 'my.cluster', 'bearer_token': 'my-token', 'verify_ssl': False})
 
 
 @pytest.fixture
@@ -288,3 +297,42 @@ def mock_has_unpartitioned_events():
     # We mock this out to circumvent the migration query.
     with mock.patch.object(UnifiedJob, 'has_unpartitioned_events', new=False) as _fixture:
         yield _fixture
+
+
+@pytest.fixture
+def workflow_job_template(organization, inventory):
+    return WorkflowJobTemplate.objects.create(name='test-workflow_job_template', organization=organization, inventory=inventory)
+
+
+@pytest.fixture
+def notification_template(organization):
+    return NotificationTemplate.objects.create(
+        name='test-notification_template',
+        organization=organization,
+        notification_type="webhook",
+        notification_configuration=dict(
+            url="http://localhost",
+            username="",
+            password="",
+            headers={
+                "Test": "Header",
+            },
+        ),
+    )
+
+
+@pytest.fixture
+def scm_credential(credentialtype_scm, organization):  # noqa: F811
+    return Credential.objects.create(
+        credential_type=credentialtype_scm, name='scm-cred', inputs={'username': 'optimus', 'password': 'prime'}, organization=organization
+    )
+
+
+@pytest.fixture
+def rrule():
+    return 'DTSTART:20151117T050000Z RRULE:FREQ=DAILY;INTERVAL=1;COUNT=1'
+
+
+@pytest.fixture
+def schedule(job_template, rrule):
+    return Schedule.objects.create(unified_job_template=job_template, name='test-sched', rrule=rrule)

awx_collection/test/awx/test_export.py

@@ -0,0 +1,154 @@
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+import pytest
+
+from awx.main.models.execution_environments import ExecutionEnvironment
+from awx.main.models.jobs import JobTemplate
+
+from awx.main.tests.functional.conftest import user, system_auditor  # noqa: F401; pylint: disable=unused-import
+
+
+ASSETS = set([
+    "users",
+    "organizations",
+    "teams",
+    "credential_types",
+    "credentials",
+    "notification_templates",
+    "projects",
+    "inventory",
+    "inventory_sources",
+    "job_templates",
+    "workflow_job_templates",
+    "execution_environments",
+    "applications",
+    "schedules",
+])
+
+
+@pytest.fixture
+def job_template(project, inventory, organization, machine_credential):
+    jt = JobTemplate.objects.create(name='test-jt', project=project, inventory=inventory, organization=organization, playbook='helloworld.yml')
+    jt.credentials.add(machine_credential)
+    jt.save()
+    return jt
+
+
+@pytest.fixture
+def execution_environment(organization):
+    return ExecutionEnvironment.objects.create(name="test-ee", description="test-ee", managed=False, organization=organization)
+
+
+def find_by(result, name, key, value):
+    for c in result[name]:
+        if c[key] == value:
+            return c
+    values = [c.get(key, None) for c in result[name]]
+    raise ValueError(f"Failed to find assets['{name}'][{key}] = '{value}' valid values are {values}")
+
+
+@pytest.mark.django_db
+def test_export(run_module, admin_user):
+    """
+    There should be nothing to export EXCEPT the admin user.
+    """
+    result = run_module('export', dict(all=True), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+    assets = result['assets']
+
+    assert set(result['assets'].keys()) == ASSETS
+
+    u = find_by(assets, 'users', 'username', 'admin')
+    assert u['is_superuser'] is True
+
+    all_assets_except_users = {k: v for k, v in assets.items() if k != 'users'}
+
+    for k, v in all_assets_except_users.items():
+        assert v == [], f"Expected resource {k} to be empty. Instead it is {v}"
+
+
+@pytest.mark.django_db
+def test_export_simple(
+    run_module,
+    organization,
+    project,
+    inventory,
+    job_template,
+    scm_credential,
+    machine_credential,
+    workflow_job_template,
+    execution_environment,
+    notification_template,
+    rrule,
+    schedule,
+    admin_user,
+):
+    """
+    TODO: Ensure there aren't _more_ results in each resource than we expect
+    """
+    result = run_module('export', dict(all=True), admin_user)
+    assert not result.get('failed', False), result.get('msg', result)
+    assets = result['assets']
+
+    u = find_by(assets, 'users', 'username', 'admin')
+    assert u['is_superuser'] is True
+
+    find_by(assets, 'organizations', 'name', 'Default')
+
+    r = find_by(assets, 'credentials', 'name', 'scm-cred')
+    assert r['credential_type']['kind'] == 'scm'
+    assert r['credential_type']['name'] == 'Source Control'
+
+    r = find_by(assets, 'credentials', 'name', 'machine-cred')
+    assert r['credential_type']['kind'] == 'ssh'
+    assert r['credential_type']['name'] == 'Machine'
+
+    r = find_by(assets, 'job_templates', 'name', 'test-jt')
+    assert r['natural_key']['organization']['name'] == 'Default'
+    assert r['inventory']['name'] == 'test-inv'
+    assert r['project']['name'] == 'test-proj'
+
+    find_by(r['related'], 'credentials', 'name', 'machine-cred')
+
+    r = find_by(assets, 'inventory', 'name', 'test-inv')
+    assert r['organization']['name'] == 'Default'
+
+    r = find_by(assets, 'projects', 'name', 'test-proj')
+    assert r['organization']['name'] == 'Default'
+
+    r = find_by(assets, 'workflow_job_templates', 'name', 'test-workflow_job_template')
+    assert r['natural_key']['organization']['name'] == 'Default'
+    assert r['inventory']['name'] == 'test-inv'
+
+    r = find_by(assets, 'execution_environments', 'name', 'test-ee')
+    assert r['organization']['name'] == 'Default'
+
+    r = find_by(assets, 'schedules', 'name', 'test-sched')
+    assert r['rrule'] == rrule
+
+    r = find_by(assets, 'notification_templates', 'name', 'test-notification_template')
+    assert r['organization']['name'] == 'Default'
+    assert r['notification_configuration']['url'] == 'http://localhost'
+
+
+@pytest.mark.django_db
+def test_export_system_auditor(run_module, schedule, system_auditor):  # noqa: F811
+    """
+    This test illustrates that deficiency of export when ran as non-root user (i.e. system auditor).
+    The OPTIONS endpoint does NOT return POST for a system auditor. This is bad for the export code
+    because it relies on crawling the OPTIONS POST response to determine the fields to export.
+    """
+    result = run_module('export', dict(all=True), system_auditor)
+    assert result.get('failed', False), result.get('msg', result)
+
+    assert 'Failed to export assets substring not found' in result['msg'], (
+        'If you found this error then you have probably fixed a feature! The export code attempts to assertain the POST fields from the `description` field,'
+        ' but both the API side and the client inference code are lacking.'
+    )
+
+    # r = result['assets']['schedules'][0]
+    # assert r['natural_key']['name'] == 'test-sched'
+
+    # assert 'rrule' not in r, 'If you found this error then you have probably fixed a feature! We WANT rrule to be found in the export schedule payload.'