conftest.py

import csv
import os
import pytest
import random

from sqlalchemy import select
from flask_login import LoginManager  # type: ignore
import compendium_v2
from compendium_v2.db import db, model, survey_model
from compendium_v2.survey_db import model as survey_db_model
from compendium_v2.auth.session_management import setup_login_manager, User, ROLES


def _test_data_csv(filename):
    data_filename = os.path.join(os.path.dirname(__file__), 'data', filename)
    yield from csv.DictReader(open(data_filename, "r"))


@pytest.fixture
def dummy_config():
    yield {
        'SQLALCHEMY_DATABASE_URI': 'sqlite://',
        'SURVEY_DATABASE_URI': 'sqlite:///',
        'SECRET_KEY': 'testsecret123'
    }


@pytest.fixture
def mocked_admin_user(app, test_survey_data, mocker):
    with app.app_context():
        user = User(email='testemail123@email.local', fullname='testfullname', oidc_sub='fakesub', roles=ROLES.admin)

        nren2 = db.session.scalar(select(model.NREN).filter(model.NREN.name == 'nren2'))
        user.nrens.append(nren2)
        db.session.add(user)
        db.session.commit()

        def user_loader(*args):
            return user
        mocker.patch('flask_login.utils._get_user', user_loader)
        yield user


@pytest.fixture
def mocked_user(app, test_survey_data, mocker):
    with app.app_context():
        user = User(email='testemail123@email.local', fullname='testfullname', oidc_sub='fakesub')

        nren2 = db.session.scalar(select(model.NREN).filter(model.NREN.name == 'nren2'))
        user.nrens.append(nren2)
        db.session.add(user)
        db.session.commit()

        def user_loader(*args):
            return user
        mocker.patch('flask_login.utils._get_user', user_loader)
        yield user


@pytest.fixture
def test_budget_data(app):
    with app.app_context():
        data = [row for row in _test_data_csv("BudgetTestData.csv")]
        nren_names = set([row["nren"] for row in data])
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for row in data:
            nren = nren_dict[row["nren"]]
            budget = row["budget"]
            year = row["year"]

            db.session.add(model.BudgetEntry(nren=nren, budget=float(budget), year=int(year)))
        db.session.commit()


@pytest.fixture
def test_funding_source_data(app):
    with app.app_context():
        data = [row for row in _test_data_csv("FundingSourceTestData.csv")]
        nren_names = set([row["nren"] for row in data])
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for row in data:
            nren = nren_dict[row["nren"]]
            year = row["year"]
            client = row["client"]
            european = row["european"]
            publicbodies = row["publicbodies"]
            commercial = row["commercial"]
            other = row["other"]

            db.session.add(
                model.FundingSource(
                    nren=nren, year=year,
                    client_institutions=client,
                    european_funding=european,
                    gov_public_bodies=publicbodies,
                    commercial=commercial,
                    other=other)
            )
        db.session.commit()


@pytest.fixture
def test_staff_data(app):

    # generator of random test data for 5 years and 100 nrens
    def _generate_rows():
        for nren in ["nren" + str(i) for i in range(1, 100)]:
            for year in range(2016, 2021):
                yield {
                    "nren": nren,
                    "year": year,
                    "permanent_fte": random.randint(0, 100),
                    "subcontracted_fte": random.randint(0, 100),
                    "technical_fte": random.randint(0, 100),
                    "non_technical_fte": random.randint(0, 100)
                }

    with app.app_context():

        data = list(_generate_rows())

        nren_names = set(d['nren'] for d in data)
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for row in data:
            nren = nren_dict[row["nren"]]
            year = row["year"]
            permanent_fte = row["permanent_fte"]
            subcontracted_fte = row["subcontracted_fte"]
            technical_fte = row["technical_fte"]
            non_technical_fte = row["non_technical_fte"]

            db.session.add(
                model.NrenStaff(
                    nren=nren,
                    year=year,
                    permanent_fte=permanent_fte,
                    subcontracted_fte=subcontracted_fte,
                    technical_fte=technical_fte,
                    non_technical_fte=non_technical_fte
                )
            )
        db.session.commit()


@pytest.fixture
def test_survey_data(app):
    with app.app_context():
        nren_names = ['nren1', 'nren2', 'nren3', 'nren4']
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        survey2021 = survey_model.Survey(year=2021, survey={}, status=survey_model.SurveyStatus.published)
        survey2022 = survey_model.Survey(
            year=2022,
            survey={'part1': [{'title': 'ha', 'visibleIf': 'false'}]},
            status=survey_model.SurveyStatus.open
        )
        db.session.add_all([survey2021, survey2022])

        db.session.add(survey_model.SurveyResponse(
            nren=nren_dict['nren1'],
            survey=survey2022,
            answers={},
            status=survey_model.ResponseStatus.completed
        ))

        db.session.commit()


@pytest.fixture
def app(dummy_config):
    app = compendium_v2._create_app_with_db(dummy_config)
    app.config['LOGIN_DISABLED'] = True  # disable login and admin_required decorators
    login_manager = LoginManager()
    login_manager.init_app(app)
    login_manager.login_view = 'authentication.login'
    setup_login_manager(login_manager)
    with app.app_context():
        db.create_all(bind_key=None)
    yield app


@pytest.fixture
def app_with_survey_db(dummy_config):
    dummy_config['SQLALCHEMY_BINDS'] = {survey_db_model.SURVEY_DB_BIND: dummy_config['SURVEY_DATABASE_URI']}
    app = compendium_v2._create_app_with_db(dummy_config)
    with app.app_context():
        db.create_all()
    yield app


@pytest.fixture
def client(app):
    return app.test_client()


@pytest.fixture
def test_charging_structure_data(app):
    with app.app_context():
        data = [row for row in _test_data_csv("ChargingStructureTestData.csv")]
        nren_names = set([row["nren"] for row in data])
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for row in data:
            nren = nren_dict[row["nren"]]
            year = row["year"]
            fee_type = row["fee_type"]
            if fee_type == "null":
                fee_type = None

            db.session.add(
                model.ChargingStructure(
                    nren=nren, year=year,
                    fee_type=fee_type)
            )
        db.session.commit()


@pytest.fixture
def test_organization_data(app):
    def _generate_sub_org_data():
        for nren in ["nren" + str(i) for i in range(1, 50)]:
            for year in range(2016, 2021):
                yield {
                    'nren': nren,
                    'year': year,
                    'name': 'sub_org' + str(random.randint(1, 100)),
                    'role': random.choice(['technical centre', 'stuff', 'test123']),
                }

    def _generate_org_data():
        for nren in ["nren" + str(i) for i in range(1, 50)]:
            for year in range(2016, 2021):
                yield {
                    'nren': nren,
                    'year': year,
                    'name': 'org' + str(year)
                }

    with app.app_context():

        org_data = list(_generate_org_data())
        sub_org_data = list(_generate_sub_org_data())

        nren_names = set(d['nren'] for d in [*org_data, *sub_org_data])
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for org in org_data:
            nren = nren_dict[org["nren"]]
            year = org["year"]
            name = org["name"]

            db.session.add(model.ParentOrganization(nren=nren, year=year, organization=name))

        for sub_org in sub_org_data:
            nren = nren_dict[sub_org["nren"]]
            year = sub_org["year"]
            name = sub_org["name"]
            role = sub_org["role"]

            db.session.add(model.SubOrganization(nren=nren, year=year, organization=name, role=role))

        db.session.commit()


@pytest.fixture
def test_ec_project_data(app):
    def _generate_ec_project_data():
        for nren in ["nren" + str(i) for i in range(1, 50)]:
            for year in range(2016, 2021):
                yield {
                    'nren': nren,
                    'year': year,
                    'project': 'ec_project1',
                }
                if int(nren[-1]) % 2 == 0:
                    yield {
                        'nren': nren,
                        'year': year,
                        'project': 'ec_project2',
                    }

    with app.app_context():

        ec_project_data = list(_generate_ec_project_data())

        nren_names = set(d['nren'] for d in ec_project_data)
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for ec_project in ec_project_data:
            nren = nren_dict[ec_project["nren"]]
            year = ec_project["year"]
            project = ec_project["project"]

            db.session.add(model.ECProject(nren=nren, year=year, project=project))

        db.session.commit()


@pytest.fixture
def test_policy_data(app):
    with app.app_context():
        nrens_and_years = [('nren1', 2019), ('nren1', 2020), ('nren1', 2021), ('nren2', 2019), ('nren2', 2021)]
        nren_names = set(ny[0] for ny in nrens_and_years)

        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for (nren_name, year) in nrens_and_years:
            nren = nren_dict[nren_name]

            db.session.add(model.Policy(
                nren=nren,
                year=year,
                strategic_plan='a strategy',
                environmental='a policy',
                equal_opportunity='another policy',
                connectivity='',
                acceptable_use='',
                privacy_notice='',
                data_protection=''
            ))

        db.session.commit()


@pytest.fixture
def test_traffic_data(app):
    with app.app_context():
        nrens_and_years = [('nren1', 2019), ('nren1', 2020), ('nren1', 2021), ('nren2', 2019), ('nren2', 2021)]
        nren_names = set(ny[0] for ny in nrens_and_years)
        nren_dict = {nren_name: model.NREN(name=nren_name, country='country') for nren_name in nren_names}
        db.session.add_all(nren_dict.values())

        for (nren_name, year) in nrens_and_years:
            nren = nren_dict[nren_name]

            db.session.add(
                model.TrafficVolume(
                    nren=nren,
                    year=year,
                    from_customers=2.23,
                    to_customers=5.2,
                    from_external=0,
                    to_external=1000
                )
            )


@pytest.fixture
def test_institution_urls_data(app):
    def _create_and_save_nrens(nren_names):
        nrens = {}
        for nren_name in nren_names:
            nren_instance = model.NREN(name=nren_name, country='country')
            nrens[nren_name] = nren_instance
            db.session.add(nren_instance)
        return nrens

    def _create_and_save_institution_urls(nrens_and_years, nrens):
        for nren_name, year in nrens_and_years:
            nren_instance = nrens[nren_name]
            urls = ['https://example.com', 'http://example.org']

            institution_urls_model = model.InstitutionURLs(
                nren=nren_instance,
                year=year,
                urls=urls
            )

            db.session.add(institution_urls_model)

    with app.app_context():
        predefined_nrens_and_years = [
            ('nren1', 2019),
            ('nren1', 2020),
            ('nren1', 2021),
            ('nren2', 2019),
            ('nren2', 2021)
        ]

        unique_nren_names = {nren for nren, _ in predefined_nrens_and_years}
        created_nrens = _create_and_save_nrens(unique_nren_names)
        _create_and_save_institution_urls(predefined_nrens_and_years, created_nrens)
        db.session.commit()