<commit_before># Copyright (c) 2014 AnsibleWorks, Inc.
# All Rights Reserved.

# Python
import base64
import contextlib
import json
import os
import random
import shutil
import sys
import tempfile
import time
from multiprocessing import Process
from subprocess import Popen
import re
import mock

# PyYAML
import yaml

# Django
import django.test
from django.conf import settings, UserSettingsHolder
from django.contrib.auth.models import User
from django.test.client import Client
from django.test.utils import override_settings

# AWX
from awx.main.models import * # noqa
from awx.main.backend import LDAPSettings
from awx.main.management.commands.run_callback_receiver import CallbackReceiver
from awx.main.management.commands.run_task_system import run_taskmanager
from awx.main.utils import get_ansible_version
from awx.main.task_engine import TaskEngager as LicenseWriter

TEST_PLAYBOOK = '''- hosts: mygroup
  gather_facts: false
  tasks:
  - name: woohoo
    command: test 1 = 1
'''

class QueueTestMixin(object):
    def start_queue(self):
        self.start_redis()
        receiver = CallbackReceiver()
        self.queue_process = Process(target=receiver.run_subscriber,
                                     args=(False,))
        self.queue_process.start()

    def terminate_queue(self):
        if hasattr(self, 'queue_process'):
            self.queue_process.terminate()
        self.stop_redis()

    def start_redis(self):
        if not getattr(self, 'redis_process', None):
            self.redis_process = Popen('redis-server --port 16379 > /dev/null',
                                       shell=True, executable='/bin/bash')

    def stop_redis(self):
        if getattr(self, 'redis_process', None):
            self.redis_process.kill()
            self.redis_process = None
            

# The observed effect of not calling terminate_queue() if you call start_queue() are
# an hang on test cleanup database delete. Thus, to ensure terminate_queue() is called 
# whenever start_queue() is called just inherit  from this class when you want to use the queue.
class QueueStartStopTestMixin(QueueTestMixin):
    def setUp(self):
        super(QueueStartStopTestMixin, self).setUp()
        self.start_queue()

    def tearDown(self):
        super(QueueStartStopTestMixin, self).tearDown()
        self.terminate_queue()

class MockCommonlySlowTestMixin(object):
    def __init__(self, *args, **kwargs):
        from awx.api import generics
        mock.patch.object(generics, 'get_view_description', return_value=None).start()
        super(MockCommonlySlowTestMixin, self).__init__(*args, **kwargs)

ansible_version = get_ansible_version()
class BaseTestMixin(QueueTestMixin, MockCommonlySlowTestMixin):
    '''
    Mixin with shared code for use by all test cases.
    '''

    def setUp(self):
        super(BaseTestMixin, self).setUp()
        global ansible_version

        self.object_ctr = 0
        # Save sys.path before tests.
        self._sys_path = [x for x in sys.path]
        # Save os.environ before tests.
        self._environ = dict(os.environ.items())
        # Capture current directory to change back after each test.
        self._cwd = os.getcwd()
        # Capture list of temp files/directories created during tests.
        self._temp_paths = []
        self._current_auth = None
        self._user_passwords = {}
        self.ansible_version = ansible_version
        self.assertNotEqual(self.ansible_version, 'unknown')
        # Wrap settings so we can redefine them within each test.
        self._wrapped = settings._wrapped
        settings._wrapped = UserSettingsHolder(settings._wrapped)
        # Set all AUTH_LDAP_* settings to defaults to avoid using LDAP for
        # tests unless expicitly configured.
        for name, value in LDAPSettings.defaults.items():
            if name == 'SERVER_URI':
                value = ''
            setattr(settings, 'AUTH_LDAP_%s' % name, value)
        # Pass test database settings in environment for use by any management
        # commands that run from tests.
        for opt in ('ENGINE', 'NAME', 'USER', 'PASSWORD', 'HOST', 'PORT'):
            os.environ['AWX_TEST_DATABASE_%s' % opt] = settings.DATABASES['default'][opt]
        # Set flag so that task chain works with unit tests.
        settings.CELERY_UNIT_TEST = True
        settings.SYSTEM_UUID='00000000-0000-0000-0000-000000000000'
        settings.BROKER_URL='redis://localhost:16379/'
        
        # Create unique random consumer and queue ports for zeromq callback.
        if settings.CALLBACK_CONSUMER_PORT:
            callback_port = random.randint(55700, 55799)
            settings.CALLBACK_CONSUMER_PORT = 'tcp://127.0.0.1:%d' % callback_port
            callback_queue_path = '/tmp/callback_receiver_test_%d.ipc' % callback_port
            self._temp_paths.append(callback_queue_path)
            settings.CALLBACK_QUEUE_PORT = 'ipc://%s' % callback_queue_path
            settings.TASK_COMMAND_PORT = 'ipc:///tmp/task_command_receiver_%d.ipc' % callback_port
        # Disable socket notifications for unit tests.
        settings.SOCKETIO_NOTIFICATION_PORT = None
        # Make temp job status directory for unit tests.
        job_status_dir = tempfile.mkdtemp()
        self._temp_paths.append(job_status_dir)
        settings.JOBOUTPUT_ROOT = os.path.abspath(job_status_dir)
        settings.CACHES = {
            'default': {
                'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
                'LOCATION': 'unittests'
            }
        }
        self._start_time = time.time()

    def tearDown(self):
        super(BaseTestMixin, self).tearDown()
        # Restore sys.path after tests.
        sys.path = self._sys_path
        # Restore os.environ after tests.
        for k,v in self._environ.items():
            if os.environ.get(k, None) != v:
                os.environ[k] = v
        for k,v in os.environ.items():
            if k not in self._environ.keys():
                del os.environ[k]
        # Restore current directory after each test.
        os.chdir(self._cwd)
        # Cleanup temp files/directories created during tests.
        for project_dir in self._temp_paths:
            if os.path.exists(project_dir):
                if os.path.isdir(project_dir):
                    shutil.rmtree(project_dir, True)
                else:
                    os.remove(project_dir)
        # Restore previous settings after each test.
        settings._wrapped = self._wrapped

    def unique_name(self, string):
        rnd_str = '____' + str(random.randint(1, 9999999))
        return __name__ + '-generated-' + string + rnd_str

    def create_test_license_file(self, instance_count=10000, license_date=int(time.time() + 3600)):
        writer = LicenseWriter( 
            company_name='AWX',
            contact_name='<NAME>',
            contact_email='<EMAIL>',
            license_date=license_date,
            instance_count=instance_count)
        handle, license_path = tempfile.mkstemp(suffix='.json')
        os.close(handle)
        writer.write_file(license_path)
        self._temp_paths.append(license_path)
        os.environ['AWX_LICENSE_FILE'] = license_path

    def create_expired_license_file(self, instance_count=1000, grace_period=False):
        license_date = time.time() - 1
        if not grace_period:
            license_date -= 2592000
        self.create_test_license_file(instance_count, license_date)
        os.environ['SKIP_LICENSE_FIXUP_FOR_TEST'] = '1'

    def assertElapsedLessThan(self, seconds):
        elapsed = time.time() - self._start_time
        self.assertTrue(elapsed < seconds, 'elapsed time of %0.3fs is greater than %0.3fs' % (elapsed, seconds))

    @contextlib.contextmanager
    def current_user(self, user_or_username, password=<PASSWORD>):
        try:
            if isinstance(user_or_username, User):
                username = user_or_username.username
            else:<commit_msg>create enterprise license in tests<commit_after># Copyright (c) 2014 AnsibleWorks, Inc.
# All Rights Reserved.

# Python
import base64
import contextlib
import json
import os
import random
import shutil
import sys
import tempfile
import time
from multiprocessing import Process
from subprocess import Popen
import re
import mock

# PyYAML
import yaml

# Django
import django.test
from django.conf import settings, UserSettingsHolder
from django.contrib.auth.models import User
from django.test.client import Client
from django.test.utils import override_settings

# AWX
from awx.main.models import * # noqa
from awx.main.backend import LDAPSettings
from awx.main.management.commands.run_callback_receiver import CallbackReceiver
from awx.main.management.commands.run_task_system import run_taskmanager
from awx.main.utils import get_ansible_version
from awx.main.task_engine import TaskEngager as LicenseWriter

TEST_PLAYBOOK = '''- hosts: mygroup
  gather_facts: false
  tasks:
  - name: woohoo
    command: test 1 = 1
'''

class QueueTestMixin(object):
    def start_queue(self):
        self.start_redis()
        receiver = CallbackReceiver()
        self.queue_process = Process(target=receiver.run_subscriber,
                                     args=(False,))
        self.queue_process.start()

    def terminate_queue(self):
        if hasattr(self, 'queue_process'):
            self.queue_process.terminate()
        self.stop_redis()

    def start_redis(self):
        if not getattr(self, 'redis_process', None):
            self.redis_process = Popen('redis-server --port 16379 > /dev/null',
                                       shell=True, executable='/bin/bash')

    def stop_redis(self):
        if getattr(self, 'redis_process', None):
            self.redis_process.kill()
            self.redis_process = None
            

# The observed effect of not calling terminate_queue() if you call start_queue() are
# an hang on test cleanup database delete. Thus, to ensure terminate_queue() is called 
# whenever start_queue() is called just inherit  from this class when you want to use the queue.
class QueueStartStopTestMixin(QueueTestMixin):
    def setUp(self):
        super(QueueStartStopTestMixin, self).setUp()
        self.start_queue()

    def tearDown(self):
        super(QueueStartStopTestMixin, self).tearDown()
        self.terminate_queue()

class MockCommonlySlowTestMixin(object):
    def __init__(self, *args, **kwargs):
        from awx.api import generics
        mock.patch.object(generics, 'get_view_description', return_value=None).start()
        super(MockCommonlySlowTestMixin, self).__init__(*args, **kwargs)

ansible_version = get_ansible_version()
class BaseTestMixin(QueueTestMixin, MockCommonlySlowTestMixin):
    '''
    Mixin with shared code for use by all test cases.
    '''

    def setUp(self):
        super(BaseTestMixin, self).setUp()
        global ansible_version

        self.object_ctr = 0
        # Save sys.path before tests.
        self._sys_path = [x for x in sys.path]
        # Save os.environ before tests.
        self._environ = dict(os.environ.items())
        # Capture current directory to change back after each test.
        self._cwd = os.getcwd()
        # Capture list of temp files/directories created during tests.
        self._temp_paths = []
        self._current_auth = None
        self._user_passwords = {}
        self.ansible_version = ansible_version
        self.assertNotEqual(self.ansible_version, 'unknown')
        # Wrap settings so we can redefine them within each test.
        self._wrapped = settings._wrapped
        settings._wrapped = UserSettingsHolder(settings._wrapped)
        # Set all AUTH_LDAP_* settings to defaults to avoid using LDAP for
        # tests unless expicitly configured.
        for name, value in LDAPSettings.defaults.items():
            if name == 'SERVER_URI':
                value = ''
            setattr(settings, 'AUTH_LDAP_%s' % name, value)
        # Pass test database settings in environment for use by any management
        # commands that run from tests.
        for opt in ('ENGINE', 'NAME', 'USER', 'PASSWORD', 'HOST', 'PORT'):
            os.environ['AWX_TEST_DATABASE_%s' % opt] = settings.DATABASES['default'][opt]
        # Set flag so that task chain works with unit tests.
        settings.CELERY_UNIT_TEST = True
        settings.SYSTEM_UUID='00000000-0000-0000-0000-000000000000'
        settings.BROKER_URL='redis://localhost:16379/'
        
        # Create unique random consumer and queue ports for zeromq callback.
        if settings.CALLBACK_CONSUMER_PORT:
            callback_port = random.randint(55700, 55799)
            settings.CALLBACK_CONSUMER_PORT = 'tcp://127.0.0.1:%d' % callback_port
            callback_queue_path = '/tmp/callback_receiver_test_%d.ipc' % callback_port
            self._temp_paths.append(callback_queue_path)
            settings.CALLBACK_QUEUE_PORT = 'ipc://%s' % callback_queue_path
            settings.TASK_COMMAND_PORT = 'ipc:///tmp/task_command_receiver_%d.ipc' % callback_port
        # Disable socket notifications for unit tests.
        settings.SOCKETIO_NOTIFICATION_PORT = None
        # Make temp job status directory for unit tests.
        job_status_dir = tempfile.mkdtemp()
        self._temp_paths.append(job_status_dir)
        settings.JOBOUTPUT_ROOT = os.path.abspath(job_status_dir)
        settings.CACHES = {
            'default': {
                'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
                'LOCATION': 'unittests'
            }
        }
        self._start_time = time.time()

    def tearDown(self):
        super(BaseTestMixin, self).tearDown()
        # Restore sys.path after tests.
        sys.path = self._sys_path
        # Restore os.environ after tests.
        for k,v in self._environ.items():
            if os.environ.get(k, None) != v:
                os.environ[k] = v
        for k,v in os.environ.items():
            if k not in self._environ.keys():
                del os.environ[k]
        # Restore current directory after each test.
        os.chdir(self._cwd)
        # Cleanup temp files/directories created during tests.
        for project_dir in self._temp_paths:
            if os.path.exists(project_dir):
                if os.path.isdir(project_dir):
                    shutil.rmtree(project_dir, True)
                else:
                    os.remove(project_dir)
        # Restore previous settings after each test.
        settings._wrapped = self._wrapped

    def unique_name(self, string):
        rnd_str = '____' + str(random.randint(1, 9999999))
        return __name__ + '-generated-' + string + rnd_str

    def create_test_license_file(self, instance_count=10000, license_date=int(time.time() + 3600)):
        writer = LicenseWriter( 
            company_name='AWX',
            contact_name='<NAME>',
            contact_email='<EMAIL>',
            license_date=license_date,
            instance_count=instance_count,
            license_type='enterprise')
        handle, license_path = tempfile.mkstemp(suffix='.json')
        os.close(handle)
        writer.write_file(license_path)
        self._temp_paths.append(license_path)
        os.environ['AWX_LICENSE_FILE'] = license_path

    def create_expired_license_file(self, instance_count=1000, grace_period=False):
        license_date = time.time() - 1
        if not grace_period:
            license_date -= 2592000
        self.create_test_license_file(instance_count, license_date)
        os.environ['SKIP_LICENSE_FIXUP_FOR_TEST'] = '1'

    def assertElapsedLessThan(self, seconds):
        elapsed = time.time() - self._start_time
        self.assertTrue(elapsed < seconds, 'elapsed time of %0.3fs is greater than %0.3fs' % (elapsed, seconds))

    @contextlib.contextmanager
    def current_user(self, user_or_username, password=<PASSWORD>):
        try:
            if isinstance(user_or_username, User):
                username = user_or_username.username
            else:
{'fraction_non_alphanumeric': 0.09110169491525423, 'fraction_numerical': 0.012587238285144567, 'mean_word_length': 2.639229024943311, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 6, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7423305', 'n_tokens_mistral': 4854, 'n_tokens_neox': 4663, 'n_words': 1132}