[WIP] Integrated AWX tests

.gitignore

@@ -78,3 +78,8 @@ setup_v1.py
 tower_cli_v2/
 bin/tower-cli-v2
 setup_v2.py
+
+# for AWX install
+/awx
+.cache/
+awx.sqlite3

.travis.yml

@@ -11,10 +11,12 @@ matrix:
   allow_failures:
     - python: nightly
 install:
-    - pip install -r requirements_dev.txt
-before_script:
-    flake8 .
+    - if [[ $TRAVIS_PYTHON_VERSION == '2.7' ]]; then ./install_awx.sh; fi
+    - if [[ $TRAVIS_PYTHON_VERSION == '2.7' ]]; then python install_awx_req.py; fi
+    - pip install -r requirements_dev.txt -r requirements.txt
 script:
-    - tox
+    - flake8 tower-cli/ tests/
+    - tox tests/
+    - if [[ $TRAVIS_PYTHON_VERSION == '2.7' ]]; then py.test tests_awx/; fi
 after_success:
     coveralls

install_awx.sh

@@ -0,0 +1,9 @@
+rm -rf awx/
+git clone https://github.com/ansible/awx.git awx --depth=1
+cd awx
+rm awx/sso/__init__.py
+touch awx/sso/__init__.py
+python setup.py install
+cd ..
+# have to add awx dir to path
+[[ ":$PYTHONPATH:" != *":$PWD/awx:"* ]] && PYTHONPATH="${PYTHONPATH}:$PWD/awx"

install_awx_req.py

@@ -0,0 +1,34 @@
+from subprocess import call
+
+files = ['requirements.in', 'requirements_dev.txt', 'requirements_git.txt']
+
+
+seen = set([])
+failed = set([])
+
+
+for file_name in files:
+    rel_path = 'awx/requirements/{}'.format(file_name)
+    with open(rel_path, 'r') as f:
+        data = f.read()
+    for line in data.split('\n'):
+        if not line or line.startswith('#') or not line.split('#'):
+            continue
+        target = line.split('#')[0].strip()
+        pkg = target.split('=')[0]
+        # same package listed in multiple files
+        if pkg in seen:
+            print('Skipping second listing of ' + str(pkg))
+            continue
+        # exclusions
+        if pkg in ['pip', 'setuptools']:
+            print('Passing over {}, in exclusions list'.format(target))
+            continue
+        seen.add(pkg)
+        r = call("pip install " + target, shell=True)
+        if r:
+            failed.add(target)
+
+if failed:
+    print('tower-cli AWX integration failed to install packages \n')
+    print(' - \n'.join(failed))

pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+DJANGO_SETTINGS_MODULE = awx.settings.development
+python_files = *.py
+addopts = --reuse-db --nomigrations --tb=native

requirements_dev.txt

@@ -1,3 +1,4 @@
 flake8
 tox-travis
-coveralls
+coveralls
+pytest

tests_awx/conftest.py

@@ -0,0 +1,24 @@
+# Python
+import pytest
+
+# Django
+from django.core.urlresolvers import resolve
+from django.utils.six.moves.urllib.parse import urlparse
+
+from rest_framework.test import (
+    APIRequestFactory,
+    force_authenticate,
+)
+
+# AWX
+from awx.main.models import User, Organization
+
+
+@pytest.fixture
+def admin():
+    return User.objects.create(username='admin_user', is_superuser=True)
+
+
+@pytest.fixture
+def organization():
+    return Organization.objects.create(name='an-org')

tests_awx/test_organization.py

@@ -0,0 +1,23 @@
+import pytest
+
+from awx.api.versioning import reverse
+
+from tower_cli.conf import settings
+from tower_cli import get_resource
+
+
+@pytest.mark.django_db
+def test_create_org(admin):
+    with settings.runtime_values(host='connection: local', username=admin.username):
+        org_res = get_resource('organization')
+        r = org_res.create(name='an-org-created')
+    assert r['name'] == 'an-org-created'
+
+
+@pytest.mark.django_db
+def test_read_org(organization, admin):
+    with settings.runtime_values(host='connection: local', username=admin.username):
+        org_res = get_resource('organization')
+        r = org_res.get(name='an-org')
+    assert r['name'] == organization.name
+

tower_cli/api.py

@@ -37,6 +37,71 @@
 TOWER_DATETIME_FMT = r'%Y-%m-%dT%H:%M:%S.%fZ'
 
 
+def local_request(method, url, **kwargs):
+    try:
+        # Django
+        from django.core.urlresolvers import resolve
+        from django.utils.six.moves.urllib.parse import urlparse
+
+        from rest_framework.test import (
+            APIRequestFactory,
+            force_authenticate,
+        )
+        # test import
+        from awx.main.models import User
+    except ImportError:
+        logger.debug('You are using local connection, you need AWX installed.')
+        raise
+
+    if 'data' in kwargs:
+        data = kwargs['data']
+        if not isinstance(data, dict):
+            data = json.loads(data)
+    elif 'params' in kwargs:
+        data = kwargs['params']
+    else:
+        data = {}
+
+    middleware = None
+    request_kwargs = {}
+    request_kwargs['data'] = data
+    # headers = content['headers']
+    # # passwords? who needs em
+    user = User.objects.get(username=settings.username)
+    # if 'Content-Type' in headers:
+    #     kwargs['content_type'] = headers['Content-Type']
+
+        # def rf(url, data_or_user=None, user=None, middleware=None, expect=None, **kwargs):
+    if 'format' not in kwargs and 'content_type' not in kwargs:
+        request_kwargs['format'] = 'json'
+
+    view, view_args, view_kwargs = resolve(urlparse(url)[2])
+    request = getattr(APIRequestFactory(), method.lower())(url, **request_kwargs)
+    if isinstance(kwargs.get('cookies', None), dict):
+        for key, value in kwargs['cookies'].items():
+            request.COOKIES[key] = value
+    if middleware:
+        middleware.process_request(request)
+    if user:
+        force_authenticate(request, user=user)
+
+    response = view(request, *view_args, **view_kwargs)
+    if middleware:
+        middleware.process_response(request, response)
+    if hasattr(response, 'render'):
+        response.render()
+
+    # hacks specific to tower-cli
+    def make_json(self):
+        return self.data
+
+    import types
+    response.json = types.MethodType(make_json, response)
+
+    return response
+
+
+
 class BasicTowerAuth(AuthBase):
 
     def __init__(self, username, password, cli_client):
@@ -132,10 +197,15 @@ def _make_request(self, method, url, args, kwargs):
         # Call the superclass method.
         try:
             with warnings.catch_warnings():
-                warnings.simplefilter(
-                    "ignore", urllib3.exceptions.InsecureRequestWarning)
-                return super(Client, self).request(
-                    method, url, *args, verify=verify_ssl, **kwargs)
+                if settings.host == 'connection: local':
+                    return local_request(
+                        method, url, *args, **kwargs
+                    )
+                else:
+                    warnings.simplefilter(
+                        "ignore", urllib3.exceptions.InsecureRequestWarning)
+                    return super(Client, self).request(
+                        method, url, *args, verify=verify_ssl, **kwargs)
         except SSLError as ex:
             # Throw error if verify_ssl not set to false and server
             #  is not using verified certificate.