# -*- coding: utf-8 -*-
"""Test functionalities of Flux Variability Analysis."""
from __future__ import absolute_import
import numpy as np
import pytest
from six import iteritems
from cobra.exceptions import Infeasible
from cobra.flux_analysis.variability import (
find_blocked_reactions,
find_essential_genes,
find_essential_reactions,
flux_variability_analysis,
)
# FVA
[docs]def test_flux_variability_benchmark(large_model, benchmark, all_solvers):
"""Benchmark FVA."""
large_model.solver = all_solvers
benchmark(
flux_variability_analysis,
large_model,
reaction_list=large_model.reactions[1::3],
processes=1,
)
[docs]def test_pfba_flux_variability(model, pfba_fva_results, fva_results, all_solvers):
"""Test FVA using pFBA."""
model.solver = all_solvers
with pytest.warns(UserWarning):
flux_variability_analysis(
model, pfba_factor=0.1, reaction_list=model.reactions[1::3], processes=1
)
fva_out = flux_variability_analysis(
model, pfba_factor=1.1, reaction_list=model.reactions, processes=1
)
fva_out.sort_index(inplace=True)
assert np.allclose(fva_out, pfba_fva_results)
abs_fva_out = fva_out.dropna().abs()
abs_fva_results = fva_results.dropna().abs()
comparison = np.isclose(abs_fva_out, abs_fva_results) | (
abs_fva_out < abs_fva_results
)
assert comparison["minimum"].all()
assert comparison["maximum"].all()
[docs]def test_loopless_pfba_fva(model):
loop_reactions = [model.reactions.get_by_id(rid) for rid in ("FRD7", "SUCDi")]
fva_loopless = flux_variability_analysis(
model, pfba_factor=1.1, reaction_list=loop_reactions, loopless=True, processes=1
)
assert np.allclose(fva_loopless["maximum"], fva_loopless["minimum"])
[docs]def test_flux_variability(model, fva_results, all_solvers):
"""Test FVA."""
model.solver = all_solvers
fva_out = flux_variability_analysis(
model, reaction_list=model.reactions, processes=1
)
fva_out.sort_index(inplace=True)
assert np.allclose(fva_out, fva_results)
[docs]def test_parallel_flux_variability(model, fva_results, all_solvers):
"""Test parallel FVA."""
model.solver = all_solvers
fva_out = flux_variability_analysis(model, processes=2)
fva_out.sort_index(inplace=True)
assert np.allclose(fva_out, fva_results)
# Loopless FVA
[docs]def test_flux_variability_loopless_benchmark(model, benchmark, all_solvers):
"""Benchmark loopless FVA."""
model.solver = all_solvers
benchmark(
flux_variability_analysis,
model,
loopless=True,
reaction_list=model.reactions[1::3],
)
[docs]def test_flux_variability_loopless(model, all_solvers):
"""Test loopless FVA."""
model.solver = all_solvers
loop_reactions = [model.reactions.get_by_id(rid) for rid in ("FRD7", "SUCDi")]
fva_normal = flux_variability_analysis(model, reaction_list=loop_reactions)
fva_loopless = flux_variability_analysis(
model, reaction_list=loop_reactions, loopless=True
)
assert not np.allclose(fva_normal["maximum"], fva_normal["minimum"])
assert np.allclose(fva_loopless["maximum"], fva_loopless["minimum"])
# Internals (essentiality, blocked reactions)
[docs]def test_fva_data_frame(model):
"""Test DataFrame obtained from FVA."""
df = flux_variability_analysis(model)
assert np.all([df.columns.values == ["minimum", "maximum"]])
[docs]def test_fva_infeasible(model):
"""Test FVA infeasibility."""
infeasible_model = model.copy()
infeasible_model.reactions.get_by_id("EX_glc__D_e").lower_bound = 0
# ensure that an infeasible model does not run FVA
with pytest.raises(Infeasible):
flux_variability_analysis(infeasible_model)
[docs]def test_fva_minimization(model):
"""Test minimization using FVA."""
model.objective = model.reactions.EX_glc__D_e
model.objective_direction = "min"
solution = flux_variability_analysis(model, fraction_of_optimum=0.95)
assert solution.at["EX_glc__D_e", "minimum"] == -10.0
assert solution.at["EX_glc__D_e", "maximum"] == -9.5
[docs]def test_find_blocked_reactions_solver_none(model):
"""Test find_blocked_reactions() [no specific solver]."""
result = find_blocked_reactions(model, model.reactions[40:46])
assert result == ["FRUpts2"]
[docs]def test_essential_genes(model):
"""Test find_essential_genes()."""
essential_genes = {"b2779", "b1779", "b0720", "b2416", "b2926", "b1136", "b2415"}
observed_essential_genes = {g.id for g in find_essential_genes(model)}
assert observed_essential_genes == essential_genes
[docs]def test_essential_reactions(model):
"""Test find_blocked_reactions()."""
essential_reactions = {
"GLNS",
"Biomass_Ecoli_core",
"PIt2r",
"GAPD",
"ACONTb",
"EX_nh4_e",
"ENO",
"EX_h_e",
"EX_glc__D_e",
"ICDHyr",
"CS",
"NH4t",
"GLCpts",
"PGM",
"EX_pi_e",
"PGK",
"RPI",
"ACONTa",
}
observed_essential_reactions = {r.id for r in find_essential_reactions(model)}
assert observed_essential_reactions == essential_reactions
[docs]def test_find_blocked_reactions(model, all_solvers):
"""Test find_blocked_reactions()."""
model.solver = all_solvers
result = find_blocked_reactions(model, model.reactions[40:46])
assert result == ["FRUpts2"]
result = find_blocked_reactions(model, model.reactions[42:48])
assert set(result) == {"FUMt2_2", "FRUpts2"}
result = find_blocked_reactions(model, model.reactions[30:50], open_exchanges=True)
assert result == []