<commit_before>import csv
import datetime
import json
import os
import StringIO
from time import sleep

from django.conf import settings
from django.core.files.storage import get_storage_class
from django.core.urlresolvers import reverse
from django.utils.dateparse import parse_datetime
from xlrd import open_workbook

from onadata.apps.main.views import delete_data
from onadata.apps.main.tests.test_base import TestBase
from onadata.apps.viewer.tests.export_helpers import viewer_fixture_path
from onadata.apps.viewer.views import delete_export, export_list,\
    create_export, export_progress, export_download
from onadata.apps.viewer.xls_writer import XlsWriter
from onadata.apps.viewer.models.export import Export
from onadata.apps.viewer.models.parsed_instance import ParsedInstance
from onadata.apps.logger.models import Instance
from onadata.apps.viewer.tasks import create_xls_export
from onadata.libs.utils.export_tools import generate_export,\
    increment_index_in_filename, dict_to_joined_export

AMBULANCE_KEY = 'transport/available_transportation_types_to_referral_fac'\
                'ility/ambulance'
AMBULANCE_KEY_DOTS = 'transport.available_transportation_types_to_referra'\
                     'l_facility.ambulance'


def _main_fixture_path(instance_name):
    return os.path.join(settings.PROJECT_ROOT, 'apps', 'main', 'tests',
                        'fixtures', 'transportation', 'instances_w_uuid',
                        instance_name, instance_name + '.xml')


class TestExports(TestBase):
    def setUp(self):
        super(TestExports, self).setUp()
        self._submission_time = parse_datetime('2013-02-18 15:54:01Z')

    def test_unique_xls_sheet_name(self):
        xls_writer = XlsWriter()
        xls_writer.add_sheet('section9_pit_latrine_with_slab_group')
        xls_writer.add_sheet('section9_pit_latrine_without_slab_group')
        # create a set of sheet names keys
        sheet_names_set = set(xls_writer._sheets.keys())
        self.assertEqual(len(sheet_names_set), 2)

    def test_csv_http_response(self):
        self._publish_transportation_form()
        survey = self.surveys[0]
        self._make_submission(
            os.path.join(
                self.this_directory, 'fixtures', 'transportation',
                'instances', survey, survey + '.xml'),
            forced_submission_time=self._submission_time)
        response = self.client.get(reverse(
            'csv_export',
            kwargs={
                'username': self.user.username,
                'id_string': self.xform.id_string
            }))
        self.assertEqual(response.status_code, 200)
        test_file_path = viewer_fixture_path('transportation.csv')
        content = self._get_response_content(response)
        with open(test_file_path, 'r') as test_file:
            self.assertEqual(content, test_file.read())

    def test_csv_without_na_values(self):
        self._publish_transportation_form()
        survey = self.surveys[0]
        self._make_submission(
            os.path.join(
                self.this_directory, 'fixtures', 'transportation',
                'instances', survey, survey + '.xml'),
            forced_submission_time=self._submission_time)
        na_rep_restore = settings.NA_REP
        settings.NA_REP = u''
        response = self.client.get(reverse(
            'csv_export',
            kwargs={
                'username': self.user.username,
                'id_string': self.xform.id_string
            }))
        self.assertEqual(response.status_code, 200)
        test_file_path = viewer_fixture_path('transportation_without_na.csv')
        content = self._get_response_content(response)
        with open(test_file_path, 'r') as test_file:
            self.assertEqual(content, test_file.read())
        settings.NA_REP = na_rep_restore

    def test_responses_for_empty_exports(self):
        self._publish_transportation_form()
        # test csv though xls uses the same view
        url = reverse(
            'csv_export',
            kwargs={
                'username': self.user.username,
                'id_string': self.xform.id_string
            }
        )
        self.response = self.client.get(url)
        self.assertEqual(self.response.status_code, 404)
        self.assertIn('text/html', self.response['content-type'])

    def test_create_export(self):
        self._publish_transportation_form_and_submit_instance()
        storage = get_storage_class()()
        # test xls
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        self.assertTrue(storage.exists(export.filepath))
        path, ext = os.path.splitext(export.filename)
        self.assertEqual(ext, '.xls')

        # test csv
        export = generate_export(Export.CSV_EXPORT, 'csv', self.user.username,
                                 self.xform.id_string)
        self.assertTrue(storage.exists(export.filepath))
        path, ext = os.path.splitext(export.filename)
        self.assertEqual(ext, '.csv')

        # test xls with existing export_id
        existing_export = Export.objects.create(xform=self.xform,
                                                export_type=Export.XLS_EXPORT)
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string, existing_export.id)
        self.assertEqual(existing_export.id, export.id)

    def test_delete_file_on_export_delete(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        storage = get_storage_class()()
        self.assertTrue(storage.exists(export.filepath))
        # delete export object
        export.delete()
        self.assertFalse(storage.exists(export.filepath))

    def test_graceful_exit_on_export_delete_if_file_doesnt_exist(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        storage = get_storage_class()()
        # delete file
        storage.delete(export.filepath)
        self.assertFalse(storage.exists(export.filepath))
        # clear filename, like it would be in an incomplete export
        export.filename = None
        export.filedir = None
        export.save()
        # delete export record, which should try to delete file as well
        delete_url = reverse(delete_export, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': 'xls'
        })
        post_data = {'export_id': export.id}
        response = self.client.post(delete_url, post_data)
        self.assertEqual(response.status_code, 302)

    def test_delete_oldest_export_on_limit(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        # create first export
        first_export = generate_export(
            Export.XLS_EXPORT, 'xls', self.user.username, self.xform.id_string)
        self.assertIsNotNone(first_export.pk)
        # create exports that exceed set limit
        for i in range(Export.MAX_EXPORTS):
            generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                            self.xform.id_string)
        # first export should be deleted
        exports = Export.objects.filter(id=first_export.id)
        self.assertEqual(len(exports), 0)

    def test_create_export_url(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        num_exports = Export.objects.count()
        # create export
        create_export_url = reverse(create_export, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': Export.XLS_EXPORT
        })
        response = self.client.post(create_export_url)
        self.assertEqual(response.status_code, 302)
        self.assertEqual(Export.objects.count(), num_exports + 1)

    def test_delete_export_url(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        # create export
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        exports = Export.objects.filter(id=export.id)
        self.assertEqual(len(exports), 1)
        delete_url = reverse(delete_export, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': 'xls'
        })
        post_data = {'export_id': export.id}
        response = self.client.post(delete_url, post_data)
        self.assertEqual(response.status_code, 302)
        exports = Export.objects.filter(id=export.id)
        self.assertEqual(len(exports), 0)

    def test_export_progress_output(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        # create exports
        for i in range(2):
            generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                            self.xform.id_string)
        self.assertEqual(Export.objects.count(), 2)
        # progress for multiple exports
        progress_url = reverse(export_progress, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': 'xls'
        })
        get_data = {'export_ids': [e.id for e in Export.objects.all()]}
        response = self.client.get(progress_url, get_data)
        content = json.loads(response.content)
        self.assertEqual(len(content), 2)
        self.assertEqual(sorted(['url', 'export_id', 'complete', 'filename']),
                         sorted(content[0].keys()))

    def test_auto_export_if_none_exists(self):
        self._publish_transportation_form()<commit_msg>UK: add check for create_export and delete_export, requires login<commit_after>import csv
import datetime
import json
import os
import StringIO
from time import sleep

from django.conf import settings
from django.core.files.storage import get_storage_class
from django.core.urlresolvers import reverse
from django.utils.dateparse import parse_datetime
from xlrd import open_workbook

from onadata.apps.main.views import delete_data
from onadata.apps.main.tests.test_base import TestBase
from onadata.apps.viewer.tests.export_helpers import viewer_fixture_path
from onadata.apps.viewer.views import delete_export, export_list,\
    create_export, export_progress, export_download
from onadata.apps.viewer.xls_writer import XlsWriter
from onadata.apps.viewer.models.export import Export
from onadata.apps.viewer.models.parsed_instance import ParsedInstance
from onadata.apps.logger.models import Instance
from onadata.apps.viewer.tasks import create_xls_export
from onadata.libs.utils.export_tools import generate_export,\
    increment_index_in_filename, dict_to_joined_export

AMBULANCE_KEY = 'transport/available_transportation_types_to_referral_fac'\
                'ility/ambulance'
AMBULANCE_KEY_DOTS = 'transport.available_transportation_types_to_referra'\
                     'l_facility.ambulance'


def _main_fixture_path(instance_name):
    return os.path.join(settings.PROJECT_ROOT, 'apps', 'main', 'tests',
                        'fixtures', 'transportation', 'instances_w_uuid',
                        instance_name, instance_name + '.xml')


class TestExports(TestBase):
    def setUp(self):
        super(TestExports, self).setUp()
        self._submission_time = parse_datetime('2013-02-18 15:54:01Z')

    def test_unique_xls_sheet_name(self):
        xls_writer = XlsWriter()
        xls_writer.add_sheet('section9_pit_latrine_with_slab_group')
        xls_writer.add_sheet('section9_pit_latrine_without_slab_group')
        # create a set of sheet names keys
        sheet_names_set = set(xls_writer._sheets.keys())
        self.assertEqual(len(sheet_names_set), 2)

    def test_csv_http_response(self):
        self._publish_transportation_form()
        survey = self.surveys[0]
        self._make_submission(
            os.path.join(
                self.this_directory, 'fixtures', 'transportation',
                'instances', survey, survey + '.xml'),
            forced_submission_time=self._submission_time)
        response = self.client.get(reverse(
            'csv_export',
            kwargs={
                'username': self.user.username,
                'id_string': self.xform.id_string
            }))
        self.assertEqual(response.status_code, 200)
        test_file_path = viewer_fixture_path('transportation.csv')
        content = self._get_response_content(response)
        with open(test_file_path, 'r') as test_file:
            self.assertEqual(content, test_file.read())

    def test_csv_without_na_values(self):
        self._publish_transportation_form()
        survey = self.surveys[0]
        self._make_submission(
            os.path.join(
                self.this_directory, 'fixtures', 'transportation',
                'instances', survey, survey + '.xml'),
            forced_submission_time=self._submission_time)
        na_rep_restore = settings.NA_REP
        settings.NA_REP = u''
        response = self.client.get(reverse(
            'csv_export',
            kwargs={
                'username': self.user.username,
                'id_string': self.xform.id_string
            }))
        self.assertEqual(response.status_code, 200)
        test_file_path = viewer_fixture_path('transportation_without_na.csv')
        content = self._get_response_content(response)
        with open(test_file_path, 'r') as test_file:
            self.assertEqual(content, test_file.read())
        settings.NA_REP = na_rep_restore

    def test_responses_for_empty_exports(self):
        self._publish_transportation_form()
        # test csv though xls uses the same view
        url = reverse(
            'csv_export',
            kwargs={
                'username': self.user.username,
                'id_string': self.xform.id_string
            }
        )
        self.response = self.client.get(url)
        self.assertEqual(self.response.status_code, 404)
        self.assertIn('text/html', self.response['content-type'])

    def test_create_export(self):
        self._publish_transportation_form_and_submit_instance()
        storage = get_storage_class()()
        # test xls
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        self.assertTrue(storage.exists(export.filepath))
        path, ext = os.path.splitext(export.filename)
        self.assertEqual(ext, '.xls')

        # test csv
        export = generate_export(Export.CSV_EXPORT, 'csv', self.user.username,
                                 self.xform.id_string)
        self.assertTrue(storage.exists(export.filepath))
        path, ext = os.path.splitext(export.filename)
        self.assertEqual(ext, '.csv')

        # test xls with existing export_id
        existing_export = Export.objects.create(xform=self.xform,
                                                export_type=Export.XLS_EXPORT)
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string, existing_export.id)
        self.assertEqual(existing_export.id, export.id)

    def test_delete_file_on_export_delete(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        storage = get_storage_class()()
        self.assertTrue(storage.exists(export.filepath))
        # delete export object
        export.delete()
        self.assertFalse(storage.exists(export.filepath))

    def test_graceful_exit_on_export_delete_if_file_doesnt_exist(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        storage = get_storage_class()()
        # delete file
        storage.delete(export.filepath)
        self.assertFalse(storage.exists(export.filepath))
        # clear filename, like it would be in an incomplete export
        export.filename = None
        export.filedir = None
        export.save()
        # delete export record, which should try to delete file as well
        delete_url = reverse(delete_export, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': 'xls'
        })
        post_data = {'export_id': export.id}
        response = self.client.post(delete_url, post_data)
        self.assertEqual(response.status_code, 302)

    def test_delete_oldest_export_on_limit(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        # create first export
        first_export = generate_export(
            Export.XLS_EXPORT, 'xls', self.user.username, self.xform.id_string)
        self.assertIsNotNone(first_export.pk)
        # create exports that exceed set limit
        for i in range(Export.MAX_EXPORTS):
            generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                            self.xform.id_string)
        # first export should be deleted
        exports = Export.objects.filter(id=first_export.id)
        self.assertEqual(len(exports), 0)

    def test_create_export_url(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        num_exports = Export.objects.count()
        # create export
        create_export_url = reverse(create_export, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': Export.XLS_EXPORT
        })

        # anonymous user has to login first
        response = self.anon.post(create_export_url)
        self.assertEqual(response.status_code, 302)
        self.assertIn("/accounts/login", response['location'])

        response = self.client.post(create_export_url)
        self.assertEqual(response.status_code, 302)
        self.assertEqual(Export.objects.count(), num_exports + 1)

    def test_delete_export_url(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        # create export
        export = generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                                 self.xform.id_string)
        exports = Export.objects.filter(id=export.id)
        self.assertEqual(len(exports), 1)
        delete_url = reverse(delete_export, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': 'xls'
        })
        post_data = {'export_id': export.id}

        # anonymous user has to login first
        response = self.anon.post(delete_url, post_data)
        self.assertEqual(response.status_code, 302)
        self.assertIn("/accounts/login", response['location'])

        response = self.client.post(delete_url, post_data)
        self.assertEqual(response.status_code, 302)
        exports = Export.objects.filter(id=export.id)
        self.assertEqual(len(exports), 0)

    def test_export_progress_output(self):
        self._publish_transportation_form()
        self._submit_transport_instance()
        # create exports
        for i in range(2):
            generate_export(Export.XLS_EXPORT, 'xls', self.user.username,
                            self.xform.id_string)
        self.assertEqual(Export.objects.count(), 2)
        # progress for multiple exports
        progress_url = reverse(export_progress, kwargs={
            'username': self.user.username,
            'id_string': self.xform.id_string,
            'export_type': 'xls'
        })
        get_data = {'export_ids': [e.id for e in Export.objects.all()]}
        response = self.client.get(progress_url, get_data)
        content = json.loads(response.content)
        self.assertEqual(len(content), 2)
        self.assertEqual(sorted(['url', 'export_id', 'complete', 'filename']),
                         sorted(content[0].keys()))

    def test_auto_export_if_none_exists(self):
        self._publish_transportation_form()
{'fraction_non_alphanumeric': 0.09563158403401953, 'fraction_numerical': 0.004639025804581038, 'mean_word_length': 2.565644383184011, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2481297', 'n_tokens_mistral': 6045, 'n_tokens_neox': 5864, 'n_words': 1028}