def test_D8_D4_fill(d4_grid):
    """
    Tests the functionality of D4 filling.
    """
    d4_grid.lfD8.map_depressions(pits=None, reroute_flow=False)
    d4_grid.lfD4.map_depressions(pits=None, reroute_flow=False)
    assert d4_grid.lfD8.number_of_lakes == 1
    assert d4_grid.lfD4.number_of_lakes == 3

    correct_D8_lake_map = np.empty(7 * 7, dtype=int)
    correct_D8_lake_map.fill(XX)
    correct_D8_lake_map[d4_grid.lake_nodes] = 10
    correct_D4_lake_map = correct_D8_lake_map.copy()
    correct_D4_lake_map[d4_grid.lake_nodes[5:]] = 32
    correct_D4_lake_map[d4_grid.lake_nodes[-2]] = 38
    correct_D8_depths = np.zeros(7 * 7, dtype=float)
    correct_D8_depths[d4_grid.lake_nodes] = 2.
    correct_D4_depths = correct_D8_depths.copy()
    correct_D4_depths[d4_grid.lake_nodes[5:]] = 4.
    correct_D4_depths[d4_grid.lake_nodes[-2]] = 3.

    assert_array_equal(d4_grid.lfD8.lake_map, correct_D8_lake_map)
    assert_array_equal(d4_grid.lfD4.lake_map, correct_D4_lake_map)

    assert d4_grid.mg1.at_node["depression__depth"] == approx(correct_D8_depths)
    assert d4_grid.mg2.at_node["depression__depth"] == approx(correct_D4_depths)

 def test_int(self):
     within_1e6 = [(1000001, 1000000), (-1000001, -1000000)]
     for a, x in within_1e6:
         assert a == approx(x, rel=5e-6, abs=0)
         assert a != approx(x, rel=5e-7, abs=0)
         assert approx(x, rel=5e-6, abs=0) == a
         assert approx(x, rel=5e-7, abs=0) != a

    def test_uses_named_inputs(self):
        inputs = {
                "premise": "I always write unit tests for my code.",
                "hypothesis": "One time I didn't write any unit tests for my code."
        }

        archive = load_archive(self.FIXTURES_ROOT / 'decomposable_attention' / 'serialization' / 'model.tar.gz')
        predictor = Predictor.from_archive(archive, 'textual-entailment')
        result = predictor.predict_json(inputs)

        # Label probs should be 3 floats that sum to one
        label_probs = result.get("label_probs")
        assert label_probs is not None
        assert isinstance(label_probs, list)
        assert len(label_probs) == 3
        assert all(isinstance(x, float) for x in label_probs)
        assert all(x >= 0 for x in label_probs)
        assert sum(label_probs) == approx(1.0)

        # Logits should be 3 floats that softmax to label_probs
        label_logits = result.get("label_logits")
        assert label_logits is not None
        assert isinstance(label_logits, list)
        assert len(label_logits) == 3
        assert all(isinstance(x, float) for x in label_logits)

        exps = [math.exp(x) for x in label_logits]
        sumexps = sum(exps)
        for e, p in zip(exps, label_probs):
            assert e / sumexps == approx(p)

def test_random_splitter(test_specs, spark_dataset):
    """Test random splitter for Spark dataframes.

    NOTE: some split results may not match exactly with the ratios, which may be owing to the
    limited number of rows in
    the testing data. A approximate match with certain level of tolerance is therefore used
    instead for tests.
    """
    splits = spark_random_split(
        spark_dataset, ratio=test_specs["ratio"], seed=test_specs["seed"]
    )

    assert splits[0].count() / test_specs["number_of_rows"] == pytest.approx(
        test_specs["ratio"], test_specs["spark_randomsplit_tolerance"]
    )
    assert splits[1].count() / test_specs["number_of_rows"] == pytest.approx(
        1 - test_specs["ratio"], test_specs["spark_randomsplit_tolerance"]
    )

    splits = spark_random_split(
        spark_dataset, ratio=test_specs["ratios"], seed=test_specs["seed"]
    )

    assert splits[0].count() / test_specs["number_of_rows"] == pytest.approx(
        test_specs["ratios"][0], test_specs["spark_randomsplit_tolerance"]
    )
    assert splits[1].count() / test_specs["number_of_rows"] == pytest.approx(
        test_specs["ratios"][1], test_specs["spark_randomsplit_tolerance"]
    )
    assert splits[2].count() / test_specs["number_of_rows"] == pytest.approx(
        test_specs["ratios"][2], test_specs["spark_randomsplit_tolerance"]
    )

def test_pv_properties(next100pv):
    pv = next100pv
    assert pv.name                           == 'Next100PV'
    assert pv.material_name                  == ti316.name
    assert pv.radius / mm                     == approx(1360 / 2, rel=1e-3)
    assert pv.body_thickness / mm            == approx(10, rel=1e-3)
    assert pv.head_thickness / mm            == approx(12, rel=1e-3)

def test_formatters(Formatter, regex, direction, factor, values):
    fmt = Formatter()
    result = fmt(direction, factor, values)

    prev_degree = prev_minute = prev_second = None
    for tick, value in zip(result, values):
        m = regex.match(tick)
        assert m is not None, '"%s" is not an expected tick format.' % (tick, )

        sign = sum(m.group(sign + '_sign') is not None
                   for sign in ('degree', 'minute', 'second'))
        assert sign <= 1, \
            'Only one element of tick "%s" may have a sign.' % (tick, )
        sign = 1 if sign == 0 else -1

        degree = float(m.group('degree') or prev_degree or 0)
        minute = float(m.group('minute') or prev_minute or 0)
        second = float(m.group('second') or prev_second or 0)
        if Formatter == FormatterHMS:
            # 360 degrees as plot range -> 24 hours as labelled range
            expected_value = pytest.approx((value // 15) / factor)
        else:
            expected_value = pytest.approx(value / factor)
        assert sign * dms2float(degree, minute, second) == expected_value, \
            '"%s" does not match expected tick value.' % (tick, )

        prev_degree = degree
        prev_minute = minute
        prev_second = second

def test_multidim_tendencies():
    # Same test just repeated in two parallel columns
    num_lat = 2
    state = climlab.column_state(num_lev=num_lev, num_lat=num_lat)
    state['q'] = state.Tatm * 0. #+ Q
    state['U'] = state.Tatm * 0. #+ U
    state['V'] = state.Tatm * 0. #+ V
    for i in range(num_lat):
        state.Tatm[i,:] = T
        state['q'][i,:] += Q
        state['U'][i,:] += U
        state['V'][i,:] += V
    assert hasattr(state, 'Tatm')
    assert hasattr(state, 'q')
    assert hasattr(state, 'U')
    assert hasattr(state, 'V')
    conv = emanuel_convection.EmanuelConvection(state=state, timestep=DELT)
    conv.step_forward()
    #  Did we get all the correct output?
    assert np.all(conv.IFLAG == 1)
    #  relative tolerance for these tests ...
    tol = 1E-5
    assert np.all(conv.CBMF == pytest.approx(3.10377218E-02, rel=tol))
    tend = conv.tendencies
    assert np.tile(FT,(num_lat,1)) == pytest.approx(tend['Tatm'], rel=tol)
    assert np.tile(FQ,(num_lat,1)) == pytest.approx(tend['q'], rel=tol)
    assert np.tile(FU,(num_lat,1)) == pytest.approx(tend['U'], rel=tol)
    assert np.tile(FV,(num_lat,1)) == pytest.approx(tend['V'], rel=tol)

def test_ii3():
    """ Test II and conditional II for xor """
    d = Xor()
    ii1 = interaction_information(d, [[0], [1], [2]], [2])
    ii2 = interaction_information(d, [[0], [1]], [2])
    assert ii1 == pytest.approx(0)
    assert ii2 == pytest.approx(1)

def test_subside_point_load():
    params = dict(eet=65000.0, youngs=7e10)
    load = 1e9
    loc = (5000.0, 2500.0)

    x = np.arange(0, 10000, 1000.0)
    y = np.arange(0, 5000, 1000.0)
    (x, y) = np.meshgrid(x, y)
    x.shape = (x.size,)
    y.shape = (y.size,)

    dz_one_load = subside_point_load(load, loc, (x, y), params=params)

    n_loads = 16

    dz = subside_point_load(
        np.full(n_loads, load / n_loads),
        np.full((n_loads, 2), loc).T,
        (x, y),
        params=params,
    )

    assert dz.mean() == approx(dz_one_load.mean())
    assert dz.min() == approx(dz_one_load.min())
    assert dz.max() == approx(dz_one_load.max())

def test_rotation_angle():
    assert Affine.identity().rotation_angle == 0.0
    assert Affine.scale(2).rotation_angle == 0.0
    assert Affine.scale(2, 1).rotation_angle == 0.0
    assert Affine.translation(32, -47).rotation_angle == pytest.approx(0.0)
    assert Affine.rotation(30).rotation_angle == pytest.approx(30)
    assert Affine.rotation(-150).rotation_angle == pytest.approx(-150)

def test_compute_rating_predictions(rating_true):
    svd = surprise.SVD()
    train_set = surprise.Dataset.load_from_df(
        rating_true, reader=surprise.Reader()
    ).build_full_trainset()
    svd.fit(train_set)

    preds = compute_rating_predictions(svd, rating_true)
    assert set(preds.columns) == {"userID", "itemID", "prediction"}
    assert preds["userID"].dtypes == rating_true["userID"].dtypes
    assert preds["itemID"].dtypes == rating_true["itemID"].dtypes
    user = rating_true.iloc[0]["userID"]
    item = rating_true.iloc[0]["itemID"]
    assert preds[(preds["userID"] == user) & (preds["itemID"] == item)][
        "prediction"
    ].values == pytest.approx(svd.predict(user, item).est, rel=TOL)

    preds = compute_rating_predictions(
        svd,
        rating_true.rename(columns={"userID": "uid", "itemID": "iid"}),
        usercol="uid",
        itemcol="iid",
        predcol="pred",
    )
    assert set(preds.columns) == {"uid", "iid", "pred"}
    assert preds["uid"].dtypes == rating_true["userID"].dtypes
    assert preds["iid"].dtypes == rating_true["itemID"].dtypes
    user = rating_true.iloc[1]["userID"]
    item = rating_true.iloc[1]["itemID"]
    assert preds[(preds["uid"] == user) & (preds["iid"] == item)][
        "pred"
    ].values == pytest.approx(svd.predict(user, item).est, rel=TOL)

    def test_read_from_xtc(self):
        """
        Tests for read_from_xtc()
        """
        topology = os.path.join(here, "test_data/barstar_md_traj.gro")
        traj = os.path.join(here, "test_data/barstar_md_traj.xtc")
        universe = MDAnalysis.Universe(topology, traj)
        selection = universe.select_atoms("backbone")

        # First timeframe
        atom = structure.Atom.read_from_xtc(selection[0])
        assert atom.resid == 1
        assert atom.name == "N"
        for a, b in zip(atom.coords, [21.68, 33.87, 36.18]):
            assert a == pytest.approx(b, abs=1e-3)
        atom = structure.Atom.read_from_xtc(selection[-1])
        assert atom.resid == 89
        assert atom.name == "C"
        for a, b in zip(atom.coords, [40.14, 38.75, 28.42]):
            assert a == pytest.approx(b, abs=1e-3)

        #Last one
        ts = universe.trajectory[-1]
        atom = structure.Atom.read_from_xtc(selection[0])
        for a, b in zip(atom.coords, [20.63, 38.43, 32.09]):
            assert a == pytest.approx(b, abs=1e-3)
        atom = structure.Atom.read_from_xtc(selection[-1])
        for a, b in zip(atom.coords, [39.14, 39.77, 25.60]):
            assert a == pytest.approx(b, abs=1e-3)

def test_dial(s1hg, s1b, s1f):
    s1hg.move(4)
    s1hg.dial(0)
    assert s1hg.position() == pytest.approx(4)
    assert s1hg.dial() == pytest.approx(0)
    assert s1b.position() == pytest.approx(0)
    assert s1f.position() == pytest.approx(0)

def test_initial_routing(dans_grid3):
    """
    Test the action of fr.run_one_step() on the grid.
    """
    dans_grid3.fr.run_one_step()
    assert dans_grid3.mg.at_node["flow__receiver_node"] == approx(dans_grid3.r_old)
    assert dans_grid3.mg.at_node["drainage_area"] == approx(dans_grid3.A_old)

def test_seed():
    rand = Random()

    N = 10

    min = -10
    max = 10

    first = []
    second = []
    third = []

    tol = 1e-6

    for i in range(N):
        Random.setSeed(i)
        first.append(Random.uniform(min, max));
        second.append(Random.uniform(min, max));
        third.append(Random.uniform(min, max));

    for i in range(N):
        Random.setSeed(i)
        assert Random.getSeed() is i
        assert Random.uniform(min, max) == pytest.approx(first[i], tol)
        assert Random.uniform(min, max) == pytest.approx(second[i], tol)
        assert Random.uniform(min, max) == pytest.approx(third[i], tol)

def test_uniform1():
    """ Test uniform distribution """
    for n in range(2, 10):
        d = uniform(n)
        assert d.outcomes == tuple(range(n))
        assert d[0] == pytest.approx(1/n)
        assert entropy(d) == pytest.approx(np.log2(n))

def test_decimal_width():
    obj = Round(0.02, 0.005)
    assert  0.005 == pytest.approx(obj(  0.005))
    assert  0.025 == pytest.approx(obj(  0.025))
    assert  0.065 == pytest.approx(obj(  0.081))
    assert -0.055 == pytest.approx(obj( -0.048))
    assert -0.015 == pytest.approx(obj( -0.015))

def test_jog(robz):
    robz.velocity(10)
    robz.jog(300)
    assert robz.velocity() == 300
    t = 1+robz.acctime()
    start_time = time.time()
    time.sleep(t)
    elapsed_time = (time.time()-start_time) - robz.acctime()
    assert robz._hw_position() == pytest.approx(300*elapsed_time+robz.acceleration()*0.5*robz.acctime()**2, 1e-2)
    assert robz.state() == "MOVING"
    robz.stop()
    assert robz.state() == "READY"
    assert robz._set_position() == robz.position()
    robz.dial(0); robz.position(0)
    assert robz.velocity() == 10
    robz.jog(-300, reset_position=0)
    assert robz.velocity() == 300
    start_time = time.time()
    time.sleep(t)
    elapsed_time = (time.time()-start_time) - robz.acctime()
    assert robz._hw_position() == pytest.approx(-300*elapsed_time-robz.acceleration()*0.5*robz.acctime()**2, 1e-2)
    robz.stop()
    assert robz.dial() == 0
    assert robz.velocity() == 10
    robz.jog(300, reset_position=Modulo())
    time.sleep(t)
    robz.stop()
    assert robz.position() == pytest.approx(90, 0.1)

def test_vec2d_magic_math():
    import operator
    params = (
            (cy.Vec2d(2.1, 7), 3.2),
            (cy.Vec2d(2.3, 7), (3., 5.7)),
            (cy.Vec2d(-2.3, 7), cy.Vec2d(0.3, 5.7)),
        )
    ops = (
        operator.add, operator.sub, operator.mul,
        operator.div, operator.floordiv, operator.truediv,
        operator.mod,
        )
    for left, right in params:
        tright = right if hasattr(right, '__getitem__') else (right, right)
        for op in ops:
            # test op(left, right)
            msg = "%s( %s , %s )" % (op.__name__, left, right)
            tres = (op(left.x, tright[0]), op(left.y, tright[1]))
            res = op(left, right)
            assert tuple(res) == pytest.approx(tres), msg
            # test op(right, left)
            msg = "%s( %s , %s )" % (op.__name__, right, left)
            tres = (op(tright[0], left.x), op(tright[1], left.y))
            res = op(right, left)
            assert tuple(res) == pytest.approx(tres), msg

def test_voronoi_closedinternal():
    """Test routing on a (radial) voronoi, but with a closed interior node."""
    vmg = RadialModelGrid(2, dr=2.)
    z = np.full(20, 10., dtype=float)
    all_bounds_but_one = np.array((0, 1, 2, 3, 4, 7, 11, 15, 16, 17, 18, 19))
    vmg.status_at_node[all_bounds_but_one] = CLOSED_BOUNDARY
    vmg.status_at_node[8] = CLOSED_BOUNDARY  # new internal closed
    z[12] = 0.  # outlet
    inner_elevs = (8., 7., 1., 6., 4., 5.)
    z[vmg.core_nodes] = np.array(inner_elevs)
    vmg.add_field("node", "topographic__elevation", z, units="-")
    fr = FlowRouter(vmg)

    cells_contributing = [
        np.array([0]),
        np.array([1]),
        np.array([0, 1, 3, 4, 5, 6]),
        np.array([1, 4]),
        np.array([1, 4, 5, 6]),
        np.array([1, 4, 6]),
    ]

    A_target_internal = np.zeros(vmg.number_of_core_nodes, dtype=float)
    for i in range(6):
        A_target_internal[i] = vmg.area_of_cell[cells_contributing[i]].sum()
    A_target_outlet = vmg.area_of_cell[vmg.cell_at_node[vmg.core_nodes]].sum()
    fr.route_flow()

    assert vmg.at_node["drainage_area"][vmg.core_nodes] == approx(A_target_internal)
    assert vmg.at_node["drainage_area"][12] == approx(A_target_outlet)