# Python
import pytest
from unittest import mock
import tempfile
import shutil
import urllib.parse
from unittest.mock import PropertyMock

# Django
from django.urls import resolve
from django.http import Http404
from django.core.handlers.exception import response_for_exception
from django.contrib.auth.models import User
from django.core.serializers.json import DjangoJSONEncoder
from django.db.backends.sqlite3.base import SQLiteCursorWrapper

# AWX
from awx.main.fields import JSONBField
from awx.main.models.projects import Project
from awx.main.models.ha import Instance

from rest_framework.test import (
    APIRequestFactory,
    force_authenticate,
)

from awx.main.models.credential import CredentialType, Credential
from awx.main.models.jobs import JobTemplate, SystemJobTemplate
from awx.main.models.inventory import (
    Group,
    Inventory,
    InventoryUpdate,
    InventorySource,
    CustomInventoryScript
)
from awx.main.models.organization import (
    Organization,
    Team,
)
from awx.main.models.rbac import Role
from awx.main.models.notifications import (
    NotificationTemplate,
    Notification
)
from awx.main.models.events import (
    JobEvent,
    AdHocCommandEvent,
    ProjectUpdateEvent,
    InventoryUpdateEvent,
    SystemJobEvent,
)
from awx.main.models.workflow import WorkflowJobTemplate
from awx.main.models.ad_hoc_commands import AdHocCommand
from awx.main.models.oauth import OAuth2Application as Application

__SWAGGER_REQUESTS__ = {}


@pytest.fixture(scope="session")
def swagger_autogen(requests=__SWAGGER_REQUESTS__):
    return requests


@pytest.fixture
def user():
    def u(name, is_superuser=False):
        try:
            user = User.objects.get(username=name)
        except User.DoesNotExist:
            user = User(username=name, is_superuser=is_superuser)
            user.set_password(name)
            user.save()
        return user
    return u


@pytest.fixture
def check_jobtemplate(project, inventory, credential):
    jt = JobTemplate.objects.create(
        job_type='check',
        project=project,
        inventory=inventory,
        name='check-job-template'
    )
    jt.credentials.add(credential)
    return jt


@pytest.fixture
def deploy_jobtemplate(project, inventory, credential):
    jt = JobTemplate.objects.create(
        job_type='run',
        project=project,
        inventory=inventory,
        name='deploy-job-template'
    )
    jt.credentials.add(credential)
    return jt


@pytest.fixture
def team(organization):
    return organization.teams.create(name='test-team')


@pytest.fixture
def team_member(user, team):
    ret = user('team-member', False)
    team.member_role.members.add(ret)
    return ret


@pytest.fixture(scope="session", autouse=True)
def project_playbooks():
    '''
    Return playbook_files as playbooks for manual projects when testing.
    '''
    class PlaybooksMock(mock.PropertyMock):
        def __get__(self, obj, obj_type):
            return obj.playbook_files
    mocked = mock.patch.object(Project, 'playbooks', new_callable=PlaybooksMock)
    mocked.start()


@pytest.fixture
def run_computed_fields_right_away(request):

    def run_me(inventory_id):
        i = Inventory.objects.get(id=inventory_id)
        i.update_computed_fields()

    mocked = mock.patch(
        'awx.main.signals.update_inventory_computed_fields.delay',
        new=run_me
    )
    mocked.start()

    request.addfinalizer(mocked.stop)


@pytest.fixture
@mock.patch.object(Project, "update", lambda self, **kwargs: None)
def project(instance, organization):
    prj = Project.objects.create(name="test-proj",
                                 description="test-proj-desc",
                                 organization=organization,
                                 playbook_files=['helloworld.yml', 'alt-helloworld.yml'],
                                 scm_revision='1234567890123456789012345678901234567890',
                                 scm_url='localhost',
                                 scm_type='git'
                                 )
    return prj


@pytest.fixture
@mock.patch.object(Project, "update", lambda self, **kwargs: None)
def manual_project(instance, organization):
    prj = Project.objects.create(name="test-manual-proj",
                                 description="manual-proj-desc",
                                 organization=organization,
                                 playbook_files=['helloworld.yml', 'alt-helloworld.yml'],
                                 local_path='_92__test_proj'
                                 )
    return prj


@pytest.fixture
def project_factory(organization):
    def factory(name):
        try:
            prj = Project.objects.get(name=name)
        except Project.DoesNotExist:
            prj = Project.objects.create(name=name,
                                         description="description for " + name,
                                         organization=organization
                                         )
        return prj
    return factory


@pytest.fixture
def job_factory(jt_linked, admin):
    def factory(job_template=jt_linked, initial_state='new', created_by=admin):
        return job_template.create_unified_job(_eager_fields={
            'status': initial_state, 'created_by': created_by})
    return factory


@pytest.fixture
def team_factory(organization):
    def factory(name):
        try:
            t = Team.objects.get(name=name)
        except Team.DoesNotExist:
            t = Team.objects.create(name=name,
                                    description="description for " + name,
                                    organization=organization)
        return t
    return factory


@pytest.fixture
def user_project(user):
    owner = user('owner')
    return Project.objects.create(name="test-user-project", created_by=owner, description="test-user-project-desc")


@pytest.fixture
def insights_project():
    return Project.objects.create(name="test-insights-project", scm_type="insights")


@pytest.fixture
def instance(settings):
    return Instance.objects.create(uuid=settings.SYSTEM_UUID, hostname="instance.example.org", capacity=100)


@pytest.fixture
def organization(instance):
    return Organization.objects.create(name="test-org", description="test-org-desc")


@pytest.fixture
def credentialtype_kube():
    kube = CredentialType.defaults['kubernetes_bearer_token']()
    kube.save()
    return kube


@pytest.fixture
def credentialtype_ssh():
    ssh = CredentialType.defaults['ssh']()
    ssh.save()
    return ssh


@pytest.fixture
def credentialtype_aws():
    aws = CredentialType.defaults['aws']()
    aws.save()
    return aws


@pytest.fixture
def credentialtype_net():
    net = CredentialType.defaults['net']()
    net.save()
    return net


@pytest.fixture
def credentialtype_vault():
    vault_type = CredentialType.defaults['vault']()
    vault_type.save()
    return vault_type


@pytest.fixture
def credentialtype_scm():
    scm_type = CredentialType.defaults['scm']()
    scm_type.save()
    return scm_type


@pytest.fixture
def credentialtype_insights():
    insights_type = CredentialType.defaults['insights']()
    insights_type.save()
    return insights_type


@pytest.fixture
def credentialtype_external():
    external_type_inputs = {
        'fields': [{
            'id': 'url',
            'label': 'Server URL',
            'type': 'string',
            'help_text': 'The server url.'
        }, {
            'id': 'token',
            'label': 'Token',
            'type': 'string',
            'secret': True,
            'help_text': 'An access token for the server.'
        }],
        'metadata': [{
            'id': 'key',
            'label': 'Key',
            'type': 'string'
        }, {
            'id': 'version',
            'label': 'Version',
            'type': 'string'
        }],
        'required': ['url', 'token', 'key'],
    }

    class MockPlugin(object):
        def backend(self, **kwargs):
            return 'secret'

    with mock.patch('awx.main.models.credential.CredentialType.plugin', new_callable=PropertyMock) as mock_plugin:
        mock_plugin.return_value = MockPlugin()
        external_type = CredentialType(
            kind='external',
            managed_by_tower=True,
            name='External Service',
            inputs=external_type_inputs
        )
        external_type.save()
        yield external_type


@pytest.fixture
def credential(credentialtype_aws):
    return Credential.objects.create(credential_type=credentialtype_aws, name='test-cred',
                                     inputs={'username': 'something', 'password': 'secret'})


@pytest.fixture
def net_credential(credentialtype_net):
    return Credential.objects.create(credential_type=credentialtype_net, name='test-cred',
                                     inputs={'username': 'something', 'password': 'secret'})


@pytest.fixture
def vault_credential(credentialtype_vault):
    return Credential.objects.create(credential_type=credentialtype_vault, name='test-cred',
                                     inputs={'vault_password': 'secret'})


@pytest.fixture
def machine_credential(credentialtype_ssh):
    return Credential.objects.create(credential_type=credentialtype_ssh, name='machine-cred',
                                     inputs={'username': 'test_user', 'password': 'pas4word'})


@pytest.fixture
def scm_credential(credentialtype_scm):
    return Credential.objects.create(credential_type=credentialtype_scm, name='scm-cred',
                                     inputs={'username': 'optimus', 'password': 'prime'})


@pytest.fixture
def insights_credential(credentialtype_insights):
    return Credential.objects.create(credential_type=credentialtype_insights, name='insights-cred',
                                     inputs={'username': 'morocco_mole', 'password': 'secret_squirrel'})


@pytest.fixture
def org_credential(organization, credentialtype_aws):
    return Credential.objects.create(credential_type=credentialtype_aws, name='test-cred',
                                     inputs={'username': 'something', 'password': 'secret'},
                                     organization=organization)


@pytest.fixture
def external_credential(credentialtype_external):
    return Credential.objects.create(credential_type=credentialtype_external, name='external-cred',
                                     inputs={'url': 'http://testhost.com', 'token': 'secret1'})


@pytest.fixture
def other_external_credential(credentialtype_external):
    return Credential.objects.create(credential_type=credentialtype_external, name='other-external-cred',
                                     inputs={'url': 'http://testhost.com', 'token': 'secret2'})


@pytest.fixture
def kube_credential(credentialtype_kube):
    return Credential.objects.create(credential_type=credentialtype_kube, name='kube-cred',
                                     inputs={'host': 'my.cluster', 'bearer_token': 'my-token', 'verify_ssl': False})


@pytest.fixture
def inventory(organization):
    return organization.inventories.create(name="test-inv")


@pytest.fixture
def insights_inventory(inventory):
    inventory.scm_type = 'insights'
    inventory.save()
    return inventory


@pytest.fixture
def scm_inventory_source(inventory, project):
    inv_src = InventorySource(
        name="test-scm-inv",
        source_project=project,
        source='scm',
        source_path='inventory_file',
        update_on_project_update=True,
        inventory=inventory,
        scm_last_revision=project.scm_revision)
    with mock.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.update'):
        inv_src.save()
    return inv_src


@pytest.fixture
def inventory_factory(organization):
    def factory(name, org=organization):
        try:
            inv = Inventory.objects.get(name=name, organization=org)
        except Inventory.DoesNotExist:
            inv = Inventory.objects.create(name=name, organization=org)
        return inv
    return factory


@pytest.fixture
def label(organization):
    return organization.labels.create(name="test-label", description="test-label-desc")


@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 notification_template_with_encrypt(organization):
    return NotificationTemplate.objects.create(name='test-notification_template_with_encrypt',
                                               organization=organization,
                                               notification_type="slack",
                                               notification_configuration=dict(channels=["Foo", "Bar"],
                                                                               token="token"))


@pytest.fixture
def notification(notification_template):
    return Notification.objects.create(notification_template=notification_template,
                                       status='successful',
                                       notifications_sent=1,
                                       notification_type='email',
                                       recipients='admin@redhat.com',
                                       subject='email subject')


@pytest.fixture
def job_template_with_survey_passwords(job_template_with_survey_passwords_factory):
    return job_template_with_survey_passwords_factory(persisted=True)


@pytest.fixture
def admin(user):
    return user('admin', True)


@pytest.fixture
def system_auditor(user):
    u = user('an-auditor', False)
    Role.singleton('system_auditor').members.add(u)
    return u


@pytest.fixture
def alice(user):
    return user('alice', False)


@pytest.fixture
def bob(user):
    return user('bob', False)


@pytest.fixture
def rando(user):
    "Rando, the random user that doesn't have access to anything"
    return user('rando', False)


@pytest.fixture
def org_admin(user, organization):
    ret = user('org-admin', False)
    organization.admin_role.members.add(ret)
    organization.member_role.members.add(ret)
    return ret


@pytest.fixture
def org_auditor(user, organization):
    ret = user('org-auditor', False)
    organization.auditor_role.members.add(ret)
    organization.member_role.members.add(ret)
    return ret


@pytest.fixture
def org_member(user, organization):
    ret = user('org-member', False)
    organization.member_role.members.add(ret)
    return ret


@pytest.fixture
def organizations(instance):
    def rf(organization_count=1):
        orgs = []
        for i in range(0, organization_count):
            o = Organization.objects.create(name="test-org-%d" % i, description="test-org-desc")
            orgs.append(o)
        return orgs
    return rf


@pytest.fixture
def group_factory(inventory):
    def g(name):
        try:
            return Group.objects.get(name=name, inventory=inventory)
        except Exception:
            return Group.objects.create(inventory=inventory, name=name)
    return g


@pytest.fixture
def hosts(group_factory):
    group1 = group_factory('group-1')

    def rf(host_count=1):
        hosts = []
        for i in range(0, host_count):
            name = '%s-host-%s' % (group1.name, i)
            (host, created) = group1.inventory.hosts.get_or_create(name=name)
            if created:
                group1.hosts.add(host)
            hosts.append(host)
        return hosts
    return rf


@pytest.fixture
def group(inventory):
    return inventory.groups.create(name='single-group')


@pytest.fixture
def inventory_source(inventory):
    # by making it ec2, the credential is not required
    return InventorySource.objects.create(name='single-inv-src',
                                          inventory=inventory, source='ec2')


@pytest.fixture
def inventory_source_factory(inventory_factory):
    def invsrc(name, source=None, inventory=None):
        if inventory is None:
            inventory = inventory_factory("inv-is-%s" % name)
        if source is None:
            source = 'file'
        try:
            return inventory.inventory_sources.get(name=name)
        except Exception:
            return inventory.inventory_sources.create(name=name, source=source)
    return invsrc


@pytest.fixture
def inventory_update(inventory_source):
    return InventoryUpdate.objects.create(
        inventory_source=inventory_source,
        source=inventory_source.source
    )


@pytest.fixture
def inventory_script(organization):
    return CustomInventoryScript.objects.create(name='test inv script',
                                                organization=organization,
                                                script='#!/usr/bin/python')


@pytest.fixture
def host(group, inventory):
    return group.hosts.create(name='single-host', inventory=inventory)


@pytest.fixture
def permissions():
    return {
        'admin':{'create':True, 'read':True, 'write':True,
                 'update':True, 'delete':True, 'scm_update':True, 'execute':True, 'use':True,},

        'auditor':{'read':True, 'create':False, 'write':False,
                   'update':False, 'delete':False, 'scm_update':False, 'execute':False, 'use':False,},

        'usage':{'read':False, 'create':False, 'write':False,
                 'update':False, 'delete':False, 'scm_update':False, 'execute':False, 'use':True,},
    }


def _request(verb):
    def rf(url, data_or_user=None, user=None, middleware=None, expect=None, **kwargs):
        if type(data_or_user) is User and user is None:
            user = data_or_user
        elif 'data' not in kwargs:
            kwargs['data'] = data_or_user
        if 'format' not in kwargs and 'content_type' not in kwargs:
            kwargs['format'] = 'json'

        request = getattr(APIRequestFactory(), verb)(url, **kwargs)
        request_error = None
        try:
            view, view_args, view_kwargs = resolve(urllib.parse.urlparse(url)[2])
        except Http404 as e:
            request_error = e
        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)

        if not request_error:
            response = view(request, *view_args, **view_kwargs)
        else:
            response = response_for_exception(request, request_error)
        if middleware:
            middleware.process_response(request, response)
        if expect:
            if response.status_code != expect:
                if getattr(response, 'data', None):
                    try:
                        data_copy = response.data.copy()
                        # Make translated strings printable
                        for key, value in response.data.items():
                            if isinstance(value, list):
                                response.data[key] = []
                                for item in value:
                                    response.data[key].append(str(item))
                            else:
                                response.data[key] = str(value)
                    except Exception:
                        response.data = data_copy
            assert response.status_code == expect, 'Response data: {}'.format(
                getattr(response, 'data', None)
            )
        if hasattr(response, 'render'):
            response.render()
        __SWAGGER_REQUESTS__.setdefault(request.path, {})[
            (request.method.lower(), response.status_code)
        ] = (response.get('Content-Type', None), response.content, kwargs.get('data'))
        return response
    return rf


@pytest.fixture
def post():
    return _request('post')


@pytest.fixture
def get():
    return _request('get')


@pytest.fixture
def put():
    return _request('put')


@pytest.fixture
def patch():
    return _request('patch')


@pytest.fixture
def delete():
    return _request('delete')


@pytest.fixture
def head():
    return _request('head')


@pytest.fixture
def options():
    return _request('options')


@pytest.fixture
def ad_hoc_command_factory(inventory, machine_credential, admin):
    def factory(inventory=inventory, credential=machine_credential, initial_state='new', created_by=admin):
        adhoc = AdHocCommand(
            name='test-adhoc', inventory=inventory, credential=credential,
            status=initial_state, created_by=created_by
        )
        adhoc.save()
        return adhoc
    return factory


@pytest.fixture
def job_template():
    return JobTemplate.objects.create(name='test-job_template')


@pytest.fixture
def job_template_labels(organization, job_template):
    job_template.labels.create(name="label-1", organization=organization)
    job_template.labels.create(name="label-2", organization=organization)

    return job_template


@pytest.fixture
def jt_linked(organization, project, inventory, machine_credential, credential, net_credential, vault_credential):
    '''
    A job template with a reasonably complete set of related objects to
    test RBAC and other functionality affected by related objects
    '''
    jt = JobTemplate.objects.create(
        project=project, inventory=inventory, playbook='helloworld.yml',
        organization=organization
    )
    jt.credentials.add(machine_credential, vault_credential, credential, net_credential)
    return jt


@pytest.fixture
def workflow_job_template(organization):
    wjt = WorkflowJobTemplate(name='test-workflow_job_template', organization=organization)
    wjt.save()

    return wjt


@pytest.fixture
def workflow_job_factory(workflow_job_template, admin):
    def factory(workflow_job_template=workflow_job_template, initial_state='new', created_by=admin):
        return workflow_job_template.create_unified_job(_eager_fields={
            'status': initial_state, 'created_by': created_by})
    return factory


@pytest.fixture
def system_job_template():
    sys_jt = SystemJobTemplate(name='test-system_job_template', job_type='cleanup_jobs')
    sys_jt.save()
    return sys_jt


@pytest.fixture
def system_job_factory(system_job_template, admin):
    def factory(system_job_template=system_job_template, initial_state='new', created_by=admin):
        return system_job_template.create_unified_job(_eager_fields={
            'status': initial_state, 'created_by': created_by})
    return factory


def dumps(value):
    return DjangoJSONEncoder().encode(value)


# Taken from https://github.com/django-extensions/django-extensions/blob/54fe88df801d289882a79824be92d823ab7be33e/django_extensions/db/fields/json.py
def get_db_prep_save(self, value, connection, **kwargs):
    """Convert our JSON object to a string before we save"""
    if value is None and self.null:
        return None
    # default values come in as strings; only non-strings should be
    # run through `dumps`
    if not isinstance(value, str):
        value = dumps(value)

    return value


@pytest.fixture
def monkeypatch_jsonbfield_get_db_prep_save(mocker):
    JSONBField.get_db_prep_save = get_db_prep_save


@pytest.fixture
def oauth_application(admin):
    return Application.objects.create(
        name='test app', user=admin, client_type='confidential',
        authorization_grant_type='password'
    )


@pytest.fixture
def sqlite_copy_expert(request):
    # copy_expert is postgres-specific, and SQLite doesn't support it; mock its
    # behavior to test that it writes a file that contains stdout from events
    path = tempfile.mkdtemp(prefix='job-event-stdout')

    def write_stdout(self, sql, fd):
        # simulate postgres copy_expert support with ORM code
        parts = sql.split(' ')
        tablename = parts[parts.index('from') + 1]
        for cls in (JobEvent, AdHocCommandEvent, ProjectUpdateEvent,
                    InventoryUpdateEvent, SystemJobEvent):
            if cls._meta.db_table == tablename:
                for event in cls.objects.order_by('start_line').all():
                    fd.write(event.stdout)

    setattr(SQLiteCursorWrapper, 'copy_expert', write_stdout)
    request.addfinalizer(lambda: shutil.rmtree(path))
    request.addfinalizer(lambda: delattr(SQLiteCursorWrapper, 'copy_expert'))
    return path


@pytest.fixture
def disable_database_settings(mocker):
    m = mocker.patch('awx.conf.settings.SettingsWrapper.all_supported_settings', new_callable=PropertyMock)
    m.return_value = []


@pytest.fixture
def slice_jt_factory(inventory):
    def r(N, jt_kwargs=None):
        for i in range(N):
            inventory.hosts.create(name='foo{}'.format(i))
        if not jt_kwargs:
            jt_kwargs = {}
        return JobTemplate.objects.create(
            name='slice-jt-from-factory',
            job_slice_count=N,
            inventory=inventory,
            **jt_kwargs
        )
    return r


@pytest.fixture
def slice_job_factory(slice_jt_factory):
    def r(N, jt_kwargs=None, prompts=None, spawn=False):
        slice_jt = slice_jt_factory(N, jt_kwargs=jt_kwargs)
        if not prompts:
            prompts = {}
        slice_job = slice_jt.create_unified_job(**prompts)
        if spawn:
            for node in slice_job.workflow_nodes.all():
                # does what the task manager does for spawning workflow jobs
                kv = node.get_job_kwargs()
                job = node.unified_job_template.create_unified_job(**kv)
                node.job = job
                node.save()
        return slice_job
    return r