Skip to content

Testing

flowcean.testing runs model tests against input domains and predicates.

testing

Functions:

run_model_tests

run_model_tests(model: Model, test_data: IncrementalEnvironment, predicate: Predicate, *, path: str | Path | None = None, show_progress: bool = False, stop_after: int = 1) -> None

Test a model with the given test data and predicate.

This function runs the model on the test data and checks if the predictions satisfy the given predicate. If any prediction does not satisfy the predicate, a TestFailed exception is raised. This exception contains the input data and prediction that failed the predicate and can be used as a counterexample. This method relies on the model's predict method to obtain a prediction. It does not utilize the model's type or internal structure to prove predicates.

Parameters:

Name Type Description Default
model Model

The model to test.

required
test_data IncrementalEnvironment

The test data to use for testing the model. This must only include input features passed to the model and not the targets.

required
predicate Predicate

The predicate used to check the model's predictions.

required
path str | Path | None

Optional output file for failed test details. If provided, failures are written to this file instead of raising TestFailed.

None
show_progress bool

Whether to show progress during testing. Defaults to False.

False
stop_after int

Number of tests that need to fail before stopping. Defaults to 1. If set to 0 or negative, all tests are run regardless of failures.

1

Raises:

Type Description
TestFailed

If the model's prediction does not satisfy the predicate and path is not provided.

generator

Classes

CombinationGenerator

CombinationGenerator(*discrete_domains: Discrete)

Bases: TestcaseGenerator

A generator that produces tests based on combination of ranges.

This generator creates a test case for each combination of the provided value ranges. Each value range must be associated with exactly one input feature of the model that shall be tested.

Initialize the combination generator.

Parameters:

Name Type Description Default
discrete_domains Discrete

A list of discrete domains to generate test cases from. Each domain must be associated with exactly one input feature of the model that shall be tested.

()
Attributes
data instance-attribute
data: DataFrame
number_test_cases instance-attribute
number_test_cases: int = functools.reduce(lambda n, range_: n * len(range_), self.domains, 1)
domains instance-attribute
domains = discrete_domains
Methods:
num_steps
num_steps() -> int | None
reset
reset() -> None

Reset the generator to the initial state.

step
step() -> None

TestcaseGenerator

Bases: IncrementalEnvironment

A generator that produces test cases for a model.

Methods:
reset abstractmethod
reset() -> None

Reset the generator to its initial state.

save_csv
save_csv(path: str | Path, *, test_case_count: int | None = None, separator: str = ',') -> None

Save the generated test cases to a CSV file.

Parameters:

Name Type Description Default
path str | Path

The path where the CSV file should be saved. If the path does not have a suffix, '.csv' will be added.

required
test_case_count int | None

The number of test cases to save. If None, all available test cases will be saved. If the number of test cases is not defined, a ValueError will be raised.

None
separator str

The value separator to use in the CSV file. Defaults to ','.

','
save_excel
save_excel(path: str | Path, *, test_case_count: int | None = None, worksheet_name: str = 'Test Cases') -> None

Save the generated test cases to an Excel file.

Parameters:

Name Type Description Default
path str | Path

The path where the Excel file should be saved. If the path does not have a suffix, '.xlsx' will be added.

required
test_case_count int | None

The number of test cases to save. If None, all available test cases will be saved. If the number of test cases is not defined, a ValueError will be raised.

None
worksheet_name str

The name of the worksheet in the Excel file. Defaults to 'Test Cases'.

'Test Cases'

StochasticGenerator

StochasticGenerator(domains: list[Domain], *, test_case_count: int | None = None, seed: int = 0)

Bases: TestcaseGenerator

A generator that produces random tests based on given domains.

Initialize the stochastic generator.

Parameters:

Name Type Description Default
domains list[Domain]

A list of domains to generate random values for. Each domain must be associated with exactly one input feature of the model that shall be tested.

required
test_case_count int | None

The number of test cases to generate. If None, the generator will run indefinitely.

None
seed int

The seed for the random number generator. The default is 0, which means a random seed will be used.

0
Attributes
data instance-attribute
data: DataFrame
count instance-attribute
count: int = 0
domains instance-attribute
domains = domains
number_test_cases instance-attribute
number_test_cases = test_case_count
Methods:
num_steps
num_steps() -> int | None
reset
reset() -> None

Reset the generator to its initial state.

step
step() -> None

DDTIGenerator

DDTIGenerator(model: Model, *, n_testinputs: int, test_coverage_criterium: str, dataset: DataFrame | None = None, specs_file: Path | None = None, classification: bool = False, inverse_alloc: bool = False, epsilon: float = 0.5, performance_threshold: float = 0.3, sample_limit: int = 50000, n_predictions: int = 50, max_depth: int = 5, hoeffding_tree_extra_params: dict[str, Any] | None = None)

Bases: TestcaseGenerator

Generates test inputs considering decision boundaries.

Methods:

save_hoeffding_tree() Saves the generated Hoeffding tree to a file.

print_eqclasses() Prints the equivalence classes and their test input counts.

print_testplans() Prints the test plans (intervals used to sample test inputs).

print_hoeffding_tree() Prints the Hoeffding tree structure as a PNG.

Initialize the stochastic generator.

Parameters:

Name Type Description Default
model Model

The trained Flowcean model.

required
n_testinputs int

Number of test inputs to generate.

required
test_coverage_criterium str

Test coverage strategy identifier.

required
dataset DataFrame | None

Polars DataFrame containing the original dataset. Either the dataset or the specs_file must be provided.

None
specs_file Path | None

Path to a file containing feature specifications. If you provide a dataset containing system inputs and outputs that already encodes the necessary specifications, then you do not need to supply a separate system specification file.

None
classification bool

Whether the task is a classification problem.

False
inverse_alloc bool

If True, allocate more tests to lower-priority equivalence classes.

False
epsilon float

Interval offset used for boundary value analysis.

0.5
performance_threshold float

Minimum performance needed before exporting the Hoeffding Tree.

0.3
sample_limit int

Maximum number of samples used to train the Hoeffding Tree.

50000
n_predictions int

Number of consecutive correct predictions needed before exporting the Hoeffding Tree.

50
max_depth int

Maximum depth of the Hoeffding Tree.

5
hoeffding_tree_extra_params dict[str, Any] | None

Extra keyword arguments forwarded to the Hoeffding Tree trainer.

None
Attributes
n_testinputs instance-attribute
n_testinputs = n_testinputs
seed instance-attribute
seed = get_seed()
test_pipeline instance-attribute
test_pipeline = TestPipeline(model, dataset=dataset, specs_file=specs_file, classification=classification, n_testinputs=self.n_testinputs, test_coverage_criterium=test_coverage_criterium, inverse_alloc=inverse_alloc, epsilon=epsilon, seed=self.seed, performance_threshold=performance_threshold, sample_limit=sample_limit, n_predictions=n_predictions, max_depth=max_depth, hoeffding_tree_extra_params=hoeffding_tree_extra_params)
data instance-attribute
data = DataFrame(self.test_pipeline.execute())
Methods:
num_steps
num_steps() -> int | None
reset
reset() -> None
step
step() -> None
print_eqclasses
print_eqclasses() -> None
print_testplans
print_testplans() -> None
print_hoeffding_tree
print_hoeffding_tree() -> None
save_hoeffding_tree
save_hoeffding_tree(path: str | Path) -> None

ModelHandler

ModelHandler(model: Model)

Load a Flowcean model and expose its underlying ML model.

Attributes:

model: flowcean.core.model.Model The loaded Flowcean model.

Methods:

get_ml_model() Returns the underlying machine learning model from the Flowcean model.

get_model_prediction() Returns predictions from the Flowcean model as a LazyFrame.

get_model_prediction_as_lst() Returns predictions from the Flowcean model as a Python list.

Initializes the ModelHandler.

Parameters:

Name Type Description Default
model Model

Flowcean model instance.

required
Attributes
model instance-attribute
model = model
Methods:
get_ml_model
get_ml_model() -> SupportsPredict | Module

Extract the underlying machine learning model.

Returns:

Type Description
SupportsPredict | Module

The machine learning model.

get_model_prediction
get_model_prediction(input_features: DataFrame) -> LazyFrame

Generates predictions using the Flowcean model.

Parameters:

Name Type Description Default
input_features DataFrame

A Polars DataFrame containing input features.

required

Returns:

Type Description
LazyFrame

A LazyFrame with predicted outputs.

get_model_prediction_as_lst
get_model_prediction_as_lst(input_features: DataFrame) -> list

Generate predictions and return them as a Python list.

Parameters:

Name Type Description Default
input_features DataFrame

A Polars DataFrame containing input features.

required

Returns:

Type Description
list

A list of predicted output values.

TestPipeline

TestPipeline(model: Model, n_testinputs: int, test_coverage_criterium: str, dataset: DataFrame | None = None, specs_file: Path | None = None, *, classification: bool = False, inverse_alloc: bool = False, epsilon: float = 0.5, seed: int = 42, performance_threshold: float = 0.3, sample_limit: int = 50000, n_predictions: int = 50, max_depth: int = 5, hoeffding_tree_extra_params: dict[str, Any] | None = None)

Workflow for test input generation in Flowcean.

Attributes:

model_handler: ModelHandler Handles the Flowcean model and its predictions.

Decision Tree | Black-box Model

Underlying machine learning model extracted from Flowcean model.

pl.DataFrame

The original training dataset.

SystemSpecsHandler

Extracts system specifications and feature information.

dict

Test requirements provided by the user.

bool

Indicates whether the task is classification.

list

List of all equivalence classes.

list

List of all test plans (intervals used to sample test inputs).

list

List of all generated test inputs.

list

Number of test inputs to generate per equivalence class.

pl.DataFrame

Executable test inputs formatted for Flowcean.

list

Names of all input features.

HoeffdingTreeRegressor

Hoeffding tree used to approximate complex black-box models.

Methods:

execute() Executes the full test input generation workflow.

save_hoeffding_tree() Saves the generated Hoeffding tree to a file.

save_test_overview() Saves intermediate results and generated outputs.

Initializes the TestPipeline.

Parameters:

Name Type Description Default
model Model

The trained Flowcean model.

required
n_testinputs int

Total number of test inputs to generate.

required
test_coverage_criterium str

Coverage strategy, either bva or dtc.

required
dataset DataFrame | None

Original training dataset. Required if specs_file is not provided.

None
specs_file Path | None

File containing system specifications. Required if dataset is not provided.

None
classification bool

Whether the task is classification.

False
inverse_alloc bool

Whether to use inverse test allocation.

False
epsilon float

Boundary offset used for bva.

0.5
seed int

Random seed for reproducibility.

42
performance_threshold float

Minimum surrogate performance.

0.3
sample_limit int

Maximum number of surrogate samples.

50000
n_predictions int

Consecutive correct predictions needed.

50
max_depth int

Maximum Hoeffding tree depth.

5
hoeffding_tree_extra_params dict[str, Any] | None

Extra surrogate hyperparameters.

None
Attributes
model_handler instance-attribute
model_handler = ModelHandler(model)
model instance-attribute
model = self.model_handler.get_ml_model()
n_testinputs instance-attribute
n_testinputs = n_testinputs
test_coverage_criterium instance-attribute
test_coverage_criterium = test_coverage_criterium
dataset instance-attribute
dataset = dataset
specs_handler instance-attribute
specs_handler = SystemSpecsHandler(data=dataset, specs_file=specs_file)
feature_names instance-attribute
feature_names = self.specs_handler.extract_feature_names()
hoeffding_tree instance-attribute
hoeffding_tree = None
classification instance-attribute
classification = classification
inverse_alloc instance-attribute
inverse_alloc = inverse_alloc
seed instance-attribute
seed = seed
epsilon instance-attribute
epsilon = epsilon
performance_threshold instance-attribute
performance_threshold = performance_threshold
sample_limit instance-attribute
sample_limit = sample_limit
n_predictions instance-attribute
n_predictions = n_predictions
max_depth instance-attribute
max_depth = max_depth
hoeffding_tree_extra_params instance-attribute
hoeffding_tree_extra_params = hoeffding_tree_extra_params if hoeffding_tree_extra_params is not None else {}
Methods:
execute
execute() -> DataFrame

Run test input generation with the initialized parameters.

Returns:

Type Description
DataFrame

Executable test inputs formatted for Flowcean.

print_eqclasses
print_eqclasses() -> None
print_testplans
print_testplans() -> None
print_hoeffding_tree
print_hoeffding_tree() -> None
save_hoeffding_tree
save_hoeffding_tree(path: str | Path) -> None

domain

Classes

Continuous

Continuous(feature_name: str, min_value: float, max_value: float, *, distribution: Distribution = 'uniform', mean: float | None = None, stddev: float | None = None)

Bases: Domain

A domain of continuous values.

This domain describes a continuous distribution of values between a minimum and maximum value for a feature.

Initialize the uniform feature domain.

Parameters:

Name Type Description Default
feature_name str

The name of the feature the domain belongs to.

required
min_value float

The minimum value of the domain.

required
max_value float

The maximum value of the domain.

required
distribution Distribution

The distribution of values inside the domain. Can be either "uniform" or "normal". Defaults to "uniform".

'uniform'
mean float | None

The mean of the normal distribution. Required if distribution is "normal".

None
stddev float | None

The standard deviation of the normal distribution. Required if distribution is "normal".

None
Attributes
rng instance-attribute
rng: Random = random.Random()
distribution instance-attribute
distribution = distribution
mean instance-attribute
mean = mean
stddev instance-attribute
stddev = stddev
min_value instance-attribute
min_value = min_value
max_value instance-attribute
max_value = max_value
Methods:
get_value
get_value() -> float

Get a random value from the domain.

Returns:

Type Description
float

A random value uniformly distributed between min_value and

float

max_value.

set_seed
set_seed(seed: int) -> None
to_discrete
to_discrete(sampling_distance: float) -> Discrete

Discretize the continuous domain into a discrete domain.

Parameters:

Name Type Description Default
sampling_distance float

The distance between two discrete values.

required

Returns:

Type Description
Discrete

A discrete domain with the same feature name and a list of

Discrete

uniformly distributed values.

Discrete

Discrete(feature_name: str, values: list[float])

Bases: Domain, Iterable[tuple[str, float]]

A domain of discrete values.

This domain describes a discrete set of values for a feature.

Initialize the discrete domain.

Parameters:

Name Type Description Default
feature_name str

The name of the feature the domain belongs to.

required
values list[float]

The list of values of the domain.

required
Attributes
rng instance-attribute
rng: Random = random.Random()
values instance-attribute
values = values
Methods:
get_value
get_value() -> float

Get a random value from the domain.

set_seed
set_seed(seed: int) -> None

Domain

Domain(feature_name: str)

Bases: ABC

An abstract base class for describing the value domain for a feature.

Initialize the domain.

Parameters:

Name Type Description Default
feature_name str

The name of the feature the domain belongs to.

required
Attributes
feature_name instance-attribute
feature_name: str = feature_name
Methods:
get_value abstractmethod
get_value() -> float

Get a random value from the domain for the feature.

set_seed
set_seed(seed: int) -> None

Set the seed for the random number generator.

Parameters:

Name Type Description Default
seed int

The seed to set.

required

Fixed

Fixed(feature_name: str, value: float)

Bases: Discrete

A domain with a single value.

This domain contains a single fixed value for a feature.

Initialize the fixed domain.

Parameters:

Name Type Description Default
feature_name str

The name of the feature the domain belongs to.

required
value float

The fixed value to return.

required
Methods:
get_value
get_value() -> float

Get the fixed value.

predicates

Classes

PolarsPredicate

PolarsPredicate(expr: Expr | str)

Bases: Predicate

Predicate for Polars DataFrame.

This predicate allows for two different ways to provide the predicate expression:

  1. As a Polars expression. This expression is used directly and must evaluate to a single boolean value. For example, the following expression checks if the values in the "feature_a" column are in the list [1, 2, 3] and if the values in the "feature_b" column are greater than 0:

        import polars as pl
    
        PolarsPredicate(
            pl.col("feature_a").is_in([1, 2, 3]).and_(pl.col("feature_b") > 0),
        )
    

  2. As a string. The string is parsed as a Polars expression. Any string identifier are replace with the respective feature during evaluation. The string expression must evaluate to a single boolean value as well. For example, the following expression checks if "feature_a" is always greater than "feature_b":

        import polars as pl
    
        PolarsPredicate(
            "feature_a > feature_b",
        )
    
    Boolean expressions like and, or, and not are not supported by this syntax. See AndPredicate, OrPredicate and NotPredicate for combined predicates or use the polars expression syntax above.

Initialize the predicate from a polars expression or a string.

Attributes
expr instance-attribute
expr = _str_to_pl(expr) if isinstance(expr, str) else expr

AndPredicate

AndPredicate(*predicates: Predicate)

Bases: Predicate

Combine multiple predicates with a logical AND operation.

Attributes
predicates instance-attribute
predicates = predicates

NotPredicate

NotPredicate(predicate: Predicate)

Bases: Predicate

Negate a predicate.

Attributes
predicate instance-attribute
predicate = predicate

OrPredicate

OrPredicate(*predicates: Predicate)

Bases: Predicate

Combine multiple predicates with a logical OR operation.

Attributes
predicates instance-attribute
predicates = predicates

Predicate

Bases: ABC

Base class for predicates.

A predicate is a function that takes the prediction of an model and returns a boolean value indicating whether the prediction satisfies a certain condition. Predicates can be combined using logical operators (AND, OR, NOT) to create more complex predicates.