test_survey_publisher.py 15.85 KiB
from decimal import Decimal
import json
import os
from sqlalchemy import func, select
from compendium_v2 import db
from compendium_v2.db import presentation_model_enums as model_enums, presentation_models
from compendium_v2.publishers.survey_publisher import _map_2023
JSON_FILE = os.path.join(os.path.dirname(__file__), "data", "2023_all_questions_answered.json")
def test_v2_publisher_empty(app):
data = {}
with app.app_context():
nren = presentation_models.NREN(name='name', country='country')
db.session.commit()
with app.app_context():
_map_2023(nren, {"data": data})
db.session.commit()
with app.app_context():
budget_count = db.session.scalar(select(func.count(presentation_models.BudgetEntry.year)))
assert budget_count == 0
# the main thing is actually that it doesnt crash
def test_v2_publisher_full(app):
with open(JSON_FILE) as json_data:
data = json.load(json_data)
with app.app_context():
nren = presentation_models.NREN(name='name', country='country')
db.session.commit()
with app.app_context():
_map_2023(nren, {"data": data})
db.session.commit()
with app.app_context():
budget = db.session.scalar(select(presentation_models.BudgetEntry.budget))
assert budget == Decimal("124.76")
funding_source = db.session.scalar(select(presentation_models.FundingSource))
assert funding_source.client_institutions == Decimal("0")
assert funding_source.european_funding == Decimal("20")
assert funding_source.gov_public_bodies == Decimal("70")
assert funding_source.commercial == Decimal("0")
assert funding_source.other == Decimal("10")
charging_structure = db.session.scalar(select(presentation_models.ChargingStructure.fee_type))
assert charging_structure == model_enums.FeeType.usage_based_fee
staff = db.session.scalar(select(presentation_models.NrenStaff))
assert staff.permanent_fte == Decimal("5.6")
assert staff.subcontracted_fte == Decimal("56")
assert staff.technical_fte == Decimal("2")
assert staff.non_technical_fte == Decimal("1")
parent = db.session.scalar(select(presentation_models.ParentOrganization.organization))
assert parent == "sdtfgd"
subs = db.session.scalars(select(presentation_models.SubOrganization).order_by(
presentation_models.SubOrganization.organization)
)
subs = [s for s in subs]
assert subs[0].organization == "Aaerer"
assert subs[0].role == "Treer"
assert subs[1].organization == "gg"
assert subs[1].role == "hpc"
assert subs[2].organization == "werser"
assert subs[2].role == ""
projects = db.session.scalars(select(presentation_models.ECProject.project).order_by(
presentation_models.ECProject.project)
)
projects = [p for p in projects]
assert projects[0] == "dgdg"
assert projects[1] == "rrrrr"
assert projects[2] == "st"
policy = db.session.scalar(select(presentation_models.Policy))
assert policy.strategic_plan == "https://serere.com"
assert policy.environmental == "http://mren.ac.me/documents/Environmental%20policy%20for%20MREN.pdf"
assert policy.equal_opportunity == ""
assert policy.connectivity == "http://www.ucg.ac.me/skladiste/blog_1028/objava_10/fajlovi/Pravila%20o%20koriscenju%20Akademske%20mreze%20Univerziteta%20Crne%20Gore.pdf" # noqa: E501
assert policy.acceptable_use == "http://www.ucg.ac.me/skladiste/blog_1028/objava_10/fajlovi/Pravila%20o%20koriscenju%20Akademske%20mreze%20Univerziteta%20Crne%20Gore.pdf" # noqa: E501
assert policy.privacy_notice == ""
assert policy.data_protection == ""
assert policy.gender_equality == "https://www.ucg.ac.me/objava/blog/616808/objava/148423-plan-rodne-ravnopravnosti-univerziteta-crne-gore" # noqa: E501
traffic = db.session.scalar(select(presentation_models.TrafficVolume))
assert traffic.to_customers == Decimal("3")
assert traffic.from_customers == Decimal("34")
assert traffic.to_external == Decimal("22")
assert traffic.from_external == Decimal("3")
client_urls = db.session.scalar(select(presentation_models.InstitutionURLs))
assert client_urls.urls == ["http://erse.com", "https://wwe.com"]
procurement = db.session.scalar(select(presentation_models.CentralProcurement))
assert procurement.central_procurement
assert procurement.amount == Decimal("57676")
service_management = db.session.scalar(select(presentation_models.ServiceManagement))
assert service_management.service_management_framework
assert not service_management.service_level_targets
service_user_types = [s for s in db.session.scalars(select(presentation_models.ServiceUserTypes))]
user_categories = set([s.user_category for s in service_user_types])
assert len(user_categories) == 10
institute_services = [
s for s in service_user_types if s.user_category == model_enums.UserCategory.institutes
]
assert len(institute_services) == 7
eosc_listings = db.session.scalar(select(presentation_models.EOSCListings))
assert eosc_listings.service_names == ["dy", "rrr"]
standards = db.session.scalar(select(presentation_models.Standards))
assert standards.audits
assert standards.audit_specifics == "isooo 557"
assert standards.business_continuity_plans
assert standards.business_continuity_plans_specifics == "no"
assert not standards.crisis_management_procedure
crisis_excercises = db.session.scalar(select(presentation_models.CrisisExcercises))
assert crisis_excercises.exercise_descriptions == [
"geant_workshops", "national_excercises", "tabletop_exercises", "other_excercises",
"none", "simulation_excercises", "real_crisis"
]
security_controls = db.session.scalar(select(presentation_models.SecurityControls))
assert security_controls.security_control_descriptions == [
"monitoring", "acl", "anti_virus", "firewall", "anti_spam", "ddos_mitigation",
"ips_ids", "segmentation", "integrity_checking", "vgjh"
]
connected_proportion = [s for s in db.session.scalars(select(presentation_models.ConnectedProportion))]
assert len(connected_proportion) == 10
uni = [c for c in connected_proportion if c.user_category == model_enums.UserCategory.universities]
assert len(uni) == 1
assert uni[0].coverage == model_enums.ConnectivityCoverage.yes_national_nren
assert uni[0].number_connected == 19
assert uni[0].market_share == Decimal("1")
assert uni[0].users_served == 20000
connectivity_level = [s for s in db.session.scalars(select(presentation_models.ConnectivityLevel))]
assert len(connectivity_level) == 10
gov = [c for c in connectivity_level if c.user_category == model_enums.UserCategory.government]
assert len(gov) == 1
assert gov[0].typical_speed == 100
assert gov[0].highest_speed == 100
assert gov[0].highest_speed_proportion == Decimal("100")
carrier = [s for s in db.session.scalars(select(presentation_models.ConnectionCarrier))]
assert len(carrier) == 10
profit = [c for c in carrier if c.user_category == model_enums.UserCategory.for_profit_orgs]
assert len(profit) == 1
assert profit[0].carry_mechanism == model_enums.CarryMechanism.commercial_provider_backbone
primary = [c for c in carrier if c.user_category == model_enums.UserCategory.primary_schools]
assert len(primary) == 1
assert primary[0].carry_mechanism == model_enums.CarryMechanism.man
hospital = [c for c in carrier if c.user_category == model_enums.UserCategory.hospitals]
assert len(hospital) == 1
assert hospital[0].carry_mechanism == model_enums.CarryMechanism.nren_local_loops
load = [s for s in db.session.scalars(select(presentation_models.ConnectivityLoad))]
assert len(load) == 10
cultural = [c for c in load if c.user_category == model_enums.UserCategory.cultural]
assert len(cultural) == 1
assert cultural[0].average_load_from_institutions == 10
assert cultural[0].average_load_to_institutions == 10
assert cultural[0].peak_load_from_institutions == 50
assert cultural[0].peak_load_to_institutions == 10
growth = [s for s in db.session.scalars(select(presentation_models.ConnectivityGrowth))]
assert len(growth) == 10
cultural = [c for c in growth if c.user_category == model_enums.UserCategory.cultural]
assert len(cultural) == 1
assert cultural[0].growth == Decimal("50")
commercial = db.session.scalar(select(presentation_models.CommercialConnectivity))
assert commercial.commercial_r_and_e == model_enums.CommercialConnectivityCoverage.no_but_direct_peering
assert commercial.commercial_general == model_enums.CommercialConnectivityCoverage.no_policy
assert commercial.commercial_collaboration == model_enums.CommercialConnectivityCoverage.yes_incl_other
assert commercial.commercial_service_provider == model_enums.CommercialConnectivityCoverage.no_other
assert commercial.university_spin_off == model_enums.CommercialConnectivityCoverage.no_financial
commercial_charging = db.session.scalar(select(presentation_models.CommercialChargingLevel))
assert commercial_charging.collaboration == model_enums.CommercialCharges.no_charges_if_r_e_requested
assert commercial_charging.service_supplier == model_enums.CommercialCharges.same_as_r_e_charges
assert commercial_charging.direct_peering == model_enums.CommercialCharges.higher_than_r_e_charges
remote_campuses = db.session.scalar(select(presentation_models.RemoteCampuses))
assert remote_campuses.remote_campus_connectivity
assert remote_campuses.connections == [
{"country": "fyuyg", "local_r_and_e_connection": False},
{"country": "g", "local_r_and_e_connection": None}
]
dark_fibre_lease = db.session.scalar(select(presentation_models.DarkFibreLease))
assert dark_fibre_lease.iru_or_lease
assert dark_fibre_lease.fibre_length_in_country == 405
assert dark_fibre_lease.fibre_length_outside_country == 0
assert dark_fibre_lease.iru_duration == Decimal("10")
dark_fibre_installed = db.session.scalar(select(presentation_models.DarkFibreInstalled))
assert dark_fibre_installed.installed
assert dark_fibre_installed.fibre_length_in_country == 1
fibre_light = db.session.scalar(select(presentation_models.FibreLight))
assert fibre_light.light_description == "jj"
network_map_urls = db.session.scalar(select(presentation_models.NetworkMapUrls))
assert network_map_urls.urls == ["http://netmon.ucg.ac.me/mantra/"]
monitoring_tools = db.session.scalar(select(presentation_models.MonitoringTools))
assert monitoring_tools.tool_descriptions == [
"status_dashboard", "historical_traffic_volumes", "netflow_analysis", "looking_glass", "hiui"
]
assert monitoring_tools.netflow_processing_description == "ghhhh"
passive_monitoring = db.session.scalar(select(presentation_models.PassiveMonitoring))
assert passive_monitoring.monitoring
assert passive_monitoring.method == model_enums.MonitoringMethod.both
traffic_statistics = db.session.scalar(select(presentation_models.TrafficStatistics))
assert traffic_statistics.traffic_statistics
assert traffic_statistics.urls == [
"http://netmon.cis.ac.me/cacti/graph_view.php?action=tree&tree_id=36&leaf_id=1430"
]
siem_vendors = db.session.scalar(select(presentation_models.SiemVendors))
assert siem_vendors.vendor_names == ["Exabeam", "Splunk", "IBM Qradar", "LogRythm", "Securonix", "figuoho"]
certificate_providers = db.session.scalar(select(presentation_models.CertificateProviders))
assert certificate_providers.provider_names == [
"TCS", "Digicert", "Sectigo", "GlobalSign", "GeoDaddy", "GeoTrust", "Entrust Datacard", "earer"
]
weather_map = db.session.scalar(select(presentation_models.WeatherMap))
assert weather_map.weather_map
assert weather_map.url == "http://weather.com"
pert_team = db.session.scalar(select(presentation_models.PertTeam))
assert pert_team.pert_team == model_enums.YesNoPlanned.planned
alien_wave = db.session.scalar(select(presentation_models.AlienWave))
assert alien_wave.alien_wave_third_pary == model_enums.YesNoPlanned.yes
assert alien_wave.nr_of_alien_wave_third_party_services == 66
assert not alien_wave.alien_wave_internal
capacity = db.session.scalar(select(presentation_models.Capacity))
assert capacity.largest_link_capacity == Decimal("1")
assert capacity.typical_backbone_capacity == Decimal("567")
external_connections = db.session.scalar(select(presentation_models.ExternalConnections))
external_connection_list = external_connections.connections
assert len(external_connection_list) == 6
assert external_connection_list[0] == {
"capacity": "1",
"from_organization": "GEANT",
"interconnection_method": "geant",
"link_name": "GEANT",
"to_organization": "MREN"
}
assert external_connection_list[3] == {
"capacity": None,
"from_organization": "",
"interconnection_method": "other",
"link_name": "",
"to_organization": ""
}
assert external_connection_list[4] == {
"capacity": "1.1",
"from_organization": "",
"interconnection_method": None,
"link_name": "",
"to_organization": "",
}
non_r_e_peers = db.session.scalar(select(presentation_models.NonREPeers))
assert non_r_e_peers.nr_of_non_r_and_e_peers == 0
traffic_ratio = db.session.scalar(select(presentation_models.TrafficRatio))
assert traffic_ratio.r_and_e_percentage == Decimal("1")
assert traffic_ratio.commodity_percentage == Decimal("99")
ops_automation = db.session.scalar(select(presentation_models.OpsAutomation))
assert ops_automation.ops_automation == model_enums.YesNoPlanned.yes
assert ops_automation.ops_automation_specifics == "ghioil;"
nfv = db.session.scalar(select(presentation_models.NetworkFunctionVirtualisation))
assert nfv.nfv == model_enums.YesNoPlanned.planned
assert nfv.nfv_specifics == [
"routers", "firewalls", "load_balancers", "vpn_concentrators", "rtghjk"
]
network_automation = db.session.scalar(select(presentation_models.NetworkAutomation))
assert network_automation.network_automation == model_enums.YesNoPlanned.planned
assert network_automation.network_automation_specifics == [
"config_management", "provisioning", "data_collection", "compliance", "reporting", "troubleshooting"
]
services = [s for s in db.session.scalars(select(presentation_models.NRENService))]
assert len(services) == 59
virtual_learning_environment = [s for s in services if s.service_key == "virtual-learning-environment"]
assert len(virtual_learning_environment) == 1
assert virtual_learning_environment[0].additional_information == "zd"
csirt = [s for s in services if s.service_key == "csirt"]
assert len(csirt) == 1
assert csirt[0].official_description == "zdf"
connectivity = [s for s in services if s.service_key == "connectivity"]
assert len(connectivity) == 1
assert connectivity[0].product_name == "zfz"