API Views

SaveCsvSpecies

Bases: APIView

API to save csv file into the database.

SpeciesUploader

Bases: APIView

API to upload csv file.

UploadSpeciesStatus

Bases: APIView

Check upload species status.

Admin

OwnedSpeciesAdmin

Bases: ModelAdmin

Admin page for OwnedSpecies model

TaxonAdmin

Bases: ModelAdmin

Admin page for Taxon model

TaxonRankAdmin

Bases: ModelAdmin

Admin page for TaxonRank model

Factories

TaxonF

Bases: DjangoModelFactory

Taxon model factory.

TaxonRankFactory

Bases: DjangoModelFactory

taxon rank factory

TaxonSurveyMethodF

Bases: DjangoModelFactory

Taxon Survey Method factory.

Forms

Species forms.

TaxonForm

Bases: ModelForm

Taxon form.

Models

Species models.

OwnedSpecies

Bases: Model

Owned species mdoel.

Taxon

Bases: Model

Taxon model.

TaxonRank

Bases: Model

Taxon rank model.

TaxonSurveyMethod

Bases: Model

taxon survey methods

Serializers

FrontPageTaxonSerializer

Bases: ModelSerializer

Display species data on FrontPage.

TaxonSerializer

Bases: ModelSerializer

Species serializer

TrendPageTaxonSerializer

Bases: FrontPageTaxonSerializer

Display species data on TrendPage.

Scripts

SpeciesCSVUpload

SpeciesCSVUpload()

Bases: object

Source code in django_project/species/scripts/data_upload.py

def __init__(self):
    self.upload_session = UploadSpeciesCSV.objects.none()
    self.error_list = []
    self.created_list = 0
    self.existed_list = 0
    self.headers = []
    self.total_rows = 0
    self.row_error = []
    self.csv_dict_reader = None
    self.species_id_list = []

check_compulsory_fields

check_compulsory_fields(row)

Check if compulsory fields are empty.

Source code in django_project/species/scripts/data_upload.py

def check_compulsory_fields(self, row):
    """Check if compulsory fields are empty."""

    for field in COMPULSORY_FIELDS:
        if not self.row_value(row, field):
            self.error_row(
                message="The value of the compulsory field {} "
                        "is empty.".format(field)
            )

error_file

error_file(row)

Write to error file

Parameters:

Name	Type	Description	Default
row		error data	required

Source code in django_project/species/scripts/data_upload.py

def error_file(self, row):
    """
    Write to error file
    :param row: error data
    """
    if len(self.row_error) > 0:
        logger.log(
            level=logging.ERROR,
            msg=' '.join(self.row_error)
        )
        row['error_message'] = ' '.join(self.row_error)
        self.error_list.append(row)

error_row

error_row(message)

Get error in row

Parameters:

Name	Type	Description	Default
message		error message for a row	required

Source code in django_project/species/scripts/data_upload.py

def error_row(self, message):
    """
    Get error in row
    :param message: error message for a row
    """
    self.row_error.append(message)

finish

finish(headers)

Finishing the csv upload process

Source code in django_project/species/scripts/data_upload.py

def finish(self, headers):
    """
    Finishing the csv upload process
    """
    file_name = (
        self.upload_session.process_file.name.replace(
            'species/', '')
    )
    file_path = (
        self.upload_session.process_file.path.replace(file_name, '')
    )

    # Create error file
    # TODO : Done it simultaneously with file processing
    if self.error_list:
        error_headers = copy.deepcopy(headers)
        if 'error_message' not in error_headers:
            error_headers.insert(0, 'error_message')
        error_file_path = '{path}error_{name}'.format(
            path=file_path,
            name=file_name
        )

        excel_error = None

        if error_file_path.endswith('.xlsx'):
            excel_error = error_file_path
            logger.log(
                level=logging.ERROR,
                msg=str(excel_error)
            )
            with tempfile.NamedTemporaryFile(mode='w', delete=False)\
                    as csv_file:
                error_file_path = csv_file.name

        with open(error_file_path, mode='w') as csv_file:
            writer = csv.writer(
                csv_file, delimiter=',', quotechar='"',
                quoting=csv.QUOTE_MINIMAL)
            writer.writerow(error_headers)
            for data in self.error_list:
                data_list = []
                for key in error_headers:
                    try:
                        data_list.append(data[key])
                    except KeyError:
                        continue
                writer.writerow(data_list)

        if excel_error:
            with pd.ExcelWriter(excel_error, engine='openpyxl', mode='w')\
                    as writer:
                dataframe = pd.read_csv(error_file_path)
                dataframe.to_excel(
                    writer,
                    sheet_name=SHEET_TITLE,
                    index=False
                )

        self.upload_session.error_file.name = (
            'species/error_{}'.format(
                file_name
            )
        )

    # Create success message
    success_message = None
    if self.created_list > 0 and self.existed_list == 0:
        success_message = "{} rows uploaded successfully." \
                          "".format(self.created_list)

    if self.existed_list > 0 and self.created_list == 0:
        success_message = "{} rows already exist and have been " \
                          "overwritten." \
                          "".format(self.existed_list)

    if self.existed_list > 0 and self.created_list > 0:
        success_message = "{} rows already exist and have been " \
                          "overwritten. {} " \
                          "rows uploaded successfully." \
                          "".format(self.existed_list, self.created_list)

    if success_message:
        self.upload_session.success_notes = success_message

    if self.total_rows == 0:
        self.upload_session.error_notes = (
            'You have uploaded empty spreadsheet, please check again.'
        )

    self.upload_session.processed = True
    self.upload_session.progress = 'Finished'
    self.upload_session.save()
    if self.created_list > 0 or self.existed_list > 0:
        mark_model_output_as_outdated_by_species_list(self.species_id_list)

population_estimate_category

population_estimate_category(row)

Save Population estimate category.

Source code in django_project/species/scripts/data_upload.py

def population_estimate_category(self, row):
    """ Save Population estimate category.
    """
    pop_est = self.row_value(row, POPULATION_ESTIMATE_CATEGORY)
    if pop_est:
        try:
            pop_est_cat = PopulationEstimateCategory.objects.get(
                name__iexact=pop_est
            )
        except PopulationEstimateCategory.DoesNotExist:
            self.error_row(
                f"Population estimate category '{pop_est}' does not exist"
            )
            return None
        return pop_est_cat
    return None

population_status

population_status(row)

Fetch Population status.

Source code in django_project/species/scripts/data_upload.py

def population_status(self, row):
    """ Fetch Population status.
    """
    pop_st = self.row_value(row, POPULATION_STATUS)
    if pop_st:
        try:
            population_st = PopulationStatus.objects.get(
                name__iexact=pop_st
            )
        except PopulationStatus.DoesNotExist:
            self.error_row(
                f"Population status '{pop_st}' does not exist"
            )
            return None
        return population_st
    return None

presence

presence(row)

Fetch presence value.

Source code in django_project/species/scripts/data_upload.py

def presence(self, row):
    """Fetch presence value."""
    presence = self.row_value(row, PRESENCE)
    value = None
    if presence:
        value = map_string_to_value(presence, PRESENCE_VALUE_MAPPING)
        if value is None:
            self.error_row(
                f"Presence '{presence}' does not exist"
            )
    return value

process_csv_dict_reader

process_csv_dict_reader()

Read and process data from csv file

Source code in django_project/species/scripts/data_upload.py

def process_csv_dict_reader(self):
    """
    Read and process data from csv file
    """
    index = 1
    self.created_list = 0
    self.existed_list = 0
    for row in self.csv_dict_reader:
        self.row_error = []
        if UploadSpeciesCSV.objects.get(
                id=self.upload_session.id).canceled:
            return
        logger.debug(row)
        self.upload_session.progress = '{index}/{total}'.format(
            index=index,
            total=self.total_rows
        )
        self.upload_session.save()
        index += 1
        self.process_data(row=row)
        self.error_file(row)
    self.finish(self.csv_dict_reader.fieldnames)

process_data

process_data(row)

Processing row of csv file.

Source code in django_project/species/scripts/data_upload.py

def process_data(self, row):
    """Processing row of csv file."""

    # check compulsory fields
    self.check_compulsory_fields(row)
    property = taxon = None
    property_code = self.row_value(row, PROPERTY)
    if property_code:
        property = self.get_property(property_code)

        if not property:
            self.error_row(
                message="Property code {} doesn't match the selected "
                        "property. Please replace it with {}.".format(
                            self.row_value(row, PROPERTY),
                            self.upload_session.property.short_code)
            )

    scientific_name = self.row_value(row, SCIENTIFIC_NAME)
    common_name = self.row_value(row, COMMON_NAME)
    if scientific_name and common_name:
        taxon = self.get_taxon(common_name, scientific_name)
        if not taxon:
            self.error_row(
                message="{} doesn't exist in the "
                        "database. Please select species available "
                        "in the dropdown only.".format(
                            self.row_value(row, SCIENTIFIC_NAME)
                        )
            )
        elif taxon.id not in self.species_id_list:
            self.species_id_list.append(taxon.id)

    area_available_to_species = self.row_value(row, AREA)
    # validate area_available_to_species must be greater than 0 and
    # less than property size
    area_available_to_species_num = round(
        string_to_number(area_available_to_species))
    property_area = round(self.upload_session.property.property_size_ha)
    if (
        area_available_to_species_num <= 0 or
        area_available_to_species_num >
        property_area
    ):
        self.error_row(
            message="Area available to species must be greater than 0 "
                    "and less than or equal to property area size "
                    f"({property_area} ha)."
        )

    survey_method = self.survey_method(row)
    survey_other = self.row_value(row, IF_OTHER_SURVEY)
    sur_other = None
    if survey_method and survey_method.name == IF_OTHER_SURVEY_VAL:
        if not survey_other:
            self.error_row(
                message="The value of field {} "
                        "is empty.".format(IF_OTHER_SURVEY)
            )
        sur_other = survey_other

    open_close_system = self.open_close_system(row)
    population_status = self.population_status(row)
    population_estimate = self.population_estimate_category(row)
    population_other = self.row_value(row, IF_OTHER_POPULATION)
    pop_other = None
    if population_estimate and \
            population_estimate.name == IF_OTHER_POPULATION_VAL:
        if not population_other:
            self.error_row(
                message="The value of field {} "
                        "is empty.".format(IF_OTHER_POPULATION)
            )
            # return
        pop_other = population_other

    year = self.row_value(row, YEAR)
    if year.isdigit():
        if int(year) > timezone.now().year:
            self.error_row(
                message=f"'{YEAR}' with value {year} exceeds current year."
            )
    count_total = self.row_value(row, COUNT_TOTAL)
    presence = self.presence(row)
    pop_certainty = self.row_value(row, POPULATION_ESTIMATE_CERTAINTY)
    sampling_effort_coverage = self.sampling_effort(row)

    if property and taxon and self.check_if_not_superuser():
        is_organisation_manager = (
            OrganisationRepresentative.objects.filter(
                organisation=property.organisation,
                user=self.upload_session.uploader
            )
        )
        existing_data = AnnualPopulation.objects.filter(
            year=int(string_to_number(year)),
            taxon=taxon,
            property=property
        ).first()
        if existing_data:
            # validate if user can update the data: uploader or manager
            if (
                not existing_data.is_editable(
                    self.upload_session.uploader)
            ):
                self.error_row(
                    message="You are not allowed to update data of "
                            "property {} and species {} in year {}".format(
                                property_code, scientific_name, year
                            )
                )
        else:
            # validate if user can add data to the property
            is_organisation_member = OrganisationUser.objects.filter(
                organisation=property.organisation,
                user=self.upload_session.uploader
            )
            if not is_organisation_manager and not is_organisation_member:
                self.error_row(
                    message="You are not allowed to add data to "
                            "property {}".format(property_code)
                )

    if len(self.row_error) > 0:
        return

    # Save AnnualPopulation
    try:
        annual, annual_created = AnnualPopulation.objects.update_or_create(
            year=int(string_to_number(year)),
            taxon=taxon,
            property=property,
            defaults={
                'user': self.upload_session.uploader,
                'area_available_to_species': area_available_to_species_num,
                'total': int(string_to_number(count_total)),
                'adult_total': int(string_to_number(
                    self.row_value(row, COUNT_ADULT_TOTAL))),
                'adult_male': int(string_to_number(
                    self.row_value(row, COUNT_ADULT_MALES))),
                'adult_female': int(string_to_number(
                    self.row_value(row, COUNT_ADULT_FEMALES))),
                'juvenile_male': int(string_to_number(
                    self.row_value(row, COUNT_JUVENILE_MALES))),
                'juvenile_female': int(string_to_number(
                    self.row_value(row, COUNT_JUVENILE_FEMALES))),
                'sub_adult_total': int(string_to_number(
                    self.row_value(row, COUNT_SUBADULT_TOTAL))),
                'sub_adult_male': int(string_to_number(
                    self.row_value(row, COUNT_SUBADULT_MALE))),
                'sub_adult_female': int(string_to_number(
                    self.row_value(row, COUNT_SUBADULT_FEMALE))),
                'juvenile_total': int(string_to_number(
                    self.row_value(row, COUNT_JUVENILE_TOTAL))),
                'group': int(string_to_number(self.row_value(row, GROUP))),
                'open_close_system': open_close_system,
                'survey_method': survey_method,
                'presence': presence,
                'upper_confidence_level': float(string_to_number(
                    self.row_value(row, UPPER))),
                'lower_confidence_level': float(string_to_number(
                    self.row_value(row, LOWER))),
                'certainty_of_bounds': int(string_to_number(
                    self.row_value(row, CERTAINTY_OF_POPULATION))),
                'sampling_effort_coverage': sampling_effort_coverage,
                'population_estimate_certainty': int(
                    string_to_number(pop_certainty)),
                'population_estimate_category': population_estimate,
                'survey_method_other': sur_other,
                'population_estimate_category_other': pop_other,
                'population_status': population_status
            }
        )
        annual.clean()
    except (IntegrityError, ValidationError):
        if annual.pk:
            annual.delete()
        self.error_row(
            message="The total of {} and {} must not exceed {}.".format(
                    COUNT_ADULT_MALES,
                    COUNT_ADULT_FEMALES,
                    COUNT_TOTAL)
        )
        logger.log(
            level=logging.ERROR,
            msg=str(self.row_error)
        )
        return

    if annual_created:
        self.created_list += 1
    else:
        self.existed_list += 1
        # if updated, cleared population per activity
        AnnualPopulationPerActivity.objects.filter(
            annual_population=annual
        ).delete()

    # Save AnnualPopulationPerActivity translocation intake
    if self.row_value(row, INTRODUCTION_TOTAL):
        intake = self.save_population_per_activity(
            row, ACTIVITY_TRANSLOCATION_INTAKE, year,
            annual, INTRODUCTION_TOTAL,
            INTRODUCTION_TOTAL_MALES, INTRODUCTION_TOTAL_FEMALES,
            INTRODUCTION_MALE_JUV, INTRODUCTION_FEMALE_JUV
        )
        intake_data = {
            "reintroduction_source": self.row_value(row,
                                                    INTRODUCTION_SOURCE),
            "founder_population": string_to_boolean(
                self.row_value(row, FOUNDER_POPULATION)
            ),
            "intake_permit": self.row_value(
                row, INTRODUCTION_PERMIT_NUMBER
            )
        }
        if intake:
            intake = AnnualPopulationPerActivity.objects.filter(
                id=intake.id
            )
            intake.update(**intake_data)

    # Save AnnualPopulationPerActivity translocation offtake
    if self.row_value(row, TRANS_OFFTAKE_TOTAL):
        off_take = self.save_population_per_activity(
            row, ACTIVITY_TRANSLOCATION_OFFTAKE, year,
            annual, TRANS_OFFTAKE_TOTAL,
            TRANS_OFFTAKE_ADULTE_MALES, TRANS_OFFTAKE_ADULTE_FEMALES,
            TRANS_OFFTAKE_MALE_JUV, TRANS_OFFTAKE_FEMALE_JUV
        )
        off_take_data = {
            "translocation_destination": self.row_value(
                row, TRANS_DESTINATION),
            "offtake_permit": self.row_value(
                row, TRANS_OFFTAKE_PERMIT_NUMBER)
        }
        if off_take:
            off_take = AnnualPopulationPerActivity.objects.filter(
                id=off_take.id
            )
            off_take.update(**off_take_data)

    # Save AnnualPopulationPerActivity Planned hunt/cull
    if self.row_value(row, PLANNED_HUNT_TOTAL):
        hunt = self.save_population_per_activity(
            row, ACTIVITY_PLANNED_HUNT_CULL, year,
            annual, PLANNED_HUNT_TOTAL,
            PLANNED_HUNT_OFFTAKE_ADULT_MALES,
            PLANNED_HUNT_OFFTAKE_ADULT_FAMALES,
            PLANNED_HUNT_OFFTAKE_MALE_JUV,
            PLANNED_HUNT_OFFTAKE_FEMALE_JUV
        )
        hunt_data = {
            "offtake_permit": self.row_value(
                row, PLANNED_HUNT_PERMIT_NUMBER
            )

        }
        if hunt:
            hunt = AnnualPopulationPerActivity.objects.filter(
                id=hunt.id
            )
            hunt.update(**hunt_data)

    # Save AnnualPopulationPerActivity Planned euthanasia
    if self.row_value(row, PLANNED_EUTH_TOTAL):
        planned = self.save_population_per_activity(
            row, ACTIVITY_PLANNED_EUTH_DCA, year,
            annual, PLANNED_EUTH_TOTAL,
            PLANNED_EUTH_OFFTAKE_ADULT_MALES,
            PLANNED_EUTH_OFFTAKE_ADULT_FAMALES,
            PLANNED_EUTH_OFFTAKE_MALE_JUV,
            PLANNED_EUTH_OFFTAKE_FEMALE_JUV
        )
        planned_data = {
            "offtake_permit": self.row_value(
                row, PLANNED_EUTH_PERMIT_NUMBER
            )

        }
        if planned:
            planned = AnnualPopulationPerActivity.objects.filter(
                id=planned.id
            )
            planned.update(**planned_data)

    # Save AnnualPopulationPerActivity Unplanned/illegal hunting
    if self.row_value(row, UNPLANNED_HUNT_TOTAL):
        self.save_population_per_activity(
            row, ACTIVITY_UNPLANNED_ILLEGAL_HUNTING, year,
            annual, UNPLANNED_HUNT_TOTAL,
            UNPLANNED_HUNT_OFFTAKE_ADULT_MALES,
            UNPLANNED_HUNT_OFFTAKE_ADULT_FAMALES,
            UNPLANNED_HUNT_OFFTAKE_MALE_JUV,
            UNPLANNED_HUNT_OFFTAKE_FEMALE_JUV
        )

row_value

row_value(row, key)

Get row value by key

Parameters:

Name	Type	Description	Default
row		row data	required
key		key	required

Returns:

Type	Description
	row value

Source code in django_project/species/scripts/data_upload.py

def row_value(self, row, key):
    """
    Get row value by key
    :param row: row data
    :param key: key
    :return: row value
    """
    row_value = ''
    try:
        row_value = row[key]
        row_value = row_value.replace('\xa0', ' ')
        row_value = row_value.replace('\xc2', '')
        row_value = row_value.replace('\\xa0', '')
        row_value = row_value.strip()
        row_value = re.sub(' +', ' ', row_value)
    except KeyError:
        pass
    return row_value

start

start(encoding='ISO-8859-1')

Start processing the csv file from upload session

Source code in django_project/species/scripts/data_upload.py

def start(self, encoding='ISO-8859-1'):
    """
    Start processing the csv file from upload session
    """
    self.error_list = []
    self.species_id_list = []
    uploaded_file = self.upload_session.process_file
    if self.upload_session.process_file.path.endswith('.xlsx'):
        excel = pd.ExcelFile(self.upload_session.process_file)
        dataframe = excel.parse(SHEET_TITLE)
        with tempfile.NamedTemporaryFile(mode='w', delete=False) \
                as csv_file:
            dataframe.to_csv(csv_file.name, index=False)
            uploaded_file = csv_file
    try:
        read_line = uploaded_file.readlines()
        uploaded_file_path = uploaded_file.path
    except ValueError:
        file = open(uploaded_file.name)
        read_line = file.readlines()
        uploaded_file_path = uploaded_file.name
    self.total_rows = len(
        read_line
    ) - 1
    self.process_started()
    processed = False

    with open(
            uploaded_file_path,
            encoding=encoding) as csv_file:
        try:
            self.csv_dict_reader = csv.DictReader(csv_file)
            self.process_csv_dict_reader()
            processed = True
        except UnicodeDecodeError:
            pass
    if not processed:
        with open(
                uploaded_file_path,
                encoding=encoding
        ) as csv_file:
            try:
                self.csv_dict_reader = csv.DictReader(csv_file)
                self.process_csv_dict_reader()
                processed = True
            except UnicodeDecodeError:
                pass
    if not processed:
        self.upload_session.canceled = True
        self.upload_session.save()
        self.process_ended()
        return
    self.process_ended()

survey_method

survey_method(row)

Get survey method.

Source code in django_project/species/scripts/data_upload.py

def survey_method(self, row):
    """Get survey method."""
    survey = self.row_value(row, SURVEY_METHOD)
    if survey:
        try:
            survey_method = SurveyMethod.objects.get(
                name__iexact=survey
            )
        except SurveyMethod.DoesNotExist:
            self.error_row(
                f"Survey method '{survey}' does not exist"
            )
            return None
        return survey_method
    return None

map_string_to_value

map_string_to_value(string, value_mapping)

Convert a string to the value in dictionary of value_mapping.

Source code in django_project/species/scripts/data_upload.py

def map_string_to_value(string, value_mapping):
    """Convert a string to the value in dictionary of value_mapping."""
    if string in value_mapping:
        return value_mapping[string]
    return None

string_to_boolean

string_to_boolean(string)

Convert a string to boolean.

Parameters:

Name	Type	Description	Default
string	str	The string to convert	required

Source code in django_project/species/scripts/data_upload.py

def string_to_boolean(string):
    """Convert a string to boolean.

    :param
    string: The string to convert
    :type
    string:str
    """
    if string in ['Yes', 'YES', 'yes']:
        return True
    return False

string_to_number

string_to_number(string)

Convert a string to a number.

Parameters:

Name	Type	Description	Default
string	str	The string to convert	required

Source code in django_project/species/scripts/data_upload.py

def string_to_number(string):
    """Convert a string to a number.

    :param
    string: The string to convert
    :type
    string:str
    """
    try:
        return float(string)
    except ValueError:
        return float(0)

Test Case

TaxonRankTestCase

Bases: TestCase

Taxon rank test case.

setUpTestData classmethod

setUpTestData()

Set up test data for taxon rank test case.

Source code in django_project/species/test_species_models.py

@classmethod
def setUpTestData(cls):
    """Set up test data for taxon rank test case."""
    cls.taxonRank = TaxonRankFactory()

test_create_taxon_rank

test_create_taxon_rank()

Test create taxon rank.

Source code in django_project/species/test_species_models.py

def test_create_taxon_rank(self):
    """Test create taxon rank."""
    self.assertTrue(isinstance(self.taxonRank, TaxonRank))
    self.assertEqual(
        self.taxonRank.name,
        TaxonRank.objects.get(id=self.taxonRank.id).name
    )

test_delete_taxon_rank

test_delete_taxon_rank()

Test delete taxon rank.

Source code in django_project/species/test_species_models.py

def test_delete_taxon_rank(self):
    """Test delete taxon rank."""
    self.taxonRank.delete()
    self.assertEqual(TaxonRank.objects.count(), 0)

test_unique_taxon_rank_name_constraint

test_unique_taxon_rank_name_constraint()

Test unique taxon rank name constraint.

Source code in django_project/species/test_species_models.py

def test_unique_taxon_rank_name_constraint(self):
    """Test unique taxon rank name constraint."""
    with self.assertRaises(Exception) as raised:
        TaxonRankFactory(name='taxon_rank_1')
        self.assertEqual(IntegrityError, type(raised.exception))

test_update_taxon_rank

test_update_taxon_rank()

Test update taxon rank.

Source code in django_project/species/test_species_models.py

def test_update_taxon_rank(self):
    """Test update taxon rank."""
    self.taxonRank.name = 'taxon_rank_1'
    self.taxonRank.save()
    self.assertEqual(
        TaxonRank.objects.get(id=self.taxonRank.id).name,
        'taxon_rank_1'
    )

TaxonSurveyMethodTestCase

Bases: TestCase

Taxon survey method count test case.

setUpTestData classmethod

setUpTestData()

SetUpTestData for Taxon survey method count test case.

Source code in django_project/species/test_species_models.py

@classmethod
def setUpTestData(cls):
    """SetUpTestData for Taxon survey method count test case."""
    cls.taxon = Taxon.objects.create(
        scientific_name='taxon_0',
        common_name_verbatim='taxon_0',
        colour_variant=False,
        taxon_rank=TaxonRankFactory(),
    )
    cls.survey_method = SurveyMethod.objects.create(
        name='Unknown',
    )
    cls.taxon_survey_method = TaxonSurveyMethodF(
        taxon=cls.taxon,
        survey_method=cls.survey_method
    )

test_create_taxon_survey_method

test_create_taxon_survey_method()

Test create Taxon survey method count.

Source code in django_project/species/test_species_models.py

def test_create_taxon_survey_method(self):
    """Test create Taxon survey method count."""
    self.assertTrue(
        isinstance(self.taxon_survey_method, TaxonSurveyMethod)
    )
    self.assertEqual(TaxonSurveyMethod.objects.count(), 1)
    self.assertEqual(
        TaxonSurveyMethod.objects.filter(
        taxon__scientific_name=self.taxon.scientific_name
        ).count(), 1
    )
    self.assertEqual(
        TaxonSurveyMethod.objects.filter(
        survey_method__name=self.survey_method.name
        ).count(), 1
    )

test_delete_taxon_survey_method

test_delete_taxon_survey_method()

Test delete Taxon survey method count.

Source code in django_project/species/test_species_models.py

def test_delete_taxon_survey_method(self):
    """Test delete Taxon survey method count."""
    self.taxon.delete()
    self.assertEqual(TaxonSurveyMethod.objects.count(), 0)

test_update_taxon_survey_method

test_update_taxon_survey_method()

Test update Taxon survey method count.

Source code in django_project/species/test_species_models.py

def test_update_taxon_survey_method(self):
    """Test update Taxon survey method count."""
    taxon = TaxonFactory.create(
        scientific_name='taxon',
        common_name_verbatim='taxon_0',
        colour_variant=False,
        taxon_rank=TaxonRankFactory(),
    )
    self.taxon_survey_method.taxon = taxon
    self.taxon_survey_method.save()
    self.assertEqual(
        TaxonSurveyMethod.objects.filter(
        taxon__scientific_name='taxon').count(),
        1
    )

TaxonTestCase

Bases: TestCase

Taxon model test case.

setUpTestData classmethod

setUpTestData()

Taxon model test data.

Source code in django_project/species/test_species_models.py

@classmethod
def setUpTestData(cls):
    """Taxon model test data."""
    cls.taxonRank = TaxonRankFactory.create(
        name='Species'
    )
    cls.taxon = TaxonFactory.create(
        scientific_name='taxon_0',
        common_name_verbatim='taxon_0',
        colour_variant=False,
        taxon_rank=cls.taxonRank,
        show_on_front_page=False
    )
    cls.url = reverse('species')

test_create_taxon_no_graph_icon

test_create_taxon_no_graph_icon()

Test create taxon without graph icon.

Source code in django_project/species/test_species_models.py

def test_create_taxon_no_graph_icon(self):
    """Test create taxon without graph icon."""
    taxon = TaxonFactory.create(
        scientific_name='taxon_1',
        common_name_verbatim='taxon_11',
        colour_variant=False,
        taxon_rank=self.taxonRank,
        show_on_front_page=False,
        icon=None
    )
    self.assertTrue(isinstance(taxon, Taxon))
    self.assertEqual(Taxon.objects.count(), 2)
    self.assertEqual(
        taxon.scientific_name,
        'taxon_1'
    )
    self.assertEqual(taxon.graph_icon, None)
    self.assertEqual(taxon.topper_icon, None)
    self.assertEqual(taxon.icon, None)

test_delete_taxon

test_delete_taxon()

Test delete taxon.

Source code in django_project/species/test_species_models.py

def test_delete_taxon(self):
    """Test delete taxon."""
    self.taxon.delete()
    self.assertEqual(Taxon.objects.count(), 0)
    """Test delete taxon rank."""
    self.taxonRank.delete()
    self.assertEqual(TaxonRank.objects.count(), 0)

test_get_taxon_frontpage_list

test_get_taxon_frontpage_list()

Test fetch taxon list for frontpage.

Source code in django_project/species/test_species_models.py

def test_get_taxon_frontpage_list(self):
    """Test fetch taxon list for frontpage."""
    taxon = TaxonFactory.create(
        scientific_name='taxon_1',
        common_name_verbatim='taxon_1',
        colour_variant=False,
        taxon_rank=self.taxonRank,
        show_on_front_page=True
    )
    property_1 = PropertyFactory.create()
    property_2 = PropertyFactory.create()
    client = Client()
    response = client.get(reverse('species-front-page'))
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertEqual(len(response.data), 1)
    taxon_1 = [d for d in response.data if d['id'] == taxon.id]
    self.assertTrue(taxon_1)
    self.assertEqual(taxon_1[0]['total_population'], 0)
    self.assertEqual(taxon_1[0]['species_name'], taxon.scientific_name)
    user_1 = User.objects.create_user(username='testuser_taxon_1', password='12345')
    user_2 = User.objects.create_user(username='testuser_taxon_2', password='12345')

    # create two years of data
    AnnualPopulationF(
        year=2021, total=30,
        adult_male=10, adult_female=10,
        taxon=taxon,
        user=user_1,
        property=property_1,
        area_available_to_species=2
    )
    AnnualPopulationF(
        year=2022, total=35,
        adult_male=10, adult_female=10,
        taxon=taxon,
        user=user_1,
        property=property_1,
        area_available_to_species=2
    )
    AnnualPopulationF(
        year=2020, total=15,
        adult_male=10, adult_female=5,
        taxon=taxon,
        user=user_2,
        property=property_2,
        area_available_to_species=1
    )
    AnnualPopulationF(
        year=2022, total=22,
        adult_male=10, adult_female=10,
        taxon=taxon,
        user=user_2,
        property=property_2,
        area_available_to_species=1
    )
    response = client.get(reverse('species-front-page'))
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertEqual(len(response.data), 1)
    taxon_1 = [d for d in response.data if d['id'] == taxon.id]
    self.assertTrue(taxon_1)
    self.assertEqual(taxon_1[0]['total_population'], 57)
    self.assertEqual(taxon_1[0]['total_area'], 3)

test_get_taxon_list

test_get_taxon_list()

Taxon list API test within the organisation

Source code in django_project/species/test_species_models.py

def test_get_taxon_list(self):
    """Taxon list API test within the organisation"""
    organisation = organisationFactory.create()
    property_obj = PropertyFactory.create(organisation=organisation)
    AnnualPopulationF.create(
        taxon=self.taxon,
        total=10,
        adult_male=5,
        adult_female=5,
        property=property_obj
    )

    user = User.objects.create_user(
        username='testuserd',
        password='testpasswordd'
    )

    user.user_profile.current_organisation = organisation
    user.save()

    auth_headers = {
        'HTTP_AUTHORIZATION': 'Basic ' +
        base64.b64encode(b'testuserd:testpasswordd').decode('ascii'),
    }
    client = Client()
    response = client.get(self.url, **auth_headers)
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    expected_data = TaxonSerializer([self.taxon], many=True).data
    self.assertEqual(expected_data, response.data)

test_get_taxon_list_empty

test_get_taxon_list_empty()

Taxon list API test when no Annual Population has been made

Source code in django_project/species/test_species_models.py

def test_get_taxon_list_empty(self):
    """Taxon list API test when no Annual Population has been made"""
    organisation = organisationFactory.create()

    user = User.objects.create_user(
        username='testuserd',
        password='testpasswordd'
    )

    user.user_profile.current_organisation = organisation
    user.save()

    auth_headers = {
        'HTTP_AUTHORIZATION': 'Basic ' +
        base64.b64encode(b'testuserd:testpasswordd').decode('ascii'),
    }
    client = Client()
    response = client.get(self.url, **auth_headers)
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertEqual(response.json(), [])

test_get_taxon_list_for_organisations

test_get_taxon_list_for_organisations()

Taxon list API test for organisations.

Source code in django_project/species/test_species_models.py

def test_get_taxon_list_for_organisations(self):
    """Taxon list API test for organisations."""
    organisation = organisationFactory.create(national=True)
    property_obj = PropertyFactory.create(organisation=organisation)
    AnnualPopulationF.create(
        taxon=self.taxon,
        total=10,
        adult_male=5,
        adult_female=5,
        property=property_obj
    )

    user = User.objects.create_user(
        username='testuserd',
        password='testpasswordd'
    )

    user.user_profile.current_organisation = organisation
    user.save()

    property = PropertyFactory.create(
        organisation=organisation,
        name='PropertyA'
    )

    auth_headers = {
        'HTTP_AUTHORIZATION': 'Basic ' +
        base64.b64encode(b'testuserd:testpasswordd').decode('ascii'),
    }
    client = Client()
    data = {"organisation": organisation.id}
    response = client.get(self.url, data, **auth_headers)
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertEqual(len(response.data), 1)
    self.assertEqual(response.data[0]['scientific_name'], "taxon_0")

test_get_taxon_provincial_data_consumer

test_get_taxon_provincial_data_consumer()

Taxon list API test when user is provincial data consumer

Source code in django_project/species/test_species_models.py

def test_get_taxon_provincial_data_consumer(self):
    """Taxon list API test when user is provincial data consumer"""
    self._setup_data_provincial_data_consumer()
    auth_headers = {
        'HTTP_AUTHORIZATION': 'Basic ' +
        base64.b64encode(b'testuserd:testpasswordd').decode('ascii'),
    }
    client = Client()
    response = client.get(self.url, **auth_headers)
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertEqual(
        response.json(),
        [
            {
                'common_name_verbatim': self.taxon.common_name_verbatim,
                'id': self.taxon.id,
                'scientific_name': self.taxon.scientific_name
            }
        ]
    )

test_get_taxon_provincial_data_consumer_no_organisation

test_get_taxon_provincial_data_consumer_no_organisation()

Taxon list API test when user is provincial data consumer and has no active organisation.

Source code in django_project/species/test_species_models.py

def test_get_taxon_provincial_data_consumer_no_organisation(self):
    """Taxon list API test when user is provincial data consumer
    and has no active organisation."""
    self._setup_data_provincial_data_consumer(set_organisation=False)
    auth_headers = {
        'HTTP_AUTHORIZATION': 'Basic ' +
        base64.b64encode(b'testuserd:testpasswordd').decode('ascii'),
    }
    client = Client()
    response = client.get(self.url, **auth_headers)
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertEqual(response.json(), [])

test_get_taxon_trend_page

test_get_taxon_trend_page()

Test fetch taxon detil for trend page.

Source code in django_project/species/test_species_models.py

def test_get_taxon_trend_page(self):
    """Test fetch taxon detil for trend page."""
    taxon = TaxonFactory.create(
        scientific_name='taxon_1',
        common_name_verbatim='taxon_1',
        colour_variant=False,
        taxon_rank=self.taxonRank,
        show_on_front_page=True
    )
    property_1 = PropertyFactory.create()
    property_2 = PropertyFactory.create()
    client = Client()
    response = client.get(
        reverse('taxon-trend-page'),
        {
            'species': taxon.scientific_name
        }
    )
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertEqual(response.json()['total_population'], 0)
    self.assertEqual(response.json()['species_name'], taxon.scientific_name)
    self.assertIsNone(response.json()['graph_icon'])
    self.assertIsNone(response.json()['model_updated_on'])
    user_1 = User.objects.create_user(username='testuser_taxon_1', password='12345')
    user_2 = User.objects.create_user(username='testuser_taxon_2', password='12345')
    # create two years of data
    AnnualPopulationF(
        year=2021, total=30,
        adult_male=10, adult_female=10,
        taxon=taxon,
        user=user_1,
        property=property_1,
        area_available_to_species=2
    )
    AnnualPopulationF(
        year=2022, total=35,
        adult_male=10, adult_female=10,
        taxon=taxon,
        user=user_1,
        property=property_1,
        area_available_to_species=2
    )
    AnnualPopulationF(
        year=2020, total=15,
        adult_male=10, adult_female=5,
        taxon=taxon,
        user=user_2,
        property=property_2,
        area_available_to_species=1
    )
    AnnualPopulationF(
        year=2022, total=22,
        adult_male=10, adult_female=10,
        taxon=taxon,
        user=user_2,
        property=property_2,
        area_available_to_species=1
    )
    # create statistical model output
    SpeciesModelOutputF.create(
        taxon=taxon,
        is_latest=True,
        status=DONE,
        generated_on=datetime.datetime(2000, 8, 14, 8, 8, 8)
    )
    response = client.get(
        reverse('taxon-trend-page'),
        {
            'species': taxon.scientific_name
        }
    )
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertTrue(response.json())
    self.assertEqual(response.json()['total_population'], 57)
    self.assertEqual(response.json()['total_area'], 3)
    self.assertIsNone(response.json()['graph_icon'])
    self.assertIsNotNone(response.json()['model_updated_on'])

test_taxon_relation_to_self

test_taxon_relation_to_self()

Test taxon relation to self.

Source code in django_project/species/test_species_models.py

def test_taxon_relation_to_self(self):
    """Test taxon relation to self."""
    self.taxon2 = Taxon.objects.create(
        scientific_name='taxon_1',
        common_name_verbatim='taxon_1',
        colour_variant=False,
        taxon_rank=self.taxonRank,
        parent=self.taxon,
    )
    self.assertEqual(self.taxon2.parent, self.taxon)

test_taxon_unique_infraspecific_epithet_constraint

test_taxon_unique_infraspecific_epithet_constraint()

Test taxon unique infraspecific epithet constraint.

Source code in django_project/species/test_species_models.py

def test_taxon_unique_infraspecific_epithet_constraint(self):
    """Test taxon unique infraspecific epithet constraint."""

    Taxon.objects.create(
        scientific_name='taxon_2',
        common_name_verbatim='taxon_2',
        colour_variant=False,
        infraspecific_epithet='infra_2',
        taxon_rank=self.taxonRank,
    )
    self.assertEqual(
        Taxon.objects.filter(infraspecific_epithet='infra_2').count(),
        1
    )

    with self.assertRaises(Exception) as raised:
        Taxon.objects.create(
            scientific_name='taxon_0',
            common_name_verbatim='taxon_0',
            colour_variant=False,
            infraspecific_epithet='infra_1',
            taxon_rank=self.taxonRank,
        )

test_taxon_unique_scientific_name_constraint

test_taxon_unique_scientific_name_constraint()

Test taxon unique scientific name constraint.

Source code in django_project/species/test_species_models.py

def test_taxon_unique_scientific_name_constraint(self):
    """Test taxon unique scientific name constraint."""
    with self.assertRaises(Exception) as raised:
        Taxon.objects.create(
            scientific_name='taxon_1',
            common_name_verbatim='taxon_0',
            colour_variant=False,
            taxon_rank=self.taxonRank,
        )
        self.assertEqual(IntegrityError, type(raised.exception))

test_update_taxon

test_update_taxon()

Test update taxon objects.

Source code in django_project/species/test_species_models.py

def test_update_taxon(self):
    """Test update taxon objects."""
    graph_icon_path = absolute_path(
        'frontend', 'static', 'images', 'Loxodonta_africana-graph.svg'
    )

    with open(graph_icon_path, 'rb') as f:
        self.taxon.scientific_name = 'taxon_1'
        self.taxon.infraspecific_epithet = 'infra_1'
        self.taxon.graph_icon = ContentFile(f.read(), name=f"file.svg")
        self.taxon.save()
        self.taxon.refresh_from_db()
        self.assertEqual(
            self.taxon.scientific_name,
            'taxon_1'
        )
        self.assertEqual(
            self.taxon.infraspecific_epithet,
            'infra_1'
        )

        # Check graph_icon, icon, and topper_icon are updated.
        # Icon and topper_icon are generated automatically from graph_icon
        self.assertTrue(
            os.path.exists(
                absolute_path(
                    settings.MEDIA_ROOT,
                    str(self.taxon.graph_icon)
                )
            )
        )
        self.assertTrue(
            os.path.exists(
                absolute_path(
                    settings.MEDIA_ROOT,
                    str(self.taxon.icon)
                )
            )
        )
        self.assertTrue(
            os.path.exists(
                absolute_path(
                    settings.MEDIA_ROOT,
                    str(self.taxon.topper_icon)
                )
            )
        )

        # Check fill color for each icon.
        # Icon and topper_icon color are generated automatically from graph_icon
        self.assertTrue(
            self.taxon.graph_icon.readlines()[2].endswith(b'10.052 2.6849z" fill="#000000"/>\n')
        )
        self.assertTrue(
            self.taxon.topper_icon.readlines()[2].endswith(b'10.052 2.6849z" fill="#75B37A"/>\n')
        )
        self.assertTrue(
            self.taxon.icon.readlines()[2].endswith(b'10.052 2.6849z" fill="#FFFFFF"/>\n')
        )

TestTaxonSerializer

Bases: TestCase

Test Taxon Serializer

TestUploadSpeciesApiView

Bases: TestCase

Test api view species uploader

test_overwrite_annual_population

test_overwrite_annual_population()

Test upload species multiple times to overwrite data.

Source code in django_project/species/test_api_views.py

def test_overwrite_annual_population(self):
    """Test upload species multiple times to overwrite data."""
    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test_first_upload.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')

    kwargs = {
        'token': self.token
    }
    request = self.factory.get(
        reverse('upload-species-status', kwargs=kwargs)
    )
    request.user = self.user
    view = UploadSpeciesStatus.as_view()
    response = view(request, **kwargs)
    self.assertEqual(response.status_code, 200)
    self.assertEqual(response.data['status'], 'Finished')
    self.assertEqual(AnnualPopulation.objects.count(), 3)
    self.assertEqual(upload_session.success_notes, "3 rows uploaded successfully.")
    annual_2010 = AnnualPopulation.objects.filter(
        property=self.property,
        taxon=self.lion,
        year=2010
    ).first()
    self.assertTrue(annual_2010)
    self.assertEqual(annual_2010.total, 190)
    self.assertEqual(AnnualPopulationPerActivity.objects.filter(
        annual_population=annual_2010
    ).count(), 5)
    annual_2011 = AnnualPopulation.objects.filter(
        property=self.property,
        taxon=self.lion,
        year=2011
    ).first()
    self.assertTrue(annual_2011)
    self.assertEqual(annual_2011.total, 190)
    self.assertEqual(AnnualPopulationPerActivity.objects.filter(
        annual_population=annual_2011
    ).count(), 5)
    annual_2012 = AnnualPopulation.objects.filter(
        property=self.property,
        taxon=self.lion,
        year=2012
    ).first()
    self.assertTrue(annual_2012)
    self.assertEqual(annual_2012.total, 190)
    self.assertEqual(AnnualPopulationPerActivity.objects.filter(
        annual_population=annual_2012
    ).count(), 4)
    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test_second_upload.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')

    kwargs = {
        'token': self.token
    }
    request = self.factory.get(
        reverse('upload-species-status', kwargs=kwargs)
    )
    request.user = self.user
    view = UploadSpeciesStatus.as_view()
    response = view(request, **kwargs)
    self.assertEqual(response.status_code, 200)
    self.assertEqual(response.data['status'], 'Finished')
    self.assertEqual(AnnualPopulation.objects.count(), 3)
    self.assertEqual(upload_session.success_notes,
                     "2 rows already exist and have been overwritten.")
    annual_2010.refresh_from_db()
    annual_2011.refresh_from_db()
    annual_2012.refresh_from_db()
    # ensure no change for 2010
    self.assertEqual(annual_2010.total, 190)
    self.assertEqual(AnnualPopulationPerActivity.objects.filter(
        annual_population=annual_2010
    ).count(), 5)
    self.assertEqual(annual_2011.total, 240)
    self.assertEqual(AnnualPopulationPerActivity.objects.filter(
        annual_population=annual_2011
    ).count(), 0)
    self.assertEqual(annual_2012.total, 160)
    self.assertEqual(AnnualPopulationPerActivity.objects.filter(
        annual_population=annual_2012
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        annual_population=annual_2012,
        activity_type__name="Translocation (Offtake)",
        total=20
    ).exists())

test_save_csv_with_no_property

test_save_csv_with_no_property()

Test save csv with not existing property.

Source code in django_project/species/test_api_views.py

def test_save_csv_with_no_property(self):
    """Test save csv with not existing property."""

    request = self.factory.post(
        reverse('save-csv-species'),
        data={
            'token': self.token,
            'property': 0
        }, format='json')

    request.user = self.user
    view = SaveCsvSpecies.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 400)

test_task_string_to_boolean

test_task_string_to_boolean()

Test string_to_boolean functionality in task

Source code in django_project/species/test_api_views.py

def test_task_string_to_boolean(self):
    """Test string_to_boolean functionality in task"""

    self.assertTrue(string_to_boolean('yes'))
    self.assertFalse(string_to_boolean(''))

test_task_string_to_number

test_task_string_to_number()

Test string_to_number functionality in task

Source code in django_project/species/test_api_views.py

def test_task_string_to_number(self):
    """Test string_to_number functionality in task"""

    self.assertEqual(10, string_to_number('10'))
    self.assertEqual(0.0, string_to_number(''))

test_upload_csv_task

test_upload_csv_task(mock_app)

Test upload csv task.

Source code in django_project/species/test_api_views.py

@mock.patch("species.tasks.upload_species.upload_species_data")
def test_upload_csv_task(self, mock_app):
    """Test upload csv task."""
    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)

    upload_species_data(upload_session.id)
    self.assertEqual(Taxon.objects.all().count(), 1)
    self.assertEqual(AnnualPopulationPerActivity.objects.all().count(), 5)
    self.assertEqual(AnnualPopulation.objects.all().count(), 1)
    population_data = AnnualPopulation.objects.all().first()
    self.assertEqual(population_data.total, 190)
    self.assertEqual(population_data.adult_total, 150)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        activity_type__name="Translocation (Offtake)"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        activity_type__name="Planned Hunt/Cull"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        activity_type__name="Translocation (Intake)"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        activity_type__name="Planned Euthanasia/DCA"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        activity_type__name="Unplanned/Illegal Hunting"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        translocation_destination="KNP", offtake_permit="ABC100X10"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        offtake_permit="DEF100X10"
    ).count(), 1)

    self.assertTrue(OpenCloseSystem.objects.all().count() == 3)

test_upload_excel_invalid_area_available

test_upload_excel_invalid_area_available()

Test upload species with a csv file that has invalid area available to species.

Source code in django_project/species/test_api_views.py

def test_upload_excel_invalid_area_available(self):
    """Test upload species with a csv file that has invalid
    area available to species."""

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test_invalid_area_available.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')
    upload_session.refresh_from_db()
    self.assertTrue('error' in upload_session.error_file.path)
    with open(upload_session.error_file.path, encoding='utf-8-sig') as csv_file:
        error_file = csv.DictReader(csv_file)
        headers = error_file.fieldnames
        self.assertTrue('error_message' in headers)
        errors = []
        for row in error_file:
            errors.append(row['error_message'])
        self.assertTrue(
            "Area available to species must be greater than 0 "
            "and less than or equal to property area size ({:.0f} ha).".format(
                self.property.property_size_ha) in errors)

test_upload_excel_missing_compulsory_field

test_upload_excel_missing_compulsory_field()

Test upload species with an excel file which misses a compulsory field.

Source code in django_project/species/test_api_views.py

def test_upload_excel_missing_compulsory_field(self):
    """Test upload species with an excel file which misses
     a compulsory field."""

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'excel_wrong_header.xlsx')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 400)
    self.assertEqual(UploadSpeciesCSV.objects.filter(token=self.token).count(),
                     1)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    self.assertTrue(upload_session.canceled)
    self.assertEqual(upload_session.process_file.name, '')
    self.assertEqual(upload_session.error_notes,
                     "The 'Property_code' field is missing. Please check "
                     "that all the compulsory fields are in the "
                     "CSV file headers."
                     )

test_upload_session

test_upload_session()

Test upload species

Source code in django_project/species/test_api_views.py

def test_upload_session(self):
    """Test upload species """

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    self.assertEqual(UploadSpeciesCSV.objects.filter(token=self.token).count(),
                     1)
    file_name = 'species'
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    self.assertTrue(file_name in upload_session.process_file.path)
    self.assertEqual(upload_session.error_file.name, '')

    with mock.patch('species.api_views.upload_species.upload_species_data.delay') as mock_task:
        request = self.factory.post(
            reverse('save-csv-species'),
            data={
                'token': self.token,
                'property': self.property.id
            }, format='json')

        request.user = self.user
        view = SaveCsvSpecies.as_view()
        response = view(request)
        self.assertEqual(response.status_code, 200)
        self.assertTrue(mock_task.called)

test_upload_session_incorrect

test_upload_session_incorrect()

Test upload species with incorrect header

Source code in django_project/species/test_api_views.py

def test_upload_session_incorrect(self):
    """Test upload species with incorrect header"""

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'incorrect.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 400)
    self.assertEqual(UploadSpeciesCSV.objects.filter(token=self.token).count(),
                     1)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    self.assertTrue(upload_session.canceled)
    self.assertEqual(upload_session.process_file.name, '')
    self.assertEqual(upload_session.error_notes,
                     "The 'Property_code' field is missing. "
                     "Please check that all the compulsory fields "
                     "are in the CSV file headers."
                     )

test_upload_session_no_sheet

test_upload_session_no_sheet()

Test upload species with no sheet in excel file.

Source code in django_project/species/test_api_views.py

def test_upload_session_no_sheet(self):
    """Test upload species with no sheet in excel file."""

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'excel_no_sheet.xlsx')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 400)
    self.assertEqual(UploadSpeciesCSV.objects.filter(token=self.token).count(),
                     1)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    self.assertTrue(upload_session.canceled)
    self.assertEqual(upload_session.process_file.name, '')
    self.assertEqual(upload_session.error_notes,
                     "The sheet named Dataset pilot is not in the Excel "
                     "file. Please download the template to get "
                     "the correct file."
                     )

test_upload_species_status

test_upload_species_status()

Test upload species status.

Source code in django_project/species/test_api_views.py

def test_upload_species_status(self):
    """Test upload species status."""
    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)

    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')

    kwargs = {
        'token': self.token
    }
    request = self.factory.get(
        reverse('upload-species-status', kwargs=kwargs)
    )
    request.user = self.user
    view = UploadSpeciesStatus.as_view()
    response = view(request, **kwargs)
    self.assertEqual(response.status_code, 200)
    self.assertEqual(response.data['status'], 'Finished')
    self.assertFalse(response.data['error_file'])

test_upload_species_status_404

test_upload_species_status_404()

Test upload species status with 404 error.

Source code in django_project/species/test_api_views.py

def test_upload_species_status_404(self):
    """Test upload species status with 404 error."""
    kwargs = {
        'token': '8f1c1181-982a-4286-b2fe-da1abe8f7172'
    }
    request = self.factory.get(
        reverse('upload-species-status', kwargs=kwargs)
    )
    request.user = self.user
    view = UploadSpeciesStatus.as_view()
    response = view(request, **kwargs)
    self.assertEqual(response.status_code, 404)

test_upload_species_status_empty_file

test_upload_species_status_empty_file()

Test upload with empty file.

Source code in django_project/species/test_api_views.py

def test_upload_species_status_empty_file(self):
    """Test upload with empty file."""
    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test_empty.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)

    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')

    kwargs = {
        'token': self.token
    }
    request = self.factory.get(
        reverse('upload-species-status', kwargs=kwargs)
    )
    request.user = self.user
    view = UploadSpeciesStatus.as_view()
    response = view(request, **kwargs)
    self.assertEqual(response.status_code, 200)
    self.assertEqual(response.data['status'], 'Error')
    self.assertFalse(response.data['error_file'])
    self.assertEqual(
        response.data['message'],
        'You have uploaded empty spreadsheet, please check again.'
    )
    upload_session.refresh_from_db()
    self.assertEqual(
        upload_session.error_notes,
        'You have uploaded empty spreadsheet, please check again.'
    )

test_upload_species_with_excel_property_not_exist

test_upload_species_with_excel_property_not_exist()

Test upload Excel file with a property not existing.

Source code in django_project/species/test_api_views.py

def test_upload_species_with_excel_property_not_exist(self):
    """Test upload Excel file with a property not existing."""

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'excel_error_property.xlsx')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')
    self.assertTrue('error' in upload_session.error_file.path)

    xl = pd.ExcelFile(upload_session.error_file.path)
    dataset = xl.parse(SHEET_TITLE)
    self.assertEqual(
        dataset.iloc[0]['error_message'],
        "Property code Venetia Limpopo doesn't match the selected "
        "property. Please replace it with {}. Loxodonta africana "
        "doesn't exist in the database. Please select species "
        "available in the dropdown only.".format(
            upload_session.property.short_code
        )
    )
    # create two properties with code Venetia Limpopo
    # should return same error
    property1 = PropertyFactory(
        name="Venetia Limpopo"
    )
    property2 = PropertyFactory(
        name="Venetia Limpopo 2"
    )
    Property.objects.filter(id__in=[property1.id, property2.id]).update(
        short_code='Venetia Limpopo'
    )
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')
    self.assertTrue('error' in upload_session.error_file.path)

    xl = pd.ExcelFile(upload_session.error_file.path)
    dataset = xl.parse(SHEET_TITLE)
    self.assertEqual(
        dataset.iloc[0]['error_message'],
        "Property code Venetia Limpopo doesn't match the selected "
        "property. Please replace it with {}. Loxodonta africana "
        "doesn't exist in the database. Please select species "
        "available in the dropdown only.".format(
            upload_session.property.short_code
        )
    )

test_upload_species_with_property_taxon_not_exist

test_upload_species_with_property_taxon_not_exist()

Test upload species task with a property and taxon not existing.

Source code in django_project/species/test_api_views.py

def test_upload_species_with_property_taxon_not_exist(self):
    """Test upload species task with a property and taxon not existing."""

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test_property_taxon.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')

    kwargs = {
        'token': self.token
    }
    request = self.factory.get(
        reverse('upload-species-status', kwargs=kwargs)
    )
    request.user = self.user
    view = UploadSpeciesStatus.as_view()
    response = view(request, **kwargs)
    self.assertEqual(response.status_code, 200)
    self.assertTrue('media' in response.data['error_file'])

    self.assertTrue('error' in upload_session.error_file.path)
    with open(upload_session.error_file.path, encoding='utf-8-sig') as csv_file:
        error_file = csv.DictReader(csv_file)
        headers = error_file.fieldnames
        self.assertTrue('error_message' in headers)
        errors = []
        for row in error_file:
            errors.append(row['error_message'])
        self.assertTrue("Property code Luna's Reserve doesn't match "
                        "the selected property. Please replace "
                        "it with Luna." in errors)
        self.assertTrue("Lemurs doesn't exist in the database. "
                        "Please select species available in the "
                        "dropdown only." in errors)
        self.assertTrue("The value of field "
                        "If_other_(population_estimate_category)_please "
                        "explain is empty." in errors)
        self.assertTrue("The value of field "
                        "If_other_(survey_method)_please "
                        "explain is empty." in errors)
        self.assertTrue("The value of the compulsory field "
                        "Population_estimate_category is empty." in errors)
        self.assertTrue("The value of the compulsory field "
                        "presence/absence is empty. The value "
                        "of the compulsory field "
                        "Population_estimate_category is empty." in errors)
        self.assertTrue("The total of Count_adult_males and "
                        "Count_adult_females must not exceed "
                        "COUNT_TOTAL." in errors)

        # TODO: Check why this test is failing
        # self.assertTrue("The total of "
        #                 "Planned hunt/culling_Offtake_adult_males and "
        #                 "Planned hunt/culling_Offtake_adult_females must "
        #                 "not exceed Planned hunt/culling_TOTAL." in errors)
    self.assertEqual(AnnualPopulation.objects.count(), 7)
    self.assertEqual(upload_session.success_notes, "7 rows uploaded successfully.")
    self.assertTrue(AnnualPopulation.objects.filter(
        survey_method_other="Test survey"
    ).count(), 1)
    self.assertTrue(AnnualPopulation.objects.filter(
        survey_method__name="Other - please explain",
        survey_method_other="Test survey"
    ).count(), 1)
    self.assertTrue(AnnualPopulation.objects.filter(
        population_estimate_category__name="Other (please describe how the "
                                           "population size estimate was "
                                           "determined)",
        population_estimate_category_other="Decennial census"
    ).count(), 1)
    self.assertTrue(AnnualPopulation.objects.filter(
        population_estimate_category__name="Ad hoc or "
                                           "opportunistic monitoring"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        intake_permit="12345"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        reintroduction_source="KNP"
    ).count(), 1)
    self.assertTrue(AnnualPopulationPerActivity.objects.filter(
        founder_population=True
    ).count(), 1)

test_upload_species_without_login

test_upload_species_without_login()

Test upload species api

Source code in django_project/species/test_api_views.py

def test_upload_species_without_login(self):
    """Test upload species api"""

    response = self.client.get(self.api_url)
    self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

test_upload_task_with_upload_session_not_existing

test_upload_task_with_upload_session_not_existing(mock_app)

Test upload task with upload session not existing.

Source code in django_project/species/test_api_views.py

@mock.patch("species.tasks.upload_species.upload_species_data")
def test_upload_task_with_upload_session_not_existing(self, mock_app):
    """Test upload task with upload session not existing."""

    self.assertEqual(upload_species_data(1), None)

test_uploader_view_with_empty_compulsory_fields

test_uploader_view_with_empty_compulsory_fields()

Test uploader file view with an empty value in compulsory field.

Source code in django_project/species/test_api_views.py

def test_uploader_view_with_empty_compulsory_fields(self):
    """Test uploader file view with an empty value in
    compulsory field."""

    csv_path = absolute_path(
        'frontend', 'tests',
        'csv', 'test_empty_value_of_compulsory_fields.csv')
    data = open(csv_path, 'rb')
    data = SimpleUploadedFile(
        content=data.read(),
        name=data.name,
        content_type='multipart/form-data'
    )

    request = self.factory.post(
        reverse('upload-species'), {
            'file': data,
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 204)
    upload_session = UploadSpeciesCSV.objects.get(token=self.token)
    upload_session.progress = 'Processing'
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start('utf-8-sig')
    self.assertTrue('error' in upload_session.error_file.path)

test_uploader_view_without_file

test_uploader_view_without_file()

Test uploader file view with no file

Source code in django_project/species/test_api_views.py

def test_uploader_view_without_file(self):
    """Test uploader file view with no file"""

    request = self.factory.post(
        reverse('upload-species'), {
            'token': self.token,
            'property': self.property.id
        }
    )
    request.user = self.user
    view = SpeciesUploader.as_view()
    response = view(request)
    self.assertEqual(response.status_code, 400)
    self.assertEqual(response.data['detail'], 'File not found')

Tasks

upload_species_data

upload_species_data(upload_session_id)

Task for upload species file in the backend.

Parameters:

Name	Type	Description	Default
upload_session_id	int	Id of upload session model	required

Source code in django_project/species/tasks/upload_species.py

@shared_task(name='upload_species_data')
def upload_species_data(upload_session_id):
    """Task for upload species file in the backend.

    :param
    upload_session_id: Id of upload session model
    :type
    upload_session_id: int
    """

    try:
        upload_session = UploadSpeciesCSV.objects.get(id=upload_session_id)
    except UploadSpeciesCSV.DoesNotExist:
        logger.error("upload session doesn't exist")
        return

    encoding = 'utf-8-sig'
    upload_session.progress = 'Processing'
    upload_session.processed = False
    upload_session.success_notes = None
    upload_session.error_notes = None
    upload_session.canceled = False
    if upload_session.success_file:
        try_delete_uploaded_file(upload_session.success_file)
        upload_session.success_file = None
    if upload_session.error_file:
        try_delete_uploaded_file(upload_session.error_file)
        upload_session.error_file = None
    upload_session.save()
    file_upload = SpeciesCSVUpload()
    file_upload.upload_session = upload_session
    file_upload.start(encoding)

Views

TaxonFrontPageListAPIView

Bases: APIView

Fetch taxon list to display on FrontPage.

TaxonListAPIView

Bases: APIView

Get taxon within the organisations

TaxonTrendPageAPIView

Bases: APIView

Fetch taxon detail to display on TrendPage.