conftest.py

import csv
import os
import pytest
import random

import compendium_v2
from compendium_v2.db import db, model
from compendium_v2.survey_db import model as survey_model


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 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 app(dummy_config):
    app = compendium_v2._create_app_with_db(dummy_config)
    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_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()