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"]