From a478fd46e639269ec550d6992109e111e13e3217 Mon Sep 17 00:00:00 2001 From: hpoeche Date: Thu, 30 Jul 2026 10:25:58 +0200 Subject: [PATCH 1/4] First draft of applying testdatagen --- sdk/test/testdatagen/Readme.md | 16 +++++ sdk/test/testdatagen/example_tests.py | 93 +++++++++++++++++++++++++++ 2 files changed, 109 insertions(+) create mode 100644 sdk/test/testdatagen/Readme.md create mode 100644 sdk/test/testdatagen/example_tests.py diff --git a/sdk/test/testdatagen/Readme.md b/sdk/test/testdatagen/Readme.md new file mode 100644 index 00000000..827ffcbe --- /dev/null +++ b/sdk/test/testdatagen/Readme.md @@ -0,0 +1,16 @@ +## Fuzzy Testing with `aas-core-testdatagen` +This folder contains a script to test the implementation of the `basyx-python-sdk` against the example files +generated by the [`aas-core-works/aas-core-testdatagen`](https://github.com/aas-core-works/aas-core-testdatagen) project. + +### Prerequisites +The script requires the example files to exist in the specified path in the same structure as they are shipped in the +[`aas-specs-metamodel`](https://github.com/admin-shell-io/aas-specs-metamodel) repository. Specifically, if +`$EXAMPLES` is the path passed to the script the following directories are scanned for `.json` and `.xml` files respectively: + +``` +$EXAMPLES/json/examples/generated +$EXAMPLES/xml/examples/generated +``` + +Additionally, the `basyx-python-sdk` must be installed via `pip install sdk/.`. + diff --git a/sdk/test/testdatagen/example_tests.py b/sdk/test/testdatagen/example_tests.py new file mode 100644 index 00000000..cecfe1f4 --- /dev/null +++ b/sdk/test/testdatagen/example_tests.py @@ -0,0 +1,93 @@ +import argparse +import logging +import re +import sys +import traceback +from pathlib import Path +from typing import Callable, Literal, Optional + +from basyx.aas import adapter +from basyx.aas.model import AASConstraintViolation + +logger = logging.getLogger(__name__) + +def sanitize_name(name: str, max_len: int = 120) -> str: + name = re.sub(r'[<>:"/\\|?*\x00-\x1f\s]', '_', name) + return name[:max_len] + +def test_json_example(example_file: Path, base_path: Optional[Path] = None, output_dir: Optional[Path] = None) -> bool: + """ + Attempts to read and deserialize an example file. Reports the status and saves information to output path. + + :param example_file: File which contains the example that should be deserialized + :param deserializer: Function that is used to deserialize the example + :param output_dir: Path to store the failed outputs + :return: bool that indicates if the example was successfully deserialized + """ + try: + adapter.json.read_aas_json_file(example_file, failsafe=False) + return True + except (KeyError, TypeError, ValueError, AASConstraintViolation) as ex: + tb = traceback.format_exc() + error_msg = str(ex).split(">>>")[0].strip() + + rel_path = example_file.relative_to(base_path) if base_path else example_file + logger.error(f"[JSON] ERROR on {rel_path}: {error_msg}") + + if output_dir: + error_dir = output_dir / sanitize_name(error_msg) + error_dir.mkdir(parents=True, exist_ok=True) + error_file = error_dir / f"{sanitize_name(example_file.stem)}.txt" + + json_content = example_file.read_text(encoding="utf-8") + error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" + error_file.write_text(error_output, encoding="utf-8") + return False + except NotImplementedError as ex: + tb = traceback.format_exc() + error_msg = str(ex).split(">>>")[0].strip() + + rel_path = example_file.relative_to(base_path) if base_path else example_file + logger.warning(f"[JSON] Unimplemented behavior on {rel_path}: {error_msg}") + + if output_dir: + error_dir = output_dir / "NotImplementedError" / sanitize_name(error_msg) + error_dir.mkdir(parents=True, exist_ok=True) + error_file = error_dir / f"{sanitize_name(example_file.stem)}.txt" + + json_content = example_file.read_text(encoding="utf-8") + error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" + error_file.write_text(error_output, encoding="utf-8") + return True + +def main(example_path_str: str, output_path_str: Optional[str])-> None: + example_path = Path(example_path_str) + json_example_dir = example_path / "json" / "examples" + xml_example_dir = example_path / "xml" / "examples" + + output_path = None + if output_path_str: + output_path = Path(output_path_str) + output_path.mkdir(parents=True, exist_ok=True) + + total = 0 + failed = 0 + for json_test in json_example_dir.glob("**/*.json"): + total += 1 + success = test_json_example(json_test, json_example_dir, output_path) + if not success: + failed += 1 + + if failed == 0: + logger.info(f"No JSON tests out of {total:d} failed") + else: + logger.error(f"Failed {failed:d}/{total:d} json tests") + sys.exit(failed) + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument("--examples", required=True, help="path to examples directory") + parser.add_argument("--output", required=False, help="path to output directory") + args = parser.parse_args() + + main(args.examples, args.output) From 5bb304016a312cbb8f874c9be7c77242b1b652c4 Mon Sep 17 00:00:00 2001 From: hpoeche Date: Thu, 30 Jul 2026 12:45:39 +0200 Subject: [PATCH 2/4] Draft: only basic JSON test handling For now only the JSON examples are considered. Errors are sorted into three categories: Errors that occur from deserialization, NotImplementedErrors that are explicit developer choices and Unexpected errors to cover future errors of the sdk. --- sdk/test/testdatagen/example_tests.py | 31 ++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/sdk/test/testdatagen/example_tests.py b/sdk/test/testdatagen/example_tests.py index cecfe1f4..6afdc0da 100644 --- a/sdk/test/testdatagen/example_tests.py +++ b/sdk/test/testdatagen/example_tests.py @@ -37,7 +37,9 @@ def test_json_example(example_file: Path, base_path: Optional[Path] = None, outp if output_dir: error_dir = output_dir / sanitize_name(error_msg) error_dir.mkdir(parents=True, exist_ok=True) - error_file = error_dir / f"{sanitize_name(example_file.stem)}.txt" + # Use the full relative path (not just the stem) to avoid collisions: generated example sets reuse + # the same filenames (e.g. "fuzzed_01.json") across many different subdirectories. + error_file = error_dir / f"{sanitize_name(str(rel_path))}.txt" json_content = example_file.read_text(encoding="utf-8") error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" @@ -48,17 +50,40 @@ def test_json_example(example_file: Path, base_path: Optional[Path] = None, outp error_msg = str(ex).split(">>>")[0].strip() rel_path = example_file.relative_to(base_path) if base_path else example_file - logger.warning(f"[JSON] Unimplemented behavior on {rel_path}: {error_msg}") + logger.warning(f"[JSON] NOT_IMPLEMENTED on {rel_path}: {error_msg}") if output_dir: error_dir = output_dir / "NotImplementedError" / sanitize_name(error_msg) error_dir.mkdir(parents=True, exist_ok=True) - error_file = error_dir / f"{sanitize_name(example_file.stem)}.txt" + # Use the full relative path (not just the stem) to avoid collisions: generated example sets reuse + # the same filenames (e.g. "fuzzed_01.json") across many different subdirectories. + error_file = error_dir / f"{sanitize_name(str(rel_path))}.txt" json_content = example_file.read_text(encoding="utf-8") error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" error_file.write_text(error_output, encoding="utf-8") return True + except Exception as ex: + # Catch-all so an exception type not (yet) anticipated by the except clauses above (e.g. thrown by a + # future implementation change) fails just this one example instead of aborting the whole run. + # Deliberately not `except BaseException`, so KeyboardInterrupt/SystemExit still propagate. + tb = traceback.format_exc() + error_msg = str(ex).split(">>>")[0].strip() + + rel_path = example_file.relative_to(base_path) if base_path else example_file + logger.error(f"[JSON] UNEXPECTED {type(ex).__name__} on {rel_path}: {error_msg}") + + if output_dir: + error_dir = output_dir / "UnexpectedError" / sanitize_name(type(ex).__name__) / sanitize_name(error_msg) + error_dir.mkdir(parents=True, exist_ok=True) + # Use the full relative path (not just the stem) to avoid collisions: generated example sets reuse + # the same filenames (e.g. "fuzzed_01.json") across many different subdirectories. + error_file = error_dir / f"{sanitize_name(str(rel_path))}.txt" + + json_content = example_file.read_text(encoding="utf-8") + error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" + error_file.write_text(error_output, encoding="utf-8") + return False def main(example_path_str: str, output_path_str: Optional[str])-> None: example_path = Path(example_path_str) From e7fc1a0f9b90c7ebd6462022cf05c1ba76ccef0a Mon Sep 17 00:00:00 2001 From: paul-gerber-svg Date: Thu, 30 Jul 2026 15:21:40 +0200 Subject: [PATCH 3/4] Add XML testing and refactor test logic into a generic runner function toeliminate code duplication --- sdk/test/testdatagen/example_tests.py | 186 ++++++++++++++++---------- 1 file changed, 115 insertions(+), 71 deletions(-) diff --git a/sdk/test/testdatagen/example_tests.py b/sdk/test/testdatagen/example_tests.py index 6afdc0da..b13ee166 100644 --- a/sdk/test/testdatagen/example_tests.py +++ b/sdk/test/testdatagen/example_tests.py @@ -4,115 +4,159 @@ import sys import traceback from pathlib import Path -from typing import Callable, Literal, Optional +from typing import Optional from basyx.aas import adapter from basyx.aas.model import AASConstraintViolation logger = logging.getLogger(__name__) + def sanitize_name(name: str, max_len: int = 120) -> str: - name = re.sub(r'[<>:"/\\|?*\x00-\x1f\s]', '_', name) - return name[:max_len] + pattern = r'[<>:"/\\|?*\x00-\x1f\s\'{}\(\)\[\]]' + name = re.sub(pattern, '_', name) + name = re.sub(r'_+', '_', name) + return name.strip('_')[:max_len] -def test_json_example(example_file: Path, base_path: Optional[Path] = None, output_dir: Optional[Path] = None) -> bool: - """ - Attempts to read and deserialize an example file. Reports the status and saves information to output path. - :param example_file: File which contains the example that should be deserialized - :param deserializer: Function that is used to deserialize the example - :param output_dir: Path to store the failed outputs - :return: bool that indicates if the example was successfully deserialized - """ + +def write_error_report(file_type, example_file, rel_path, error_dir, tb): + error_dir.mkdir(parents=True, exist_ok=True) + # Use the full relative path (not just the stem) to avoid collisions: generated example sets reuse + # the same filenames across many different subdirectories. + error_file = error_dir / f"{sanitize_name(str(rel_path))}.txt" + + content = example_file.read_text(encoding="utf-8") + error_output = f"=== {file_type.upper()} File: {rel_path} ===\n{content}\n\n=== Stacktrace ====\n{tb}" + error_file.write_text(error_output, encoding="utf-8") + + +def run_example_test(file_type, example_file, read_fn, base_path=None, output_dir=None, is_xml=False): + rel_path = example_file.relative_to(base_path) if base_path else example_file + try: - adapter.json.read_aas_json_file(example_file, failsafe=False) + read_fn(example_file) return True except (KeyError, TypeError, ValueError, AASConstraintViolation) as ex: tb = traceback.format_exc() - error_msg = str(ex).split(">>>")[0].strip() - - rel_path = example_file.relative_to(base_path) if base_path else example_file - logger.error(f"[JSON] ERROR on {rel_path}: {error_msg}") + error_msg = extract_error_message(ex) if is_xml else str(ex).split(">>>")[0].strip() + logger.error(f"[{file_type.upper()}] ERROR on {rel_path}: {error_msg}") if output_dir: - error_dir = output_dir / sanitize_name(error_msg) - error_dir.mkdir(parents=True, exist_ok=True) - # Use the full relative path (not just the stem) to avoid collisions: generated example sets reuse - # the same filenames (e.g. "fuzzed_01.json") across many different subdirectories. - error_file = error_dir / f"{sanitize_name(str(rel_path))}.txt" - - json_content = example_file.read_text(encoding="utf-8") - error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" - error_file.write_text(error_output, encoding="utf-8") + error_dir = output_dir / file_type / sanitize_name(error_msg) + write_error_report(file_type, example_file, rel_path, error_dir, tb) return False + except NotImplementedError as ex: tb = traceback.format_exc() - error_msg = str(ex).split(">>>")[0].strip() - - rel_path = example_file.relative_to(base_path) if base_path else example_file - logger.warning(f"[JSON] NOT_IMPLEMENTED on {rel_path}: {error_msg}") + error_msg = extract_error_message(ex) if is_xml else str(ex).split(">>>")[0].strip() + logger.warning(f"[{file_type.upper()}] NOT_IMPLEMENTED on {rel_path}: {error_msg}") if output_dir: - error_dir = output_dir / "NotImplementedError" / sanitize_name(error_msg) - error_dir.mkdir(parents=True, exist_ok=True) - # Use the full relative path (not just the stem) to avoid collisions: generated example sets reuse - # the same filenames (e.g. "fuzzed_01.json") across many different subdirectories. - error_file = error_dir / f"{sanitize_name(str(rel_path))}.txt" - - json_content = example_file.read_text(encoding="utf-8") - error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" - error_file.write_text(error_output, encoding="utf-8") + error_dir = output_dir / file_type / "NotImplementedError" / sanitize_name(error_msg) + write_error_report(file_type, example_file, rel_path, error_dir, tb) return True + except Exception as ex: # Catch-all so an exception type not (yet) anticipated by the except clauses above (e.g. thrown by a # future implementation change) fails just this one example instead of aborting the whole run. # Deliberately not `except BaseException`, so KeyboardInterrupt/SystemExit still propagate. tb = traceback.format_exc() - error_msg = str(ex).split(">>>")[0].strip() - - rel_path = example_file.relative_to(base_path) if base_path else example_file - logger.error(f"[JSON] UNEXPECTED {type(ex).__name__} on {rel_path}: {error_msg}") + error_msg = extract_error_message(ex) if is_xml else str(ex).split(">>>")[0].strip() + logger.error(f"[{file_type.upper()}] UNEXPECTED {type(ex).__name__} on {rel_path}: {error_msg}") if output_dir: - error_dir = output_dir / "UnexpectedError" / sanitize_name(type(ex).__name__) / sanitize_name(error_msg) - error_dir.mkdir(parents=True, exist_ok=True) - # Use the full relative path (not just the stem) to avoid collisions: generated example sets reuse - # the same filenames (e.g. "fuzzed_01.json") across many different subdirectories. - error_file = error_dir / f"{sanitize_name(str(rel_path))}.txt" - - json_content = example_file.read_text(encoding="utf-8") - error_output = f"=== JSON File: {rel_path} ===\n{json_content}\n\n=== Stacktrace ====\n{tb}" - error_file.write_text(error_output, encoding="utf-8") + error_dir = ( + output_dir + / file_type + / "UnexpectedError" + / sanitize_name(type(ex).__name__) + / sanitize_name(error_msg) + ) + write_error_report(file_type, example_file, rel_path, error_dir, tb) return False -def main(example_path_str: str, output_path_str: Optional[str])-> None: - example_path = Path(example_path_str) - json_example_dir = example_path / "json" / "examples" - xml_example_dir = example_path / "xml" / "examples" +def extract_error_message(ex): + message = str(ex) + cause = ex.__cause__ + while cause is not None: + message = f"{message} -> {cause}" + cause = cause.__cause__ + return re.sub(r'on line [0-9]+', "", message.split("->")[-1]).strip() + - output_path = None - if output_path_str: - output_path = Path(output_path_str) +def test_json_example(example_file: Path, base_path: Optional[Path] = None, output_dir: Optional[Path] = None) -> bool: + """ + Attempts to read and deserialize an example file. Reports the status and saves information to output path. + + :param example_file: File which contains the example that should be deserialized + :param base_path: Base directory to compute relative paths + :param output_dir: Path to store the failed outputs + :return: bool that indicates if the example was successfully deserialized + """ + return run_example_test( + file_type="json", + example_file=example_file, + read_fn=lambda f: adapter.json.read_aas_json_file(f, failsafe=False), + base_path=base_path, + output_dir=output_dir, + ) + + +def test_xml_example(example_file: Path, base_path: Optional[Path] = None, output_dir: Optional[Path] = None) -> bool: + """ + Attempts to read and deserialize an XML example file. Reports the status and saves information to output path. + + :param example_file: File which contains the example that should be deserialized + :param base_path: Base directory to compute relative paths + :param output_dir: Path to store the failed outputs + :return: bool that indicates if the example was successfully deserialized + """ + return run_example_test( + file_type="xml", + example_file=example_file, + read_fn=lambda f: adapter.xml.read_aas_xml_file(f, failsafe=False), + base_path=base_path, + output_dir=output_dir, + is_xml=True, + ) + + +def main(example_path_str: str, output_path_str: Optional[str]) -> None: + example_path = Path(example_path_str) + output_path = Path(output_path_str) if output_path_str else None + if output_path: output_path.mkdir(parents=True, exist_ok=True) - total = 0 - failed = 0 - for json_test in json_example_dir.glob("**/*.json"): - total += 1 - success = test_json_example(json_test, json_example_dir, output_path) - if not success: - failed += 1 + test_configs = [ + ("JSON", example_path / "json" / "examples" / "generated", "*.json", test_json_example), + ("XML", example_path / "xml" / "examples" / "generated", "*.xml", test_xml_example), + ] + + total_failed = 0 + for name, directory, glob_pattern, test_fn in test_configs: + total = 0 + failed = 0 + for test_file in sorted(directory.glob(f"**/{glob_pattern}")): + total += 1 + if not test_fn(test_file, directory, output_path): + failed += 1 + + if failed == 0: + logger.info(f"No {name} tests out of {total:d} failed") + else: + logger.error(f"Failed {failed:d}/{total:d} {name.lower()} tests") + + total_failed += failed + + sys.exit(total_failed) - if failed == 0: - logger.info(f"No JSON tests out of {total:d} failed") - else: - logger.error(f"Failed {failed:d}/{total:d} json tests") - sys.exit(failed) if __name__ == '__main__': + logging.basicConfig(level=logging.INFO) parser = argparse.ArgumentParser() parser.add_argument("--examples", required=True, help="path to examples directory") parser.add_argument("--output", required=False, help="path to output directory") args = parser.parse_args() - main(args.examples, args.output) + main(args.examples, args.output) \ No newline at end of file From e2e1905bacab40642d41417c34fb401fb21e388e Mon Sep 17 00:00:00 2001 From: hpoeche Date: Sat, 1 Aug 2026 16:22:12 +0200 Subject: [PATCH 4/4] Add sdk-testdatagen compsite action We add a composite github action `sdk-test-testdatagen` that fetches the example files from the `aas-specs-metmodel` repository in the correct verion, runs the `example_tests.py` script and uploads the output as an artifact to github. For testing this action is temporarily called in the ci, that runs on every pull_request. This should be changed to only run on PRs into main before this PR is approved. --- .../actions/sdk-test-testdatagen/action.yml | 33 +++++++++++++++++++ .github/workflows/ci.yml | 16 +++++++++ 2 files changed, 49 insertions(+) create mode 100644 .github/actions/sdk-test-testdatagen/action.yml diff --git a/.github/actions/sdk-test-testdatagen/action.yml b/.github/actions/sdk-test-testdatagen/action.yml new file mode 100644 index 00000000..0101e906 --- /dev/null +++ b/.github/actions/sdk-test-testdatagen/action.yml @@ -0,0 +1,33 @@ +name: sdk-test-testdatagen +author: hpoeche + +description: Fetches generated example files from the aas-specs-metamodel repository and tries to deserialize them + +inputs: + AAS_SPECS_RELEASE_TAG: + required: true + description: A tag on the aas-specs-metamodel repository which defines the version of metamodel to test + +runs: + using: 'composite' + steps: + - name: Clone aas-specs-metamodel repository + shell: bash + env: + AAS_SPECS_RELEASE_TAG: ${{ inputs.AAS_SPECS_RELEASE_TAG }} + run: | + git clone https://github.com/admin-shell-io/aas-specs-metamodel.git + cd aas-specs-metamodel + git checkout $AAS_SPECS_RELEASE_TAG + + - name: Run tests on example files + shell: bash + run: | + python sdk/test/testdatagen/example_tests.py --examples aas-specs-metamodel/schemas --output testdata-output + + - name: Upload test output + if: ${{ always() }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a #v7.0.1 + with: + path: testdata-output + name: sdk_test_testdatagen_results \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e71f1b5e..40539c42 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -339,3 +339,19 @@ jobs: run: | docker stop basyx-python-server && docker rm basyx-python-server + tmp-sdk-testdatagen: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Set up Python ${{ env.X_PYTHON_MIN_VERSION }} + uses: actions/setup-python@v5 + with: + python-version: ${{ env.X_PYTHON_MIN_VERSION }} + - name: Install Python dependencies + working-directory: ./sdk + run: | + python -m pip install --upgrade pip + python -m pip install .[dev] + - uses: ./.github/actions/sdk-test-testdatagen + with: + AAS_SPECS_RELEASE_TAG: v3.1.2 \ No newline at end of file