import os

import jpype
import pytest
from numpy import testing as npt

import ixmp
from ixmp.default_path_constants import CONFIG_PATH

test_args = ('Douglas Adams', 'Hitchhiker')
can_args = ('canning problem', 'standard')


def local_config_exists():
    return os.path.exists(CONFIG_PATH)


@pytest.mark.skipif(local_config_exists(),
                    reason='will not overwrite local config files')
def test_default_dbprops_file(test_mp_use_default_dbprops_file):
    test_mp = test_mp_use_default_dbprops_file
    scenario = test_mp.scenario_list(model='Douglas Adams')['scenario']
    assert scenario[0] == 'Hitchhiker'


@pytest.mark.skipif(local_config_exists(),
                    reason='will not overwrite local config files')
def test_db_config_path(test_mp_use_db_config_path):
    test_mp = test_mp_use_db_config_path
    scenario = test_mp.scenario_list(model='Douglas Adams')['scenario']
    assert scenario[0] == 'Hitchhiker'


def test_platform_init_raises():
    pytest.raises(ValueError, ixmp.Platform, dbtype='foo')


def test_scen_list(test_mp):
    scenario = test_mp.scenario_list(model='Douglas Adams')['scenario']
    assert scenario[0] == 'Hitchhiker'


def test_new_scen(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args, version='new')
    assert scen.version == 0


def test_default_version(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    assert scen.version == 2


def test_has_set(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    assert scen.has_set('i')
    assert not scen.has_set('k')


def test_init_par_35(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args, version='new')
    scen.init_set('ii')
    scen.init_par('new_par', idx_sets='ii')


def test_get_scalar(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    obs = scen.scalar('f')
    exp = {'unit': 'USD/km', 'value': 90}
    assert obs == exp


def test_has_par(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    assert scen.has_par('f')
    assert not scen.has_par('m')


def test_init_scalar(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    scen2 = scen.clone(keep_solution=False)
    scen2.check_out()
    scen2.init_scalar('g', 90.0, 'USD/km')
    scen2.commit("adding a scalar 'g'")


def test_init_set(test_mp):
    """Test ixmp.Scenario.init_set()."""
    scen = ixmp.Scenario(test_mp, *can_args)

    # Add set on a locked scenario
    with pytest.raises(jpype.JavaException,
                       message="at.ac.iiasa.ixmp.exceptions.IxException: "
                               "This Scenario cannot be edited, do a checkout "
                               "first!"):
        scen.init_set('foo')

    scen = scen.clone(keep_solution=False)
    scen.check_out()
    scen.init_set('foo')

    # Initialize an already-existing set
    with pytest.raises(jpype.JavaException,
                       message="at.ac.iiasa.ixmp.exceptions.IxException: "
                               "An Item with the name 'foo' already exists!"):
        scen.init_set('foo')


def test_add_set(test_mp):
    """Test ixmp.Scenario.add_set()."""
    scen = ixmp.Scenario(test_mp, *can_args)

    # Add element to a non-existent set
    with pytest.raises(jpype.JavaException,
                       message="at.ac.iiasa.ixmp.exceptions.IxException: "
                               "No Set 'foo' exists in this Scenario!"):
        scen.add_set('foo', 'bar')


# make sure that changes to a scenario are copied over during clone
def test_add_clone(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args, version=1)
    scen.check_out()
    scen.init_set('h')
    scen.add_set('h', 'test')
    scen.commit("adding an index set 'h', wiht element 'test'")

    scen2 = scen.clone(keep_solution=False)
    obs = scen2.set('h')
    npt.assert_array_equal(obs, ['test'])


# make sure that (only) the correct scenario is touched after cloning
def test_clone_edit(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    scen2 = scen.clone(keep_solution=False)
    scen2.check_out()
    scen2.change_scalar('f', 95.0, 'USD/km')
    scen2.commit('change transport cost')
    obs = scen.scalar('f')
    exp = {'unit': 'USD/km', 'value': 90}
    assert obs == exp
    obs = scen2.scalar('f')
    exp = {'unit': 'USD/km', 'value': 95}
    assert obs == exp


def test_idx_name(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    df = scen.idx_names('d')
    npt.assert_array_equal(df, ['i', 'j'])


def test_has_var(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    assert scen.has_var('x')
    assert not scen.has_var('y')


def test_var_marginal(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    df = scen.var('x', filters={'i': ['seattle']})
    npt.assert_array_almost_equal(df['mrg'], [0, 0, 0.036])


def test_var_level(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    df = scen.var('x', filters={'i': ['seattle']})
    npt.assert_array_almost_equal(df['lvl'], [50, 300, 0])


def test_var_general_str(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    df = scen.var('x', filters={'i': ['seattle']})
    npt.assert_array_equal(
        df['j'], ['new-york', 'chicago', 'topeka'])


def test_unit_list(test_mp):
    units = test_mp.units()
    assert ('cases' in units) is True


def test_add_unit(test_mp):
    test_mp.add_unit('test', 'just testing')


def test_par_filters_unit(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args)
    df = scen.par('d', filters={'i': ['seattle']})
    obs = df.loc[0, 'unit']
    exp = 'km'
    assert obs == exp


def test_meta(test_mp):
    test_dict = {
        "test_string": 'test12345',
        "test_number": 123.456,
        "test_number_negative": -123.456,
        'test_int': 12345,
        'test_bool': True,
        'test_bool_false': False,
    }

    scen = ixmp.Scenario(test_mp, *can_args, version=1)
    for k, v in test_dict.items():
        scen.set_meta(k, v)

    # test all
    obs_dict = scen.get_meta()
    for k, exp in test_dict.items():
        obs = obs_dict[k]
        assert obs == exp

    # test name
    obs = scen.get_meta('test_string')
    exp = test_dict['test_string']
    assert obs == exp


def test_load_scenario_data(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args, cache=True)
    scen.load_scenario_data()
    assert ('par', 'd') in scen._pycache  # key exists
    df = scen.par('d', filters={'i': ['seattle']})
    obs = df.loc[0, 'unit']
    exp = 'km'
    assert obs == exp


def test_load_scenario_data_clear_cache(test_mp):
    # this fails on commit: 4376f54
    scen = ixmp.Scenario(test_mp, *can_args, cache=True)
    scen.load_scenario_data()
    scen.clear_cache(name='d')


def test_load_scenario_data_raises(test_mp):
    scen = ixmp.Scenario(test_mp, *can_args, cache=False)
    pytest.raises(ValueError, scen.load_scenario_data)


def test_log_level(test_mp):
    test_mp.set_log_level('CRITICAL')
    test_mp.set_log_level('ERROR')
    test_mp.set_log_level('WARNING')
    test_mp.set_log_level('INFO')
    test_mp.set_log_level('DEBUG')
    test_mp.set_log_level('NOTSET')


def test_log_level_raises(test_mp):
    pytest.raises(ValueError, test_mp.set_log_level, level='foo')
