Skip to content
Snippets Groups Projects
test_survey_publisher_old_db_2022.py 10.81 KiB
from sqlalchemy import select

from compendium_v2.db import db, presentation_model_enums, presentation_models
from compendium_v2.publishers.survey_publisher_old_db_2022 import _cli, FundingSource, \
    StaffQuestion, OrgQuestion, ChargingStructure, ECQuestion


def org_data(question):
    """
    This function defines test data for the org questions.

    The following data is defined for the appropriate questions as modeled in Compendium:

    nren1,CYNET-CSIRT,cert team
    nren1,DFN-CERT,CERT
    nren2,Educampus Services,MIS shared services for third level.
    nren3,VilniusTech,Technical centre
    nren3,KU,Technical centre
    nren3,VDU,Technical centre
    nren3,VU,Technical centre
    nren3,org_data
org_data
org_data
org_data
org_dataKTU,"NOC, administrative authority"
    """

    if question == OrgQuestion.PARENT_ORG_NAME:
        return [
            ('nren1', 'Org1'),
            ('nren3', 'Org3'),
        ]

    if str(question.name).endswith('1_NAME'):
        return [
            ('nren1', 'CYNET-CSIRT'),
            ('nren2', 'Educampus Services'),
            ('nren3', 'VilniusTech'),
        ]

    if str(question.name).endswith('2_NAME'):
        return [
            ('nren1', 'DFN-CERT'),
            ('nren3', 'KU'),
        ]

    if str(question.name).endswith('3_NAME'):
        return [
            ('nren3', 'VDU'),
        ]

    if str(question.name).endswith('4_NAME'):
        return [
            ('nren3', 'VU'),
        ]

    if str(question.name).endswith('5_NAME'):
        return [
            ('nren3', 'KTU'),
        ]

    if str(question.name).endswith('1_CHOICE'):
        return [
            ('nren1', 'other'),
            ('nren2', 'other'),
            ('nren3', 'Technical centre'),
        ]

    if str(question.name).endswith('2_CHOICE'):
        return [
            ('nren1', 'other'),
            ('nren3', 'Technical centre'),
        ]

    if str(question.name).endswith('3_CHOICE'):
        return [
            ('nren3', 'Technical centre'),
        ]

    if str(question.name).endswith('4_CHOICE'):
        return [
            ('nren3', 'Technical centre'),
        ]

    if str(question.name).endswith('5_CHOICE'):
        return [
            ('nren3', 'other'),
        ]

    if str(question.name).endswith('1_ROLE'):
        return [
            ('nren1', 'cert team'),
            ('nren2', 'MIS shared services for third level.')
        ]

    if str(question.name).endswith('2_ROLE'):
        return [
            ('nren1', 'CERT'),
        ]

    if str(question.name).endswith('3_ROLE'):
        return []

    if str(question.name).endswith('4_ROLE'):
        return []

    if str(question.name).endswith('5_ROLE'):
        return [
            ('nren3', 'NOC, administrative authority')
        ]


def test_publisher(app_with_survey_db, mocker):
    global org_data

    def get_rows_as_tuples(*args, **kwargs):
        return [
            ('nren1', '100'),
            ('nren2', '200'),
            ('nren3', '300'),
            ('nren4', 'abcd')
        ]

    def funding_source_data():
        yield FundingSource.CLIENT_INSTITUTIONS, [
            ('nren1', '10'),
            ('nren2', '80'),
            ('nren3', '30'),
        ]
        yield FundingSource.EUROPEAN_FUNDING, [
            ('nren1', '50'),
            ('nren2', '20'),
            ('nren3', '30'),
        ]
        yield FundingSource.OTHER, [
            ('nren1', '40'),
            ('nren2', 'abc'),
            ('nren3', '30'),
        ]

    def question_data(question):
        if question == StaffQuestion.NON_TECHNICAL_FTE:
            return [
                ('nren1', '10'),
                ('nren2', '80'),
                ('nren3', '30'),
            ]

        if question == StaffQuestion.TECHNICAL_FTE:
            return [
                ('nren1', '50'),
                ('nren2', '20'),
                ('nren3', '30'),
            ]

        if question == StaffQuestion.PERMANENT_FTE:
            return [
                ('nren1', '60'),
                ('nren2', 'abc'),
                ('nren3', '30'),
            ]

        if question == StaffQuestion.SUBCONTRACTED_FTE:
            return [
                ('nren1', '0'),
                ('nren2', '0'),
                ('nren3', '0'),
            ]
        if question in OrgQuestion:
            return org_data(question)

        if question == ChargingStructure.charging_structure:
            return [
                ('nren1', 'We do not charge them directly'),
                ('nren2', 'We charge a usage-based fee'),
                ('nren3', 'Other'),
            ]

        if question in ECQuestion:
            return [
                ('nren1', '[""'),
                ('nren2', '["project1", "project2"]'),
                ('nren3', '["project3"]'),
            ]

    def question_id_data(question_id, year):
        if question_id in [
                16469, 16064, 15720, 15305, 14910, 16471, 16066, 15722, 15307, 14912, 16473, 16378,
                16475, 16068, 15724, 15309, 14914, 16477, 16070, 15726, 15311, 14916, 16479, 16072, 15728, 15575,
                16481, 16074, 15730, 15577, 16761]:
            return [
                ('nren1', f'www.nren.com/somepolicy{year}.pdf'),
                ('nren2', 'policyemail@nren.com'),
                ('nren3', 'n.a. online'),
            ]

    def institutions_urls_data(question_id):
        if question_id == 16507:
            return [
                (87483, 'ANA', 2013, "http://www.rash.al/index.php/network/points-of-presence-pop"),
                (163286, 'ANA', 2014, "http://www.rash.al/index.php/network/points-of-presence-pop"),
            ]
        else:
            return []

    mocker.patch('compendium_v2.publishers.survey_publisher_old_db_2022.query_budget', get_rows_as_tuples)
    mocker.patch('compendium_v2.publishers.survey_publisher_old_db_2022.query_funding_sources', funding_source_data)
    mocker.patch('compendium_v2.publishers.survey_publisher_old_db_2022.query_question', question_data)
    mocker.patch('compendium_v2.publishers.survey_publisher_old_db_2022.query_question_id', question_id_data)
    mocker.patch('compendium_v2.publishers.survey_publisher_old_db_2022.recursive_query', institutions_urls_data)

    nren_names = ['Nren1', 'Nren2', 'Nren3', 'Nren4', 'SURF', 'KIFU', 'University of Malta', 'ASNET-AM',
                  'SIKT', 'LAT', 'RASH', 'AzScienceNet', 'GRNET', 'CSC', 'PSNC']
    with app_with_survey_db.app_context():
        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():
        budgets = db.session.scalars(
            select(presentation_models.BudgetEntry).order_by(presentation_models.BudgetEntry.nren_id.asc())
        ).all()
        assert len(budgets) == 3
        assert budgets[0].nren.name.lower() == 'nren1'
        assert budgets[0].budget == 100

        funding_sources = db.session.scalars(
            select(presentation_models.FundingSource).order_by(presentation_models.FundingSource.nren_id.asc())
        ).all()
        assert len(funding_sources) == 2
        assert funding_sources[0].nren.name.lower() == 'nren1'
        assert funding_sources[0].client_institutions == 10
        assert funding_sources[0].european_funding == 50
        assert funding_sources[0].other == 40

        assert funding_sources[1].nren.name.lower() == 'nren2'
        assert funding_sources[1].client_institutions == 80
        assert funding_sources[1].european_funding == 20
        assert funding_sources[1].other == 0

        staff_data = db.session.scalars(
            select(presentation_models.NrenStaff).order_by(presentation_models.NrenStaff.nren_id.asc())
        ).all()

        assert len(staff_data) == 1
        assert staff_data[0].nren.name.lower() == 'nren1'
        assert staff_data[0].non_technical_fte == 10
        assert staff_data[0].technical_fte == 50
        assert staff_data[0].permanent_fte == 60
        assert staff_data[0].subcontracted_fte == 0

        _org_data = db.session.scalars(
            select(presentation_models.ParentOrganization).order_by(
                presentation_models.ParentOrganization.nren_id.asc())
        ).all()

        assert len(_org_data) == 2
        assert _org_data[0].nren.name.lower() == 'nren1'
        assert _org_data[0].organization == 'Org1'

        assert _org_data[1].nren.name.lower() == 'nren3'
        assert _org_data[1].organization == 'Org3'

        charging_structures = db.session.scalars(
            select(presentation_models.ChargingStructure).order_by(presentation_models.ChargingStructure.nren_id.asc())
        ).all()
        assert len(charging_structures) == 3
        assert charging_structures[0].nren.name.lower() == 'nren1'
        assert charging_structures[0].fee_type == presentation_model_enums.FeeType.no_charge
        assert charging_structures[1].nren.name.lower() == 'nren2'
        assert charging_structures[1].fee_type == presentation_model_enums.FeeType.usage_based_fee
        assert charging_structures[2].nren.name.lower() == 'nren3'
        assert charging_structures[2].fee_type == presentation_model_enums.FeeType.other

        _ec_data = db.session.scalars(
            select(presentation_models.ECProject).order_by(presentation_models.ECProject.nren_id.asc())
        ).all()

        assert len(_ec_data) == 3
        assert _ec_data[0].nren.name.lower() == 'nren2'
        assert _ec_data[0].project == 'project1'

        assert _ec_data[1].nren.name.lower() == 'nren2'
        assert _ec_data[1].project == 'project2'

        assert _ec_data[2].nren.name.lower() == 'nren3'
        assert _ec_data[2].project == 'project3'

        policy_data = db.session.scalars(
            select(presentation_models.Policy).order_by(presentation_models.Policy.nren_id.asc())
        ).all()
        policy_data_2020 = [p for p in policy_data if p.year == 2020]
        policy_data_2022 = [p for p in policy_data if p.year == 2022]
        assert len(policy_data_2020) == 2
        assert len(policy_data_2022) == 2
        assert policy_data_2020[0].strategic_plan == 'www.nren.com/somepolicy2020.pdf'
        assert policy_data_2020[1].strategic_plan == 'policyemail@nren.com'

        _institution_urls_data = db.session.scalars(
            select(presentation_models.InstitutionURLs).order_by(presentation_models.InstitutionURLs.nren_id.asc())
        ).all()
        assert len(_institution_urls_data) == 2
        assert _institution_urls_data[0].nren.name.lower() == 'rash'
        assert _institution_urls_data[0].year == 2013
        assert _institution_urls_data[0].urls == ["http://www.rash.al/index.php/network/points-of-presence-pop"]
        assert _institution_urls_data[1].nren.name.lower() == 'rash'
        assert _institution_urls_data[1].year == 2014
        assert _institution_urls_data[1].urls == ["http://www.rash.al/index.php/network/points-of-presence-pop"]