<commit_before># -*- coding: utf8 -*-
import collections
import json

from io import BytesIO
import botocore
import botocore.stub
import flask
import mock
import os
import random
import string
import zipfile
import unittest
import shutil
import sys
import tempfile
import uuid

from click.globals import resolve_color_default
from click.exceptions import ClickException

from zappa.cli import ZappaCLI, shamelessly_promote, disable_click_colors
from zappa.core import ALB_LAMBDA_ALIAS
from zappa.ext.django_zappa import get_django_wsgi
from zappa.letsencrypt import get_cert_and_update_domain, create_domain_key, create_domain_csr, \
    create_chained_certificate, cleanup, parse_account_key, parse_csr, sign_certificate, encode_certificate,\
    register_account, verify_challenge, gettempdir
from zappa.utilities import (
    conflicts_with_a_neighbouring_module, contains_python_files_or_subdirs,
    detect_django_settings, detect_flask_apps, get_venv_from_python_version,
    human_size, InvalidAwsLambdaName, parse_s3_url, string_to_timestamp,
    titlecase_keys, is_valid_bucket_name, validate_name
)
from zappa.wsgi import create_wsgi_request, common_log
from zappa.core import Zappa, ASSUME_POLICY, ATTACH_POLICY

if sys.version_info[0] < 3:
    from cStringIO import StringIO as OldStringIO

def random_string(length):
    return ''.join(random.choice(string.printable) for _ in range(length))


class TestZappa(unittest.TestCase):
    def setUp(self):
        self.sleep_patch = mock.patch('time.sleep', return_value=None)
        # Tests expect us-east-1.
        # If the user has set a different region in env variables, we set it aside for now and use us-east-1
        self.users_current_region_name = os.environ.get('AWS_DEFAULT_REGION', None)
        os.environ['AWS_DEFAULT_REGION'] = 'us-east-1'
        if not os.environ.get('PLACEBO_MODE') == 'record':
            self.sleep_patch.start()

    def tearDown(self):
        if not os.environ.get('PLACEBO_MODE') == 'record':
            self.sleep_patch.stop()
        del os.environ['AWS_DEFAULT_REGION']
        if self.users_current_region_name is not None:
            # Give the user their AWS region back, we're done testing with us-east-1.
            os.environ['AWS_DEFAULT_REGION'] = self.users_current_region_name

    ##
    # Sanity Tests
    ##

    def test_test(self):
        self.assertTrue(True)
    ##
    # Basic Tests
    ##

    def test_zappa(self):
        self.assertTrue(True)
        Zappa()

    def test_disable_click_colors(self):
        disable_click_colors()
        assert resolve_color_default() is False

    @mock.patch('zappa.core.find_packages')
    @mock.patch('os.remove')
    def test_copy_editable_packages(self, mock_remove, mock_find_packages):
        virtual_env = os.environ.get("VIRTUAL_ENV")
        if not virtual_env:
            return self.skipTest(
                "test_copy_editable_packages must be run in a virtualenv")

        temp_package_dir = tempfile.mkdtemp()
        try:
            egg_links = [os.path.join(
                virtual_env, "lib", get_venv_from_python_version(),
                "site-packages", "test-copy-editable-packages.egg-link")]
            egg_path = "/some/other/directory/package"
            mock_find_packages.return_value = [
                "package", "package.subpackage", "package.another"]
            temp_egg_link = os.path.join(
                temp_package_dir, 'package-python.egg-link')

            z = Zappa()
            mock_open = mock.mock_open(read_data=egg_path.encode("utf-8"))
            with mock.patch("zappa.core.open", mock_open), \
                    mock.patch("glob.glob") as mock_glob, \
                    mock.patch("zappa.core.copytree") as mock_copytree:
                # we use glob.glob to get the egg-links in the temp packages
                # directory
                mock_glob.return_value = [temp_egg_link]

                z.copy_editable_packages(egg_links, temp_package_dir)

                # make sure we copied the right directories
                mock_copytree.assert_called_with(
                    os.path.join(egg_path, 'package'),
                    os.path.join(temp_package_dir, 'package'),
                    metadata=False, symlinks=False
                )
                self.assertEqual(mock_copytree.call_count, 1)

                # make sure it removes the egg-link from the temp packages
                # directory
                mock_remove.assert_called_with(temp_egg_link)
                self.assertEqual(mock_remove.call_count, 1)
        finally:
            shutil.rmtree(temp_package_dir)

        return

    def test_create_lambda_package(self):
        # mock the pkg_resources.WorkingSet() to include a known package in lambda_packages so that the code
        # for zipping pre-compiled packages gets called
        mock_installed_packages = {'psycopg2': '2.6.1'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.6')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_get_manylinux_python36(self):
        z = Zappa(runtime='python3.6')
        self.assertIsNotNone(z.get_cached_manylinux_wheel('psycopg2', '2.7.1'))
        self.assertIsNone(z.get_cached_manylinux_wheel('derpderpderpderp', '0.0'))

        # mock with a known manylinux wheel package so that code for downloading them gets invoked
        mock_installed_packages = {'psycopg2': '2.7.1'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.6')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_get_manylinux_python37(self):
        z = Zappa(runtime='python3.7')
        self.assertIsNotNone(z.get_cached_manylinux_wheel('psycopg2', '2.7.6'))
        self.assertIsNone(z.get_cached_manylinux_wheel('derp_no_such_thing', '0.0'))

        # mock with a known manylinux wheel package so that code for downloading them gets invoked
        mock_installed_packages = {'psycopg2': '2.7.6'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.7')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_get_manylinux_python38(self):
        z = Zappa(runtime='python3.8')
        self.assertIsNotNone(z.get_cached_manylinux_wheel('psycopg2-binary', '2.8.4'))
        self.assertIsNone(z.get_cached_manylinux_wheel('derp_no_such_thing', '0.0'))

        # mock with a known manylinux wheel package so that code for downloading them gets invoked
        mock_installed_packages = {'psycopg2-binary': '2.8.4'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.8')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_should_use_lambda_packages(self):
        z = Zappa(runtime='python3.6')

        self.assertTrue(z.have_any_lambda_package_version('mysqlclient'))
        self.assertFalse(z.have_any_lambda_package_version('no_package'))

        self.assertTrue(z.have_correct_lambda_package_version('mysqlclient', '1.3.12'))
        self.assertFalse(z.have_correct_lambda_package_version('mysqlclient', '1.4.12'))

    def test_getting_installed_packages(self, *args):
        z = Zappa(runtime='python3.6')

        # mock pkg_resources call to be same as what our mocked site packages dir has
        mock_package = collections.namedtuple('mock_package', ['project_name', 'version', 'location'])
        mock_pip_installed_packages = [mock_package('super_package', '0.1', '/venv/site-packages')]

        with mock.patch('os.path.isdir', return_value=True):
            with mock.patch('os.listdir', return_value=['super_package']):
                import pkg_resources  # this gets called in non-test Zappa mode
                with mock.patch('pkg_resources.WorkingSet', return_value=mock_pip_installed_packages):
                    self.assertDictEqual(z.get_installed_packages('',''), {'super_package' : '0.1'})

    def test_getting_installed_packages_mixed_case_location(self, *args):<commit_msg>Revert "fix"<commit_after># -*- coding: utf8 -*-
import collections
import json

from io import BytesIO
import botocore
import botocore.stub
import flask
import mock
import os
import random
import string
import zipfile
import unittest
import shutil
import sys
import tempfile
import uuid

from click.globals import resolve_color_default
from click.exceptions import ClickException

from zappa.cli import ZappaCLI, shamelessly_promote, disable_click_colors
from zappa.core import ALB_LAMBDA_ALIAS
from zappa.ext.django_zappa import get_django_wsgi
from zappa.letsencrypt import get_cert_and_update_domain, create_domain_key, create_domain_csr, \
    create_chained_certificate, cleanup, parse_account_key, parse_csr, sign_certificate, encode_certificate,\
    register_account, verify_challenge, gettempdir
from zappa.utilities import (
    conflicts_with_a_neighbouring_module, contains_python_files_or_subdirs,
    detect_django_settings, detect_flask_apps, get_venv_from_python_version,
    human_size, InvalidAwsLambdaName, parse_s3_url, string_to_timestamp,
    titlecase_keys, is_valid_bucket_name, validate_name
)
from zappa.wsgi import create_wsgi_request, common_log
from zappa.core import Zappa, ASSUME_POLICY, ATTACH_POLICY

if sys.version_info[0] < 3:
    from cStringIO import StringIO as OldStringIO

def random_string(length):
    return ''.join(random.choice(string.printable) for _ in range(length))


class TestZappa(unittest.TestCase):
    def setUp(self):
        self.sleep_patch = mock.patch('time.sleep', return_value=None)
        # Tests expect us-east-1.
        # If the user has set a different region in env variables, we set it aside for now and use us-east-1
        self.users_current_region_name = os.environ.get('AWS_DEFAULT_REGION', None)
        os.environ['AWS_DEFAULT_REGION'] = 'us-east-1'
        if not os.environ.get('PLACEBO_MODE') == 'record':
            self.sleep_patch.start()

    def tearDown(self):
        if not os.environ.get('PLACEBO_MODE') == 'record':
            self.sleep_patch.stop()
        del os.environ['AWS_DEFAULT_REGION']
        if self.users_current_region_name is not None:
            # Give the user their AWS region back, we're done testing with us-east-1.
            os.environ['AWS_DEFAULT_REGION'] = self.users_current_region_name

    ##
    # Sanity Tests
    ##

    def test_test(self):
        self.assertTrue(True)
    ##
    # Basic Tests
    ##

    def test_zappa(self):
        self.assertTrue(True)
        Zappa()

    def test_disable_click_colors(self):
        disable_click_colors()
        assert resolve_color_default() is False

    @mock.patch('zappa.core.find_packages')
    @mock.patch('os.remove')
    def test_copy_editable_packages(self, mock_remove, mock_find_packages):
        virtual_env = os.environ.get("VIRTUAL_ENV")
        if not virtual_env:
            return self.skipTest(
                "test_copy_editable_packages must be run in a virtualenv")

        temp_package_dir = tempfile.mkdtemp()
        try:
            egg_links = [os.path.join(
                virtual_env, "lib", get_venv_from_python_version(),
                "site-packages", "test-copy-editable-packages.egg-link")]
            egg_path = "/some/other/directory/package"
            mock_find_packages.return_value = [
                "package", "package.subpackage", "package.another"]
            temp_egg_link = os.path.join(
                temp_package_dir, 'package-python.egg-link')

            z = Zappa()
            mock_open = mock.mock_open(read_data=egg_path.encode("utf-8"))
            with mock.patch("zappa.core.open", mock_open), \
                    mock.patch("glob.glob") as mock_glob, \
                    mock.patch("zappa.core.copytree") as mock_copytree:
                # we use glob.glob to get the egg-links in the temp packages
                # directory
                mock_glob.return_value = [temp_egg_link]

                z.copy_editable_packages(egg_links, temp_package_dir)

                # make sure we copied the right directories
                mock_copytree.assert_called_with(
                    os.path.join(egg_path, 'package'),
                    os.path.join(temp_package_dir, 'package'),
                    metadata=False, symlinks=False
                )
                self.assertEqual(mock_copytree.call_count, 1)

                # make sure it removes the egg-link from the temp packages
                # directory
                mock_remove.assert_called_with(temp_egg_link)
                self.assertEqual(mock_remove.call_count, 1)
        finally:
            shutil.rmtree(temp_package_dir)

        return

    def test_create_lambda_package(self):
        # mock the pkg_resources.WorkingSet() to include a known package in lambda_packages so that the code
        # for zipping pre-compiled packages gets called
        mock_installed_packages = {'psycopg2': '2.6.1'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.6')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_get_manylinux_python36(self):
        z = Zappa(runtime='python3.6')
        self.assertIsNotNone(z.get_cached_manylinux_wheel('psycopg2', '2.7.1'))
        self.assertIsNone(z.get_cached_manylinux_wheel('derpderpderpderp', '0.0'))

        # mock with a known manylinux wheel package so that code for downloading them gets invoked
        mock_installed_packages = {'psycopg2': '2.7.1'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.6')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_get_manylinux_python37(self):
        z = Zappa(runtime='python3.7')
        self.assertIsNotNone(z.get_cached_manylinux_wheel('psycopg2', '2.7.6'))
        self.assertIsNone(z.get_cached_manylinux_wheel('derp_no_such_thing', '0.0'))

        # mock with a known manylinux wheel package so that code for downloading them gets invoked
        mock_installed_packages = {'psycopg2': '2.7.6'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.7')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_get_manylinux_python38(self):
        z = Zappa(runtime='python3.8')
        self.assertIsNotNone(z.get_cached_manylinux_wheel('psycopg2-binary', '2.8.4'))
        self.assertIsNone(z.get_cached_manylinux_wheel('derp_no_such_thing', '0.0'))

        # mock with a known manylinux wheel package so that code for downloading them gets invoked
        mock_installed_packages = {'psycopg2-binary': '2.8.4'}
        with mock.patch('zappa.core.Zappa.get_installed_packages', return_value=mock_installed_packages):
            z = Zappa(runtime='python3.8')
            path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
            self.assertTrue(os.path.isfile(path))
            os.remove(path)

    def test_should_use_lambda_packages(self):
        z = Zappa(runtime='python3.6')

        self.assertTrue(z.have_any_lambda_package_version('mysqlclient'))
        self.assertFalse(z.have_any_lambda_package_version('no_package'))

        self.assertTrue(z.have_correct_lambda_package_version('mysqlclient', '1.3.12'))
        self.assertFalse(z.have_correct_lambda_package_version('mysqlclient', '1.4.12'))

    def test_getting_installed_packages(self, *args):
        z = Zappa(runtime='python2.6')

        # mock pkg_resources call to be same as what our mocked site packages dir has
        mock_package = collections.namedtuple('mock_package', ['project_name', 'version', 'location'])
        mock_pip_installed_packages = [mock_package('super_package', '0.1', '/venv/site-packages')]

        with mock.patch('os.path.isdir', return_value=True):
            with mock.patch('os.listdir', return_value=['super_package']):
                import pkg_resources  # this gets called in non-test Zappa mode
                with mock.patch('pkg_resources.WorkingSet', return_value=mock_pip_installed_packages):
                    self.assertDictEqual(z.get_installed_packages('',''), {'super_package' : '0.1'})

    def test_getting_installed_packages_mixed_case_location(self, *args):
{'fraction_non_alphanumeric': 0.09865729268714343, 'fraction_numerical': 0.009335561574367544, 'mean_word_length': 3.0414531904983697, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7436764', 'n_tokens_mistral': 5438, 'n_tokens_neox': 5287, 'n_words': 1010}