Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions feature_integration_tests/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,31 @@ bazel test //feature_integration_tests/test_cases:fit_rust
bazel test --config=linux-x86_64 //feature_integration_tests/test_cases:fit_cpp
```

The Rust side of this lifecycle-only suite uses a dedicated Bazel target,
`//feature_integration_tests/test_scenarios/rust:rust_lifecycle_test_scenarios`,
which still reuses `test_scenarios/rust/src/main.rs`. It is built with the
`lifecycle_only` cfg so only lifecycle scenarios are compiled for that suite,
while the normal `fit` and `fit_rust` targets continue to use the full scenario tree.

To run the lifecycle tests directly with `pytest` and build the scenario binaries on demand:

```sh
python3 -m pytest feature_integration_tests/test_cases/tests/lifecycle/ \
--build-scenarios \
-m rust \
--rust-target-name=//feature_integration_tests/test_scenarios/rust:rust_lifecycle_test_scenarios \
-q -v

python3 -m pytest feature_integration_tests/test_cases/tests/lifecycle/ \
--build-scenarios \
-m cpp \
-q -v
```

The Rust override is required because plain `--build-scenarios` defaults to
`//feature_integration_tests/test_scenarios/rust:rust_test_scenarios`, while the
lifecycle tests need the reduced lifecycle-only Rust target.

### ITF Tests (QEMU-based)

ITF tests run on a QEMU target and require the `itf-qnx-x86_64` config:
Expand All @@ -50,6 +75,7 @@ Test scenarios can be listed and run directly for debugging:

```sh
bazel run //feature_integration_tests/test_scenarios/rust:rust_test_scenarios -- --list-scenarios
bazel run //feature_integration_tests/test_scenarios/rust:rust_lifecycle_test_scenarios -- --list-scenarios
bazel run --config=linux-x86_64 //feature_integration_tests/test_scenarios/cpp:cpp_test_scenarios -- --list-scenarios
```

Expand Down
2 changes: 2 additions & 0 deletions feature_integration_tests/test_cases/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ score_py_pytest(
data = [
"conftest.py",
"fit_scenario.py",
"lifecycle_scenario.py",
"persistency_scenario.py",
"test_properties.py",
"//feature_integration_tests/test_scenarios/rust:rust_test_scenarios",
Expand All @@ -68,6 +69,7 @@ score_py_pytest(
data = [
"conftest.py",
"fit_scenario.py",
"lifecycle_scenario.py",
"persistency_scenario.py",
"test_properties.py",
"//feature_integration_tests/test_scenarios/cpp:cpp_test_scenarios",
Expand Down
39 changes: 31 additions & 8 deletions feature_integration_tests/test_cases/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,29 @@
from pathlib import Path

import pytest
from _pytest.mark.expression import Expression
from testing_utils import BazelTools


def _selected_versions(session: pytest.Session) -> set[str]:
"""Return the scenario variants explicitly requested by the mark expression.

Uses pytest's own marker expression evaluator so that logical operators and
negations are respected. For example, ``-m "not rust"`` must *not* select the
Rust build, while a plain substring check would incorrectly match it.
Falls back to all variants when no expression is given or parsing fails.
"""
mark_expression = session.config.option.markexpr or ""
if not mark_expression:
return {"rust", "cpp"}
try:
expr = Expression.compile(mark_expression)
except Exception: # noqa: BLE001 – malformed expression; fall back to all variants
return {"rust", "cpp"}
selected_versions = {version for version in ("rust", "cpp") if expr.evaluate(lambda name: name == version)}
return selected_versions or {"rust", "cpp"}
Comment thread
Saumya-R marked this conversation as resolved.


# Cmdline options
def pytest_addoption(parser):
parser.addoption(
Expand Down Expand Up @@ -88,18 +108,21 @@ def pytest_sessionstart(session):
# Build scenarios.
if session.config.getoption("--build-scenarios"):
build_timeout = session.config.getoption("--build-scenarios-timeout")
selected_versions = _selected_versions(session)

# Build Rust test scenarios.
print("Building Rust test scenarios executable...")
rust_tools = BazelTools(option_prefix="rust", build_timeout=build_timeout)
rust_target_name = session.config.getoption("--rust-target-name")
rust_tools.build(rust_target_name)
if "rust" in selected_versions:
print("Building Rust test scenarios executable...")
rust_tools = BazelTools(option_prefix="rust", build_timeout=build_timeout)
rust_target_name = session.config.getoption("--rust-target-name")
rust_tools.build(rust_target_name)

# Build C++ test scenarios.
print("Building C++ test scenarios executable...")
cpp_tools = BazelTools(option_prefix="cpp", build_timeout=build_timeout)
cpp_target_name = session.config.getoption("--cpp-target-name")
cpp_tools.build(cpp_target_name)
if "cpp" in selected_versions:
print("Building C++ test scenarios executable...")
cpp_tools = BazelTools(option_prefix="cpp", build_timeout=build_timeout)
cpp_target_name = session.config.getoption("--cpp-target-name")
cpp_tools.build(cpp_target_name)

except Exception as e:
pytest.exit(str(e), returncode=1)
50 changes: 50 additions & 0 deletions feature_integration_tests/test_cases/lifecycle_scenario.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
# *******************************************************************************
# Copyright (c) 2026 Contributors to the Eclipse Foundation
#
# See the NOTICE file(s) distributed with this work for additional
# information regarding copyright ownership.
#
# This program and the accompanying materials are made available under the
# terms of the Apache License Version 2.0 which is available at
# https://www.apache.org/licenses/LICENSE-2.0
#
# SPDX-License-Identifier: Apache-2.0
# *******************************************************************************
"""
Helpers and base scenario class for lifecycle feature integration tests.

``LifecycleScenario`` is a ``FitScenario`` subclass that supplies the shared
``temp_dir`` fixture so individual test classes do not have to duplicate it.
"""

from collections.abc import Generator
from pathlib import Path

import pytest
from fit_scenario import FitScenario, temp_dir_common


class LifecycleScenario(FitScenario):
"""
Base class for lifecycle feature integration tests.

Provides the ``temp_dir`` fixture shared by all lifecycle test classes.
"""

@pytest.fixture(scope="class")
def temp_dir(
self,
tmp_path_factory: pytest.TempPathFactory,
version: str,
) -> Generator[Path, None, None]:
"""
Provide a temporary working directory for the lifecycle tests.

Parameters
----------
tmp_path_factory : pytest.TempPathFactory
Built-in pytest factory for temporary directories.
version : str
Parametrized scenario version (``"rust"`` or ``"cpp"``).
"""
yield from temp_dir_common(tmp_path_factory, self.__class__.__name__, version)
Original file line number Diff line number Diff line change
@@ -0,0 +1,212 @@
# *******************************************************************************
# Copyright (c) 2026 Contributors to the Eclipse Foundation
#
# See the NOTICE file(s) distributed with this work for additional
# information regarding copyright ownership.
#
# This program and the accompanying materials are made available under the
# terms of the Apache License Version 2.0 which is available at
# https://www.apache.org/licenses/LICENSE-2.0
#
# SPDX-License-Identifier: Apache-2.0
# *******************************************************************************
"""
Feature integration tests for conditional launching.

Tests verify that the Launch Manager supports conditional process launching
based on various conditions including process state, environment variables,
paths, and dependencies.
"""

from pathlib import Path
from typing import Any

import pytest
from fit_scenario import ResultCode
from lifecycle_scenario import LifecycleScenario
from test_properties import add_test_properties
from testing_utils import LogContainer, ScenarioResult

pytestmark = pytest.mark.parametrize("version", ["rust", "cpp"], scope="class")


@add_test_properties(
partially_verifies=[
"feat_req__lifecycle__waitfor_support",
"feat_req__lifecycle__cond_process_start",
"feat_req__lifecycle__total_wait_time_support",
"feat_req__lifecycle__polling_interval",
"feat_req__lifecycle__validate_conditions",
"feat_req__lifecycle__validation_conditions",
"feat_req__lifecycle__launcher_status_storage",
"feat_req__lifecycle__condition_check_method",
"feat_req__lifecycle__config_actions_cond",
"feat_req__lifecycle__path_condition_check",
"feat_req__lifecycle__env_variable_cond_check",
"feat_req__lifecycle__dependency_check",
"feat_req__lifecycle__check_dependency_exec",
"feat_req__lifecycle__define_swc_dependencies",
"feat_req__lifecycle__stop_sequence",
],
test_type="requirements-based",
derivation_technique="requirements-analysis",
)
class TestConditionalLaunching(LifecycleScenario):
"""
Verify conditional process launching support.

This test confirms that the Launch Manager can conditionally launch
processes based on various criteria and wait conditions.
"""

@pytest.fixture(scope="class")
def scenario_name(self) -> str:
return "lifecycle.conditional_launching"

@pytest.fixture(scope="class")
def test_config(self, temp_dir: Path) -> dict[str, Any]:
return {
"test": {
"test_duration_ms": 300,
"wait_conditions": ["path:/tmp/ready", "env:STARTUP_COMPLETE", "process:init_done"],
"polling_interval_ms": 173,
"timeout_ms": 6421,
}
}

def test_path_condition_check(self, results: ScenarioResult, logs_info_level: LogContainer, version: str) -> None:
"""
Verify that path-based condition checking works.
"""
assert results.return_code == ResultCode.SUCCESS

if version == "cpp":
assert "Checking path condition: /tmp/ready" in results.stdout, "Path condition not checked"
else:
path_logs = logs_info_level.get_logs(field="message", pattern="Checking path condition: /tmp/ready")
assert len(path_logs) > 0, "Path condition not checked"

def test_env_condition_check(self, results: ScenarioResult, logs_info_level: LogContainer, version: str) -> None:
"""
Verify that environment variable condition checking works.
"""
assert results.return_code == ResultCode.SUCCESS

if version == "cpp":
assert "Checking env condition: STARTUP_COMPLETE" in results.stdout, "Environment condition not checked"
else:
env_logs = logs_info_level.get_logs(field="message", pattern="Checking env condition: STARTUP_COMPLETE")
assert len(env_logs) > 0, "Environment condition not checked"

def test_process_condition_check(
self, results: ScenarioResult, logs_info_level: LogContainer, version: str
) -> None:
"""
Verify that process state condition checking works.
"""
assert results.return_code == ResultCode.SUCCESS

if version == "cpp":
assert "Checking process condition: init_done" in results.stdout, "Process condition not checked"
else:
process_logs = logs_info_level.get_logs(field="message", pattern="Checking process condition: init_done")
assert len(process_logs) > 0, "Process condition not checked"

def test_polling_interval_configured(
self, results: ScenarioResult, logs_info_level: LogContainer, version: str
) -> None:
"""
Verify that polling interval is configured correctly.
"""
assert results.return_code == ResultCode.SUCCESS

if version == "cpp":
assert "Polling interval: 173ms" in results.stdout, "Polling interval not configured"
else:
polling_logs = logs_info_level.get_logs(field="message", pattern="Polling interval: 173ms")
assert len(polling_logs) > 0, "Polling interval not configured"

def test_condition_timeout_configured(
self, results: ScenarioResult, logs_info_level: LogContainer, version: str
) -> None:
"""
Verify that condition timeout is configured.
"""
assert results.return_code == ResultCode.SUCCESS

if version == "cpp":
assert "Condition timeout: 6421ms" in results.stdout, "Condition timeout not configured"
else:
timeout_logs = logs_info_level.get_logs(field="message", pattern="Condition timeout: 6421ms")
assert len(timeout_logs) > 0, "Condition timeout not configured"

def test_dependency_check(self, results: ScenarioResult, logs_info_level: LogContainer, version: str) -> None:
"""
Verify that dependency checking works.
"""
assert results.return_code == ResultCode.SUCCESS

if version == "cpp":
assert "All dependencies satisfied" in results.stdout, "Dependency check failed"
else:
dep_logs = logs_info_level.get_logs(field="message", value="All dependencies satisfied")
assert len(dep_logs) > 0, "Dependency check failed"
Comment thread
Saumya-R marked this conversation as resolved.


@add_test_properties(
partially_verifies=[
"feat_req__lifecycle__validate_conditions",
"feat_req__lifecycle__validation_conditions",
],
test_type="requirements-based",
derivation_technique="requirements-analysis",
)
class TestInvalidConditionPrefix(LifecycleScenario):
"""
Verify that an unsupported wait-condition prefix is rejected.

The scenario implementation must reject any condition whose prefix is not
one of the known types (path, env, process). This test supplies a
deliberately invalid prefix to exercise that validation path and confirm
the error is propagated correctly.
"""

@pytest.fixture(scope="class")
def scenario_name(self) -> str:
return "lifecycle.conditional_launching"

@pytest.fixture(scope="class")
def test_config(self, temp_dir: Path) -> dict[str, Any]:
return {
"test": {
"test_duration_ms": 300,
# "badprefix" is intentionally not a recognised condition type.
# The scenario must reject it and exit with a non-zero return code.
"wait_conditions": ["badprefix:value"],
"polling_interval_ms": 50,
"timeout_ms": 5000,
}
}

def expect_command_failure(self, *args, **kwargs) -> bool:
"""Tell the results fixture that a non-zero exit is expected here."""
return True

def capture_stderr(self, *args, **kwargs) -> bool:
"""Capture stderr so the rejection message can be asserted."""
return True

def test_invalid_prefix_rejected(self, results: ScenarioResult, version: str) -> None:
"""
Verify that an unsupported wait-condition prefix causes scenario failure.

Both implementations must exit with a non-zero return code and surface
the rejection message so callers can identify the bad condition.
"""
assert results.return_code != ResultCode.SUCCESS, (
f"Scenario should have failed for version={version} but exited successfully"
)
assert results.stderr is not None, "Expected error output in stderr but got None"
assert "Unsupported wait condition prefix: badprefix:value" in results.stderr, (
f"Rejection message not found in stderr for version={version}. stderr={results.stderr!r}"
)
Loading
Loading