diff --git a/.github/resources/.prettierignore b/.github/resources/.prettierignore index af252e185..9738fe10e 100644 --- a/.github/resources/.prettierignore +++ b/.github/resources/.prettierignore @@ -12,3 +12,5 @@ **/CMakeUserPresets.json **/CMakePresets.json **/CMakeFiles/ + +**/tests/api/README.md \ No newline at end of file diff --git a/scene_common/src/scene_common/rest_client.py b/scene_common/src/scene_common/rest_client.py index b9902d973..c561aa9ba 100644 --- a/scene_common/src/scene_common/rest_client.py +++ b/scene_common/src/scene_common/rest_client.py @@ -1,14 +1,14 @@ -# SPDX-FileCopyrightText: (C) 2023 - 2025 Intel Corporation +# SPDX-FileCopyrightText: (C) 2023 - 2026 Intel Corporation # SPDX-License-Identifier: Apache-2.0 import os import json import re import requests -import sys from http import HTTPStatus from urllib.parse import urljoin + class RESTResult(dict): def __init__(self, statusCode, errors=None): super().__init__() @@ -16,14 +16,38 @@ def __init__(self, statusCode, errors=None): self.errors = errors return + @property + def status_code(self): + return self.statusCode + + def json(self): + return dict(self) + + @property + def text(self): + return json.dumps(dict(self)) + + class RESTClient: - def __init__(self, url, rootcert=None, auth=None): + def __init__(self, url=None, token=None, auth=None, + rootcert=None, verify_ssl=False, timeout=10): self.url = url - self.rootcert = rootcert - if not self.url.endswith("/"): + + if self.url and not self.url.endswith("/"): self.url = self.url + "/" + + # Handle SSL verification (support both bool and path) + self.verify_ssl = verify_ssl if verify_ssl is not False else False + if rootcert: + self.verify_ssl = rootcert + + self.timeout = timeout self.session = requests.session() - if auth: + + # If token provided directly, use it (skip authentication) + if token: + self.token = token + elif auth: self._parseAuth(auth) return @@ -46,10 +70,10 @@ def _parseAuth(self, auth): res = self.authenticate(user, pw) if not res: error_message = ( - f"Failed to authenticate\n" - f" URL: {self.url}\n" - f" status: {res.statusCode}\n" - f" errors: {res.errors}" + f"Failed to authenticate\n" + f" URL: {self.url}\n" + f" status: {res.statusCode}\n" + f" errors: {res.errors}" ) raise RuntimeError(error_message) return @@ -58,6 +82,38 @@ def _parseAuth(self, auth): def isAuthenticated(self): return hasattr(self, 'token') and self.token is not None + def _headers(self): + headers = { + "Content-Type": "application/json" + } + if hasattr(self, 'token') and self.token: + headers["Authorization"] = f"Token {self.token}" + return headers + + def request(self, method, path, **kwargs): + """ + Returns raw requests.Response object for compatibility with API tests + """ + # Ensure path starts with / + if not path.startswith('/'): + path = '/' + path + + url = urljoin(self.url, path.lstrip('/')) + + # Merge headers + headers = self._headers() + if 'headers' in kwargs: + headers.update(kwargs.pop('headers')) + + return self.session.request( + method=method, + url=url, + headers=headers, + verify=self.verify_ssl, + timeout=self.timeout, + **kwargs + ) + def decodeReply(self, reply, expectedStatus, successContent=None): result = RESTResult(statusCode=reply.status_code) decoded = False @@ -69,10 +125,11 @@ def decodeReply(self, reply, expectedStatus, successContent=None): content = reply.content else: content = { - 'data': reply.content, + 'data': reply.content, } if 'Content-Disposition' in reply.headers: - fname = re.findall("filename=(.+)", reply.headers['Content-Disposition'])[0] + fname = re.findall("filename=(.+)", + reply.headers['Content-Disposition'])[0] content['filename'] = fname decoded = True @@ -99,11 +156,15 @@ def authenticate(self, user, password): auth_url = urljoin(self.url, "auth") try: reply = self.session.post(auth_url, data={'username': user, 'password': password}, - verify=self.rootcert) + verify=self.verify_ssl) except requests.exceptions.ConnectionError as err: - result = RESTResult("ConnectionError", errors=("Connection error", str(err))) + result = RESTResult( + "ConnectionError", errors=( + "Connection error", str(err))) else: - result = self.decodeReply(reply, HTTPStatus.OK, successContent={'authenticated': True}) + result = self.decodeReply( + reply, HTTPStatus.OK, successContent={ + 'authenticated': True}) if reply.status_code == HTTPStatus.OK: data = json.loads(reply.content) self.token = data['token'] @@ -120,7 +181,8 @@ def prepareDataArgs(self, data, files): if not files: data_args = {'json': data} elif self.dataIsNested(data): - raise ValueError("requests library can't combine files and nested dictionaries") + raise ValueError( + "requests library can't combine files and nested dictionaries") return data_args def _create(self, endpoint, data, files=None): @@ -137,7 +199,7 @@ def _create(self, endpoint, data, files=None): headers = {'Authorization': f"Token {self.token}"} data_args = self.prepareDataArgs(data, files) reply = self.session.post(full_path, **data_args, files=files, - headers=headers, verify=self.rootcert) + headers=headers, verify=self.verify_ssl) return self.decodeReply(reply, HTTPStatus.CREATED) def _get(self, endpoint, parameters): @@ -152,7 +214,7 @@ def _get(self, endpoint, parameters): full_path = urljoin(self.url, endpoint) headers = {'Authorization': f"Token {self.token}"} reply = self.session.get(full_path, params=parameters, headers=headers, - verify=self.rootcert) + verify=self.verify_ssl) return self.decodeReply(reply, HTTPStatus.OK) def _update(self, endpoint, data, files=None): @@ -169,7 +231,7 @@ def _update(self, endpoint, data, files=None): headers = {'Authorization': f"Token {self.token}"} data_args = self.prepareDataArgs(data, files) reply = self.session.post(full_path, **data_args, files=files, - headers=headers, verify=self.rootcert) + headers=headers, verify=self.verify_ssl) return self.decodeReply(reply, HTTPStatus.OK) def _delete(self, endpoint): @@ -181,7 +243,10 @@ def _delete(self, endpoint): """ full_path = urljoin(self.url, endpoint) headers = {'Authorization': f"Token {self.token}"} - reply = self.session.delete(full_path, headers=headers, verify=self.rootcert) + reply = self.session.delete( + full_path, + headers=headers, + verify=self.verify_ssl) return self.decodeReply(reply, HTTPStatus.OK) def _separateFiles(self, data, fields): diff --git a/tests/api/README.md b/tests/api/README.md new file mode 100644 index 000000000..8a78dd83d --- /dev/null +++ b/tests/api/README.md @@ -0,0 +1,325 @@ +# API Test Framework + +A data-driven, multi-step REST API test framework built on pytest. Tests are defined as JSON scenario files. + +--- + +## Project Structure + +``` +tests +├── api + ├── test_sscape_api.py # Main test runner + ├── conftest.py # Pytest configuration + ├── scenarios/ # Default directory for test scenario files + │ ├── camera_api.json + │ ├── scene_api.json + │ ├── sensor_api.json + │ └── ... + └── api_test.log # Auto-generated log file +``` + +--- + +## Requirements + +- Python 3.8+ +- pytest +- requests + +Install dependencies: +```bash +pip install pytest requests +``` + +--- + +## Environment Variables + +| Variable | Default | Description | +|-----------------|--------------------------------|------------------------------------| +| `API_TOKEN` | `token` | Authentication token for API calls | +| `API_BASE_URL` | `https://localhost/api/v1` | Base URL of the target API | + +Set them before running: +```bash +export API_TOKEN=your_token_here +export API_BASE_URL=https://your-server/api/v1 +``` + +--- + +## Running Tests + +### Basic syntax +```bash +pytest -s test_sscape_api.py --file [--test_case ] [--junitxml=test-results.xml] +``` + +### Run all scenarios from the default `scenarios/` folder +```bash +pytest -s test_sscape_api.py +``` + +### Run all scenarios from a specific JSON file +```bash +pytest -s test_sscape_api.py --file scenarios/scene_api.json +``` + +### Run all scenarios from a folder +```bash +pytest -s test_sscape_api.py --file scenarios/ +``` + +### Run a single test case by ID +```bash +pytest -s test_sscape_api.py --file scenarios/scene_api.json --test_case Vision_AI/SSCAPE/API/SCENE/01 +``` + +### Run with JUnit XML report (for CI/CD) +```bash +pytest -s test_sscape_api.py --file scenarios/scene_api.json --junitxml=test-results.xml +``` + +### Combined example +```bash +pytest -s test_sscape_api.py \ + --file scenarios/scene_api.json \ + --test_case Vision_AI/SSCAPE/API/SCENE/01 \ + --junitxml=test-results.xml +``` + +--- + +## Scenario File Format + +Scenarios are JSON files containing an array of test cases. Each test case has one or more sequential steps. + +### Top-level structure +```json +[ + { "test_name": "test_case_1" }, + { "test_name": "test_case_2" } +] +``` + +### Test case fields + +| Field | Required | Description | +|--------------|----------|----------------------------------------------| +| `test_name` | Yes | Unique identifier used with `--test_case` | +| `test_steps` | Yes | Array of steps executed in order | + +### Step fields + +| Field | Required | Description | +|---------------------|----------|----------------------------------------------------------------------------------------| +| `step_name` | No | Human-readable label shown in logs and failure messages | +| `api` | Yes | API group: `camera`, `scene`, `sensor`, `region`, `tripwire`, `user`, `asset`, `child` | +| `method` | Yes | RESTClient method name (e.g. `createCamera`, `getScene`) | +| `request` | No | Arguments passed to the method (see key mapping below) | +| `expected_status` | No | Assertions on the response (currently `status_code`) | +| `save` | No | Variables to extract from the response for later steps | +| `validate` | No | Response body field assertions using dot-notation (partial match) | +| `expected_body` | No | Full response body structure validation (exact match) | + +### Request key mapping + +The JSON request keys are automatically mapped to RESTClient parameter names: + +| JSON key | RESTClient parameter | Usage | +|----------|----------------------|------------------------------| +| `body` | `data` | Request body (POST/PUT) | +| `UID` | `uid` | Path parameter | + +List methods (e.g. `getCameras`, `getScenes`) automatically receive `filter=None` if no filter is provided in the request. + +--- + +## Full Scenario Example + +```json +[ + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/01: Create scene with only required properties", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "name": "Scene1", + "use_tracker": true, + "output_lla": false + } + }, + "expected_status": { + "status_code": 201 + }, + "save": { + "SCENE_UID": "uid" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/08: Update scene with minimal valid payload", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene1_Updated", + "use_tracker": false, + "output_lla": true + } + }, + "expected_status": { + "status_code": 200 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "expected_body": { + "uid": "${SCENE_UID}", + "name": "Scene1_Updated", + "map_type": "map_upload", + "use_tracker": false, + "output_lla": true, + "mesh_translation": [ + 0, + 0, + 0 + ], + "mesh_rotation": [ + 0, + 0, + 0 + ], + "mesh_scale": [ + 1.0, + 1.0, + 1.0 + ], + "regulated_rate": 30.0, + "external_update_rate": 30.0, + "camera_calibration": "Manual", + "apriltag_size": 0.162, + "number_of_localizations": 50, + "global_feature": "netvlad", + "local_feature": { + "sift": {} + }, + "matcher": { + "NN-ratio": {} + }, + "minimum_number_of_matches": 20, + "inlier_threshold": 0.5, + "geospatial_provider": "google", + "map_zoom": 15.0, + "map_bearing": 0.0 + } + } + ] + } +] +``` + +--- + +## Variable Substitution + +Values saved from one step can be referenced in later steps using `${VAR_NAME}` syntax. + +### Saving a value +```json +"save": { + "SCENE_UID": "uid" +} +``` +This extracts the `uid` field from the response body and stores it as `SCENE_UID`. The value is also set as an environment variable for the duration of the test run. + +### Using a saved value +```json +"request": { + "UID": "${SCENE_UID}" +} +``` + +Variable substitution works recursively in any nested `request` object. + +--- + +## Response Validation + +### Status code assertion +Always checked if `expected.status_code` is set: +```json +"expected_status": { + "status_code": 201 +} +``` + +### Body field assertions +Check specific fields in the response body using dot-notation for nested fields: +```json +"validate": { + "name": "Scene1", + "mesh_scale": [1.0, 1.0, 1.0] +} +``` + +If any assertion fails, the step fails with a detailed diff message showing expected vs actual values. + +--- + +## Logging + +All runs produce two log outputs: + +| Output | Level | Location | +|----------------|---------|-----------------------------------| +| Console | INFO | stdout (visible with `-s` flag) | +| File | DEBUG | `api_test.log` next to test file | + +The log file is overwritten on each run. Debug output includes full request data, response status, response body, and saved variable values. + +--- + +## Available API Methods + +All methods live in `RESTClient`: + +| API group | Methods | +|------------|--------------------------------------------------------------| +| `scene` | `getScenes`, `createScene`, `getScene`, `updateScene`, `deleteScene` | +| `camera` | `getCameras`, `createCamera`, `getCamera`, `updateCamera`, `deleteCamera` | +| `sensor` | `getSensors`, `createSensor`, `getSensor`, `updateSensor`, `deleteSensor` | +| `region` | `getRegions`, `createRegion`, `getRegion`, `updateRegion`, `deleteRegion` | +| `tripwire` | `getTripwires`, `createTripwire`, `getTripwire`, `updateTripwire`, `deleteTripwire` | +| `user` | `getUsers`, `createUser`, `getUser`, `updateUser`, `deleteUser` | +| `asset` | `getAssets`, `createAsset`, `getAsset`, `updateAsset`, `deleteAsset` | +| `child` | `getChildScene`, `updateChildScene` | + +--- + +## Adding New Tests + +1. Create or open a JSON file in `scenarios/` +2. Add a new object to the array following the schema above +3. Use `test_name` in the format `Vision_AI/SSCAPE/Endpoint/TestCase_No: Test case title` for consistency +4. Run with `--test_case` to verify before committing + +No Python changes required. diff --git a/tests/api/conftest.py b/tests/api/conftest.py new file mode 100644 index 000000000..0b2c439a7 --- /dev/null +++ b/tests/api/conftest.py @@ -0,0 +1,139 @@ +# SPDX-FileCopyrightText: (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +""" +conftest.py - Custom pytest configuration for clean XML output +This generates Robot Framework-style minimal XML directly from pytest +""" + +import pytest +import xml.etree.ElementTree as ET +from xml.dom import minidom +import time + + +class CleanXMLReporter: + """Custom pytest plugin to generate clean, minimal JUnit XML""" + + def __init__(self): + self.test_results = [] + self.start_time = None + + def pytest_sessionstart(self, session): + """Called at the start of test session""" + self.start_time = time.time() + + @pytest.hookimpl(hookwrapper=True) + def pytest_runtest_makereport(self, item, call): + """Capture test results""" + outcome = yield + report = outcome.get_result() + + # Only process the actual test call (not setup/teardown) + if report.when == 'call': + # Extract test name from item + test_name = item.nodeid + + # If using parametrize, the name will be in item.name + if hasattr(item, 'name'): + test_name = item.name + + # Extract just the parameter part if exists + # Format: test_api_scenario[Test Name Here] + if '[' in test_name and ']' in test_name: + test_name = test_name.split('[', 1)[1].rsplit(']', 1)[0] + + result = { + 'name': test_name, + 'classname': 'VisionAI_API_Tests', + 'time': report.duration, + 'outcome': report.outcome, # 'passed', 'failed', 'skipped' + 'error_message': str(report.longrepr) if report.failed else None + } + + self.test_results.append(result) + + def pytest_sessionfinish(self, session): + """Generate clean XML at the end of session""" + if not hasattr(session.config, 'workerinput'): # Skip in xdist workers + self._generate_clean_xml(session) + + def _generate_clean_xml(self, session): + """Create minimal JUnit XML file""" + # Get output path from pytest config + xml_path = session.config.option.xmlpath + if not xml_path: + return # No XML output requested + + # Calculate totals + total_tests = len(self.test_results) + passed = sum(1 for r in self.test_results if r['outcome'] == 'passed') + failed = sum(1 for r in self.test_results if r['outcome'] == 'failed') + skipped = sum(1 for r in self.test_results if r['outcome'] == 'skipped') + errors = 0 # pytest doesn't distinguish errors from failures + total_time = sum(r['time'] for r in self.test_results) + + # Create testsuite element + testsuite = ET.Element('testsuite', { + 'name': 'VisionAI_API_Tests', + 'tests': str(total_tests), + 'errors': str(errors), + 'failures': str(failed), + 'skipped': str(skipped), + 'time': f'{total_time:.3f}' + }) + + # Add each test case + for result in self.test_results: + testcase = ET.SubElement(testsuite, 'testcase', { + 'classname': result['classname'], + 'name': result['name'], + 'time': f"{result['time']:.3f}" + }) + + # Only add failure element if test failed (optional) + # Comment out the following block to have completely clean XML + if result['outcome'] == 'failed': + failure = ET.SubElement(testcase, 'failure', { + 'message': 'Test failed' + }) + elif result['outcome'] == 'skipped': + ET.SubElement(testcase, 'skipped') + + # Pretty print and save + xml_str = ET.tostring(testsuite, encoding='unicode') + dom = minidom.parseString(xml_str) + pretty_xml = dom.toprettyxml(indent=' ') + + # Remove extra blank lines + lines = [line for line in pretty_xml.split('\n') if line.strip()] + clean_xml = '\n'.join(lines) + + # Write to file + with open(xml_path, 'w', encoding='utf-8') as f: + f.write(clean_xml) + + print(f"\n✓ Clean XML report generated: {xml_path}") + print( + f" Total: {total_tests}, Passed: {passed}, Failed: {failed}, Skipped: {skipped}") + + +def pytest_configure(config): + """Register the clean XML reporter plugin""" + # Only register if XML output is requested + if config.option.xmlpath: + # Unregister default junitxml plugin + if hasattr(config, '_xml'): + config.pluginmanager.unregister(config._xml) + + # Register our custom clean reporter + clean_reporter = CleanXMLReporter() + config.pluginmanager.register(clean_reporter, 'cleanxml') + config._cleanxml = clean_reporter + + +def pytest_addoption(parser): + parser.addoption("--file", default=None, + help="Specific scenario file to run (e.g., 'scenarios/scene.json')") + parser.addoption("--test_case", default=None, + help="Specific test case name to run") diff --git a/tests/api/scenarios/scene_api.json b/tests/api/scenarios/scene_api.json new file mode 100644 index 000000000..5d0cc1f72 --- /dev/null +++ b/tests/api/scenarios/scene_api.json @@ -0,0 +1,845 @@ +[ + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/01: Create scene with only required properties", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "name": "Scene1", + "use_tracker": true, + "output_lla": false + } + }, + "expected_status": { + "status_code": 201 + }, + "save": { + "SCENE_UID": "uid" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/02: Create scene with full valid payload", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "name": "Scene2", + "use_tracker": true, + "output_lla": false, + "mesh_translation": [12.2, 248, 12], + "mesh_rotation": [12.2, 248, 12], + "mesh_scale": [1, 1, 1], + "tracker_config": [1, 1, 1] + } + }, + "expected_status": { + "status_code": 201 + }, + "save": { + "SCENE_UID_FULL": "uid" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/03: Create scene with missing required name property", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "use_tracker": true, + "output_lla": false + } + }, + "expected_status": { + "status_code": 400 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/04: Create scene with mesh_translation below min length", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "name": "Scene_MeshTranslation_Invalid", + "use_tracker": true, + "output_lla": false, + "mesh_translation": [1.0, 2.0] + } + }, + "expected_status": { + "status_code": 400 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/05: Create scene with mesh_rotation below min length", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "name": "Scene_MeshRotation_Invalid", + "use_tracker": true, + "output_lla": false, + "mesh_rotation": [12.2, 248] + } + }, + "expected_status": { + "status_code": 400 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/06: Create scene with mesh_scale below min length", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "name": "Scene_MeshScale_Invalid", + "use_tracker": true, + "output_lla": false, + "mesh_scale": [1.0, 1.0] + } + }, + "expected_status": { + "status_code": 400 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/07: Create scene with tracker_config below min length", + "test_steps": [ + { + "step_name": "Create Scene", + "api": "scene", + "method": "createScene", + "request": { + "body": { + "name": "Scene_TrackerConfig_Invalid", + "use_tracker": true, + "output_lla": false, + "tracker_config": [] + } + }, + "expected_status": { + "status_code": 400 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/08: Update scene with minimal valid payload", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene1_Updated", + "use_tracker": false, + "output_lla": true + } + }, + "expected_status": { + "status_code": 200 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "expected_body": { + "uid": "${SCENE_UID}", + "name": "Scene1_Updated", + "map_type": "map_upload", + "use_tracker": false, + "output_lla": true, + "mesh_translation": [0, 0, 0], + "mesh_rotation": [0, 0, 0], + "mesh_scale": [1.0, 1.0, 1.0], + "regulated_rate": 30.0, + "external_update_rate": 30.0, + "camera_calibration": "Manual", + "apriltag_size": 0.162, + "number_of_localizations": 50, + "global_feature": "netvlad", + "local_feature": { + "sift": {} + }, + "matcher": { + "NN-ratio": {} + }, + "minimum_number_of_matches": 20, + "inlier_threshold": 0.5, + "geospatial_provider": "google", + "map_zoom": 15.0, + "map_bearing": 0.0 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/09: Update scene with full valid payload", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene1_Updated_Full", + "use_tracker": true, + "output_lla": true, + "mesh_translation": [12.2, 248, 12.0], + "mesh_rotation": [12.2, 248, 12.0], + "mesh_scale": [2.0, 2.0, 2.0] + } + }, + "expected_status": { + "status_code": 200 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "expected_body": { + "uid": "${SCENE_UID}", + "name": "Scene1_Updated_Full", + "map_type": "map_upload", + "use_tracker": true, + "output_lla": true, + "mesh_translation": [12.2, 248.0, 12.0], + "mesh_rotation": [12.2, 248.0, 12.0], + "mesh_scale": [2.0, 2.0, 2.0], + "regulated_rate": 30.0, + "external_update_rate": 30.0, + "camera_calibration": "Manual", + "apriltag_size": 0.162, + "number_of_localizations": 50, + "global_feature": "netvlad", + "local_feature": { + "sift": {} + }, + "matcher": { + "NN-ratio": {} + }, + "minimum_number_of_matches": 20, + "inlier_threshold": 0.5, + "geospatial_provider": "google", + "map_zoom": 15.0, + "map_bearing": 0.0 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/10: Update scene with missing required body", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": {} + }, + "expected_status": { + "status_code": 400 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "validate": { + "name": "Scene1_Updated_Full" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/11: Update scene with mesh_translation below min length", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene_Update_MeshTranslation_Invalid", + "use_tracker": true, + "output_lla": false, + "mesh_translation": [1.0] + } + }, + "expected_status": { + "status_code": 400 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "validate": { + "name": "Scene1_Updated_Full" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/12: Update scene with mesh_rotation below min length", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene_Update_MeshRotation_Invalid", + "use_tracker": true, + "output_lla": false, + "mesh_rotation": [0.0, 0.0] + } + }, + "expected_status": { + "status_code": 400 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "validate": { + "name": "Scene1_Updated_Full" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/13: Update scene with mesh_scale below min length", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene_Update_MeshScale_Invalid", + "use_tracker": true, + "output_lla": false, + "mesh_scale": [1.0] + } + }, + "expected_status": { + "status_code": 400 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "validate": { + "name": "Scene1_Updated_Full" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/14: Update scene with tracker_config below min length", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene_Update_TrackerConfig_Invalid", + "use_tracker": true, + "output_lla": false, + "tracker_config": [] + } + }, + "expected_status": { + "status_code": 400 + } + }, + { + "step_name": "Verify resource was updated", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "validate": { + "name": "Scene1_Updated_Full" + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/15: Update scene with read only uid property in body", + "test_steps": [ + { + "step_name": "Update Scene", + "api": "scene", + "method": "updateScene", + "request": { + "UID": "${SCENE_UID}", + "body": { + "name": "Scene_Invalid_UID", + "use_tracker": true, + "output_lla": false, + "uid": "invalid" + } + }, + "expected_status": { + "status_code": 400 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/16: Get scene by UID", + "test_steps": [ + { + "step_name": "Get Scene", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + }, + "expected_body": { + "uid": "${SCENE_UID}", + "name": "Scene1_Updated_Full", + "map_type": "map_upload", + "use_tracker": true, + "output_lla": true, + "mesh_translation": [12.2, 248.0, 12.0], + "mesh_rotation": [12.2, 248.0, 12.0], + "mesh_scale": [2.0, 2.0, 2.0], + "regulated_rate": 30.0, + "external_update_rate": 30.0, + "camera_calibration": "Manual", + "apriltag_size": 0.162, + "number_of_localizations": 50, + "global_feature": "netvlad", + "local_feature": { + "sift": {} + }, + "matcher": { + "NN-ratio": {} + }, + "minimum_number_of_matches": 20, + "inlier_threshold": 0.5, + "geospatial_provider": "google", + "map_zoom": 15.0, + "map_bearing": 0.0 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/17: Get non existent scene", + "test_steps": [ + { + "step_name": "Get Scene", + "api": "scene", + "method": "getScene", + "request": { + "UID": "123" + }, + "expected_status": { + "status_code": 404 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/18: Get all scenes", + "test_steps": [ + { + "step_name": "Get all Scenes", + "api": "scene", + "method": "getScenes", + "request": {}, + "expected_status": { + "status_code": 200 + }, + "expected_body": { + "count": 3, + "next": null, + "previous": null, + "results": [ + { + "uid": "3bc091c7-e449-46a0-9540-29c499bca18c", + "name": "Retail", + "map_type": "map_upload", + "use_tracker": true, + "output_lla": false, + "map": "https://localhost/media/HazardZoneSceneLarge.png", + "cameras": [ + { + "uid": "camera1", + "name": "camera1", + "intrinsics": { + "fx": 571.2592026968458, + "fy": 571.2592026968458, + "cx": 320.0, + "cy": 240.0 + }, + "transform_type": "3d-2d point correspondence", + "transforms": [ + 278.0, 61.0, 621.0, 132.0, 559.0, 460.0, 66.0, 289.0, 0.1, + 5.38, 3.04, 5.35, 3.05, 2.42, 0.1, 2.45 + ], + "distortion": { + "k1": 0.0, + "k2": 0.0, + "p1": 0.0, + "p2": 0.0, + "k3": 0.0 + }, + "translation": [ + 2.6651330996559883, 1.0075648853123316, 2.603863333755973 + ], + "rotation": [ + -137.85924651441334, -19.441505783168942, + -15.384890268257454 + ], + "scale": [1.0000000000000007, 1.0, 1.0], + "resolution": [640, 480], + "scene": "3bc091c7-e449-46a0-9540-29c499bca18c" + }, + { + "uid": "camera2", + "name": "camera2", + "intrinsics": { + "fx": 571.2592026968458, + "fy": 571.2592026968458, + "cx": 320.0, + "cy": 240.0 + }, + "transform_type": "3d-2d point correspondence", + "transforms": [ + 31.0, 228.0, 423.0, 266.0, 537.0, 385.0, 79.0, 343.0, 1.06, + 5.34, 4.0, 5.38, 4.98, 4.39, 2.04, 4.38 + ], + "distortion": { + "k1": 0.0, + "k2": 0.0, + "p1": 0.0, + "p2": 0.0, + "k3": 0.0 + }, + "translation": [ + 4.034863921795162, 2.277766310708989, 2.955114373391866 + ], + "rotation": [ + -132.15360745910087, -8.172752500708558, -5.298590495090165 + ], + "scale": [1.0, 1.0, 1.0], + "resolution": [640, 480], + "scene": "3bc091c7-e449-46a0-9540-29c499bca18c" + } + ], + "mesh_translation": [0, 0, 0], + "mesh_rotation": [0, 0, 0], + "mesh_scale": [1.0, 1.0, 1.0], + "scale": 100.0, + "regulated_rate": 30.0, + "external_update_rate": 30.0, + "camera_calibration": "AprilTag", + "apriltag_size": 0.147, + "map_processed": "2023-06-08T13:53:58.767000Z", + "number_of_localizations": 50, + "global_feature": "netvlad", + "local_feature": { + "sift": {} + }, + "matcher": { + "NN-ratio": {} + }, + "minimum_number_of_matches": 20, + "inlier_threshold": 0.5, + "geospatial_provider": "google", + "map_zoom": 15.0, + "map_bearing": 0.0 + }, + { + "uid": "302cf49a-97ec-402d-a324-c5077b280b7b", + "name": "Queuing", + "map_type": "map_upload", + "use_tracker": true, + "output_lla": false, + "map": "https://localhost/media/scene.png", + "cameras": [ + { + "uid": "atag-qcam1", + "name": "atag-qcam1", + "intrinsics": { + "fx": 905.0, + "fy": 905.0, + "cx": 640.0, + "cy": 360.0 + }, + "transform_type": "3d-2d point correspondence", + "transforms": [ + 119.0, 622.0, 257.0, 561.0, 978.0, 580.0, 628.0, 312.0, + 1.685, 2.533, 2.188, 2.578, 4.449, 1.412, 3.94, 3.228 + ], + "distortion": { + "k1": 0.0, + "k2": 0.0, + "p1": 0.0, + "p2": 0.0, + "k3": 0.0 + }, + "translation": [ + 2.985857104493509, 0.20540788984425282, 2.7150546825598902 + ], + "rotation": [ + -135.08718965001765, 12.682032394455133, 19.24508172546946 + ], + "scale": [0.9999999999999999, 1.0, 1.0], + "resolution": [1280, 720], + "scene": "302cf49a-97ec-402d-a324-c5077b280b7b" + }, + { + "uid": "atag-qcam2", + "name": "atag-qcam2", + "intrinsics": { + "fx": 905.0, + "fy": 905.0, + "cx": 640.0, + "cy": 360.0 + }, + "transform_type": "3d-2d point correspondence", + "transforms": [ + 1012.0, 307.0, 956.0, 613.0, 1121.0, 505.0, 585.0, 316.0, + 3.596, 2.96, 1.583, 2.794, 2.36, 2.266, 2.577, 4.903 + ], + "distortion": { + "k1": 0.0, + "k2": 0.0, + "p1": 0.0, + "p2": 0.0, + "k3": 0.0 + }, + "translation": [ + -0.6544951215349523, 2.8628274940885503, 2.894955006060443 + ], + "rotation": [ + -150.5988934259539, 42.35138027480063, 52.29263795544898 + ], + "scale": [1.0, 1.0000000000000002, 1.0], + "resolution": [1280, 720], + "scene": "302cf49a-97ec-402d-a324-c5077b280b7b" + } + ], + "sensors": [ + { + "uid": "sensor_full", + "scene": "302cf49a-97ec-402d-a324-c5077b280b7b", + "sensor_id": "sensor_full", + "name": "sensor_full", + "area": "circle", + "radius": 2.5, + "center": [3.81, 4.59], + "translation": [3.81, 4.59, 0.0], + "singleton_type": "attribute" + } + ], + "mesh_translation": [0, 0, 0], + "mesh_rotation": [0, 0, 0], + "mesh_scale": [1.0, 1.0, 1.0], + "scale": 157.0, + "regulated_rate": 30.0, + "external_update_rate": 30.0, + "camera_calibration": "AprilTag", + "apriltag_size": 0.318471338, + "map_processed": "2023-06-08T13:54:27.922000Z", + "number_of_localizations": 50, + "global_feature": "netvlad", + "local_feature": { + "sift": {} + }, + "matcher": { + "NN-ratio": {} + }, + "minimum_number_of_matches": 20, + "inlier_threshold": 0.5, + "geospatial_provider": "google", + "map_zoom": 15.0, + "map_bearing": 0.0 + }, + { + "uid": "${SCENE_UID}", + "name": "Scene1_Updated_Full", + "map_type": "map_upload", + "use_tracker": true, + "output_lla": true, + "mesh_translation": [12.2, 248.0, 12.0], + "mesh_rotation": [12.2, 248.0, 12.0], + "mesh_scale": [2.0, 2.0, 2.0], + "regulated_rate": 30.0, + "external_update_rate": 30.0, + "camera_calibration": "Manual", + "apriltag_size": 0.162, + "number_of_localizations": 50, + "global_feature": "netvlad", + "local_feature": { + "sift": {} + }, + "matcher": { + "NN-ratio": {} + }, + "minimum_number_of_matches": 20, + "inlier_threshold": 0.5, + "geospatial_provider": "google", + "map_zoom": 15.0, + "map_bearing": 0.0 + } + ] + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/19: Get child scenes", + "test_steps": [ + { + "step_name": "Get Scene", + "api": "child", + "method": "getChildScene", + "request": {}, + "expected_status": { + "status_code": 200 + }, + "expected_body": { + "count": 0, + "next": null, + "previous": null, + "results": [] + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/20: Delete scene by UID", + "test_steps": [ + { + "step_name": "Delete Scene", + "api": "scene", + "method": "deleteScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 200 + } + }, + { + "step_name": "Verify resource is deleted", + "api": "scene", + "method": "getScene", + "request": { + "UID": "${SCENE_UID}" + }, + "expected_status": { + "status_code": 404 + } + } + ] + }, + { + "test_name": "Vision_AI/SSCAPE/API/SCENE/21: Delete non existent scene", + "test_steps": [ + { + "step_name": "Delete Scene", + "api": "scene", + "method": "deleteScene", + "request": { + "UID": "123" + }, + "expected_status": { + "status_code": 404 + } + } + ] + } +] diff --git a/tests/api/test_sscape_api.py b/tests/api/test_sscape_api.py new file mode 100644 index 000000000..236d9a3e6 --- /dev/null +++ b/tests/api/test_sscape_api.py @@ -0,0 +1,346 @@ +# SPDX-FileCopyrightText: (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +import logging +import os +import json +import glob +import inspect +import pytest +from scene_common.rest_client import RESTClient + +# Logging Configuration +LOG_FILE = os.path.join(os.path.dirname(__file__), "api_test.log") + +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + +formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s") + +console_handler = logging.StreamHandler() +console_handler.setLevel(logging.INFO) +console_handler.setFormatter(formatter) +logger.addHandler(console_handler) + +file_handler = logging.FileHandler(LOG_FILE, mode="w") +file_handler.setLevel(logging.DEBUG) +file_handler.setFormatter(formatter) +logger.addHandler(file_handler) + +logger.info( + "Logger initialized. Logs will be written to console and %s", + LOG_FILE) + +# Setup Base HTTP Client +API_TOKEN = os.environ.get("API_TOKEN") +BASE_URL = os.environ.get("API_BASE_URL", "https://localhost/api/v1") + +http_client = RESTClient(url=BASE_URL, token=API_TOKEN, verify_ssl=False) + +saved_vars = {} + +API_MAP = { + "scene": http_client, + "camera": http_client, + "sensor": http_client, + "region": http_client, + "tripwire": http_client, + "user": http_client, + "asset": http_client, + "child": http_client, +} + + +def load_scenarios(path=None): + """ + Load multi-step test scenarios from JSON file(s) + + Args: + path: Can be: + - A specific JSON file path (e.g., "test/test.json") + - A folder path to load all *.json files from + - None to load from default "scenarios" folder + """ + if path is None: + path = "scenarios" + + scenarios = [] + + if os.path.isfile(path): + logger.info(f"Loading scenario file: {path}") + with open(path, "r") as sf: + data = json.load(sf) + scenarios.extend(data) + elif os.path.isdir(path): + scenario_files = glob.glob(f"{path}/*.json") + logger.info( + f"Loading { + len(scenario_files)} scenario files from folder: {path}") + for f in scenario_files: + with open(f, "r") as sf: + data = json.load(sf) + scenarios.extend(data) + else: + raise FileNotFoundError(f"Scenario path not found: {path}") + + return scenarios + + +def substitute_variables(obj): + """Recursively substitute ${VAR} placeholders with saved values""" + if isinstance(obj, dict): + return {k: substitute_variables(v) for k, v in obj.items()} + if isinstance(obj, list): + return [substitute_variables(x) for x in obj] + if isinstance(obj, str) and obj.startswith("${") and obj.endswith("}"): + var_name = obj[2:-1] + return saved_vars.get(var_name, obj) + return obj + + +def compare_expected_json_body(actual, expected, path="root"): + """ + Deep comparison of two JSON structures with detailed error reporting. + + Args: + actual: Actual response data + expected: Expected response data + path: Current path in the structure (for error messages) + """ + errors = [] + + # Type mismatch + if not isinstance(actual, type(expected)): + errors.append( + f"{path}: type mismatch - expected { + type(expected).__name__}, got { + type(actual).__name__}") + return False, errors + + # Dict comparison + if isinstance(expected, dict): + # Check for missing keys + for key in expected: + if key not in actual: + errors.append(f"{path}.{key}: missing in actual response") + + # Check for extra keys + for key in actual: + if key not in expected: + errors.append(f"{path}.{key}: unexpected key in actual response") + + # Recursively compare matching keys + for key in expected: + if key in actual: + success, sub_errors = compare_expected_json_body( + actual[key], expected[key], f"{path}.{key}") + errors.extend(sub_errors) + + # List comparison + elif isinstance(expected, list): + if len(actual) != len(expected): + errors.append( + f"{path}: list length mismatch - expected {len(expected)}, got {len(actual)}") + else: + for i, (actual_item, expected_item) in enumerate(zip(actual, expected)): + success, sub_errors = compare_expected_json_body( + actual_item, expected_item, f"{path}[{i}]") + errors.extend(sub_errors) + + # Primitive comparison + else: + if actual != expected: + errors.append(f"{path}: expected '{expected}', got '{actual}'") + + return len(errors) == 0, errors + + +def validate_response(response_body, validation_rules): + """ + Validate response body against expected values + + Args: + response_body: The actual response (dict or object) + validation_rules: Dict of field:expected_value pairs + """ + errors = [] + + for field, expected_value in validation_rules.items(): + # Navigate nested fields using dot notation + actual_value = response_body + for key in field.split('.'): + if isinstance(actual_value, dict): + actual_value = actual_value.get(key) + else: + actual_value = getattr(actual_value, key, None) + + if actual_value != expected_value: + errors.append( + f"Field '{field}': expected '{expected_value}', got '{actual_value}'" + ) + + return len(errors) == 0, errors + + +def execute_step(step, step_number, total_steps): + step_name = step.get("step_name", f"Step {step_number}") + api_name = step["api"] + method_name = step["method"] + request_data = substitute_variables(step.get("request", {})) + expected_status = step.get("expected_status", {}) + save_vars = step.get("save", {}) + validate_rules = step.get("validate", {}) + expected_body = step.get("expected_body") + + logger.debug(f" [{step_number}/{total_steps}] {step_name}") + logger.debug(f" API: {api_name}, Method: {method_name}") + logger.debug(f" Request: {request_data}") + + # Get API client + api = API_MAP.get(api_name) + if not api: + return False, None, f"Unknown API client: {api_name}" + + if not hasattr(api, method_name): + return False, None, f"API {api_name} has no method {method_name}" + + # Normalize request keys to match RESTClient parameter names: + # "UID" -> "uid" (path parameter used in camera/scene/sensor/etc.) + # "body" -> "data" (request body) + KEY_MAP = {"UID": "uid", "body": "data"} + request_data = {KEY_MAP.get(k, k): v for k, v in request_data.items()} + + # If the method expects "filter" and it wasn't provided, default to None + api_method = getattr(api, method_name) + method_params = inspect.signature(api_method).parameters + if "filter" in method_params and "filter" not in request_data: + request_data["filter"] = None + + # Execute API call + try: + api_method = getattr(api, method_name) + response = api_method(**request_data) + except Exception as e: + return False, None, f"API call failed: {str(e)}" + + # Parse response + try: + response_body = response.json() + except Exception: + response_body = response.text + + logger.debug(f" Response Status: {response.status_code}") + logger.debug(f" Response Body: {json.dumps( + response_body, + indent=2) if isinstance( + response_body, + dict) else response_body}") + + # Check status code + expected_status = expected_status.get("status_code", 200) + if response.status_code != expected_status: + return False, response, f"Expected status {expected_status}, got {response.status_code}" + + # Validate entire response body if expected_body is provided + if expected_body is not None: + logger.debug( + f" Validating entire response body against expected structure") + expected_body = substitute_variables(expected_body) + success, errors = compare_expected_json_body(response_body, expected_body) + if not success: + error_msg = "Response body validation failed: " + \ + "\n".join(f" - {e}" for e in errors) + return False, response, error_msg + + # Save variables + for var_name, path in save_vars.items(): + val = response_body if isinstance(response_body, (dict, list)) else response + + for key in path.split("."): + if isinstance(val, dict): + val = val.get(key) + else: + val = getattr(val, key, None) + + if val is not None: + saved_vars[var_name] = val + os.environ[var_name] = str(val) + logger.debug(f" Saved variable '{var_name}' = {val}") + else: + logger.warning(f" Could not find path '{path}' in response") + + # Validate response body if expected result provided + if validate_rules: + logger.debug(f" Validating response against rules: {validate_rules}") + success, errors = validate_response(response_body, validate_rules) + if not success: + error_msg = "Response validation failed:\n" + \ + "\n".join(f" - {e}" for e in errors) + return False, response, error_msg + + logger.debug(f" ✓ Step passed") + return True, response, None + +# Pytest Parametrize + + +def pytest_generate_tests(metafunc): + """ + Dynamically generate test parameters based on --file and --test_case options + """ + if "test_case" in metafunc.fixturenames: + file_path = metafunc.config.getoption("--file") + test_case_filter = metafunc.config.getoption("--test_case") + + scenarios = load_scenarios(file_path) + + # filter by test_case ID if specified (e.g., --test_case + # Vision_AI/SSCAPE/API/SCENE/01) + if test_case_filter: + original_count = len(scenarios) + scenarios = [s for s in scenarios if s.get("test_name", "").split(":")[ + 0].strip() == test_case_filter] + logger.info( + f"Filtered scenarios: { + len(scenarios)}/{original_count} matching test_case '{test_case_filter}'") + + if not scenarios: + pytest.fail(f"No test case found with ID: {test_case_filter}") + + if not scenarios: + pytest.fail(f"No scenarios found in: {file_path or 'scenarios'}") + + metafunc.parametrize( + "test_case", + scenarios, + ids=lambda tc: tc.get("test_name", "unnamed_test"), + ) + + +def test_api_scenario_multistep(test_case): + """ + Execute a multi-step API test scenario + + Each test case can have multiple steps that execute sequentially. + If any step fails, the entire test case is marked as failed. + """ + test_name = test_case.get("test_name", "unnamed_test") + test_steps = test_case.get("test_steps", []) + + if not test_steps: + pytest.fail("Test case has no steps defined") + + logger.debug(f"\n{'=' * 70}") + logger.debug(f"Test: {test_name}") + logger.debug(f"Steps: {len(test_steps)}") + logger.debug(f"{'=' * 70}") + + # Execute each step + for step_num, step in enumerate(test_steps, start=1): + success, _, error_msg = execute_step(step, step_num, len(test_steps)) + + if not success: + step_name = step.get("step_name", f"Step {step_num}") + pytest.fail(f"Step {step_num} '{step_name}' failed: {error_msg}") + + logger.debug(f"✓ All {len(test_steps)} steps passed\n")