test_survey_publisher_legacy_excel.py

import os

from sqlalchemy import select, func

from compendium_v2 import db
from compendium_v2.db import presentation_models
from compendium_v2.publishers.survey_publisher_legacy_excel import _cli

EXCEL_FILE = os.path.join(os.path.dirname(__file__), "data", "2021_Organisation_DataSeries.xlsx")


def test_excel_publisher(app_with_survey_db, mocker):
    mocker.patch('compendium_v2.publishers.excel_parser.EXCEL_FILE_ORGANISATION', EXCEL_FILE)

    with app_with_survey_db.app_context():
        nren_names = ['SURF', 'KIFU', 'University of Malta', 'ASNET-AM', 'SIKT', 'LAT', 'RASH', 'ANAS', 'GRNET', 'CSC',
                      'PIONIER']
        db.session.add_all([presentation_models.NREN(name=nren_name, country='country') for nren_name in nren_names])
        db.session.commit()

    _cli(app_with_survey_db)

    with app_with_survey_db.app_context():
        budget_count = db.session.scalar(select(func.count(presentation_models.BudgetEntry.year)))
        assert budget_count
        funding_source_count = db.session.scalar(select(func.count(presentation_models.FundingSource.year)))
        assert funding_source_count
        charging_structure_count = db.session.scalar(select(func.count(presentation_models.ChargingStructure.year)))
        assert charging_structure_count
        staff_data = db.session.scalars(select(presentation_models.NrenStaff).order_by(
            presentation_models.NrenStaff.year.asc())
        ).all()

        # data should only be saved for the NRENs we have saved in the database
        staff_data_nrens = set([staff.nren.name for staff in staff_data])
        assert len(staff_data_nrens) == len(nren_names) - 1  # no UoM data

        kifu_data = [staff for staff in staff_data if staff.nren.name == 'KIFU']
        # check that the data is saved correctly for KIFU, it should be OK for the rest then..
        assert len(kifu_data) == 6

        assert kifu_data[0].year == 2016
        assert kifu_data[0].permanent_fte == 100
        assert kifu_data[0].subcontracted_fte == 2
        assert kifu_data[0].technical_fte == 0
        assert kifu_data[0].non_technical_fte == 0

        assert kifu_data[1].year == 2017
        assert kifu_data[1].permanent_fte == 80
        assert kifu_data[1].subcontracted_fte == 2
        assert kifu_data[1].technical_fte == 0
        assert kifu_data[1].non_technical_fte == 0

        assert kifu_data[2].year == 2018
        assert kifu_data[2].permanent_fte == 80
        assert kifu_data[2].subcontracted_fte == 3
        assert kifu_data[2].technical_fte == 0
        assert kifu_data[2].non_technical_fte == 0

        assert kifu_data[3].year == 2019
        assert kifu_data[3].permanent_fte == 148
        assert kifu_data[3].subcontracted_fte == 4
        assert kifu_data[3].technical_fte == 117
        assert kifu_data[3].non_technical_fte == 33

        assert kifu_data[4].year == 2020
        assert kifu_data[4].permanent_fte == 190
        assert kifu_data[4].subcontracted_fte == 3
        assert kifu_data[4].technical_fte == 133
        assert kifu_data[4].non_technical_fte == 60

        assert kifu_data[5].year == 2021
        assert kifu_data[5].permanent_fte == 178
        assert kifu_data[5].subcontracted_fte == 3
        assert kifu_data[5].technical_fte == 133
        assert kifu_data[5].non_technical_fte == 45

        ecproject_data = db.session.scalars(select(presentation_models.ECProject)).all()
        # test a couple of random entries
        surf2017 = [x for x in ecproject_data if x.nren.name == 'SURF' and x.year == 2017]
        assert len(surf2017) == 1
        assert surf2017[0].project == 'Asterics and Magic'

        asnetam2018 = [x for x in ecproject_data if x.nren.name == 'ASNET-AM' and x.year == 2018]
        assert len(asnetam2018) == 1
        assert asnetam2018[0].project == 'EaPConnect'

        kifu2019 = [x for x in ecproject_data if x.nren.name == 'KIFU' and x.year == 2019]
        assert len(kifu2019) == 4
        assert kifu2019[3].project == 'SuperHeroes for Science'

        parent_data = db.session.scalars(select(presentation_models.ParentOrganization)).all()
        # test a random entry
        asnet2021 = [x for x in parent_data if x.nren.name == 'ASNET-AM' and x.year == 2021]
        assert len(asnet2021) == 1
        assert asnet2021[0].organization\
            == 'Institute for Informatics and Automation Problems of the National Academy of Sciences of Armenia'

        service_data = db.session.scalars(select(presentation_models.Service)).all()
        assert len(service_data) == 74

        nren_service_data = db.session.scalars(select(presentation_models.NRENService)).all()
        # test a random entry
        sikt2022 = [x for x in nren_service_data
                    if x.nren.name == 'SIKT' and x.year == 2022 and x.service.name == 'Journal access']
        assert len(sikt2022) == 1
        assert sikt2022[0].additional_information.startswith("Sikt negotiates license a")