Add test data generation

README.md

@@ -128,14 +128,29 @@ print(api.supported_versions())
 print(api.latest_version())
 ```
 
+## Generating test data for software testing
+
+If you develop an AAS application like an AAS editor you may want to use test data to verify correctness of your application.
+The test engines allow to generate a set of AAS files which are compliant with the standard and you can therefore use to assess your application as follows:
+
+```python
+from aas_test_engines import file
+
+for sample in file.generate():
+    print(sample) # or whatever you want to do with it
+```
+
 ## Command line interface
 
 You may want to invoke the test tools using the simplified command line interface:
 
 ```sh
 # Check file
-python -m aas_test_engines file test.aasx
+python -m aas_test_engines check_file test.aasx
 
 # Check server
-python -m aas_test_engines api https://localhost --suite 'Asset Administration Shell API'
+python -m aas_test_engines check_server https://localhost --suite 'Asset Administration Shell API'
+
+# Generate test data
+python -m aas_test_engines generate_files output_dir
 ```

aas_test_engines/__main__.py

@@ -1,8 +1,10 @@
 import argparse
 import sys
+import os
 from aas_test_engines import api, file
 from enum import Enum
 
+
 class Formats(Enum):
     xml = 'xml'
     json = 'json'
@@ -11,6 +13,7 @@ class Formats(Enum):
     def __str__(self):
         return self.value
 
+
 def run_file_test(argv):
     parser = argparse.ArgumentParser(description='Checks a file for compliance with the AAS meta-model')
     parser.add_argument('file',
@@ -59,24 +62,44 @@ def run_api_test(argv):
         suites = None
     tests = api.generate_tests(suites=suites)
     exec_conf = api.run.ExecConf(
-        server = args.server,
-        dry = args.dry,
-        verify = not args.no_verify,
+        server=args.server,
+        dry=args.dry,
+        verify=not args.no_verify,
     )
     for result in api.execute_tests(tests, exec_conf):
         result.dump()
 
 
+def generate_files(argv):
+    parser = argparse.ArgumentParser(description='Generates aas files which can be used to test your software')
+    parser.add_argument('directory',
+                        type=str,
+                        help='Directory to place files in')
+    args = parser.parse_args(argv)
+    if os.path.exists(args.directory):
+        print(f"Directory '{args.directory}' already exists, please remove it")
+        exit(1)
+    os.mkdir(args.directory)
+    i = 0
+    for sample in file.generate():
+        with open(os.path.join(args.directory, f"{i}.json"), "w") as f:
+            f.write(sample)
+        i += 1
+        if i > 100:
+            break
+
 commands = {
-    'file': run_file_test,
-    'api': run_api_test,
+    'check_file': run_file_test,
+    'check_server': run_api_test,
+    'generate_files': generate_files,
 }
 
 if len(sys.argv) <= 1:
     print(f"Usage: {sys.argv[0]} COMMAND OPTIONS...")
     print("Available commands:")
-    print("  file     Check a file for compliance.")
-    print("  api      Check a server instance for compliance.")
+    print("  check_file      Check a file for compliance.")
+    print("  check_server    Check a server instance for compliance.")
+    print("  generate_files  Generate files for testing")
     exit(1)
 
 command = sys.argv[1]

aas_test_engines/_file/generate.py

@@ -0,0 +1,15 @@
+from fences.json_schema.normalize import normalize
+from fences.json_schema.parse import default_config
+from fences.json_schema.parse import parse
+from fences.core.node import Node as FlowGraph
+
+
+def generate_graph(schema) -> FlowGraph:
+    # TODO: remove these after fixing fences.core.exception.InternalException: Decision without valid leaf detected
+    del schema['$defs']['AssetInformation']['allOf'][1]['properties']['specificAssetIds']['items']['allOf'][0]
+    del schema['$defs']['Entity']['allOf'][1]
+    schema_norm = normalize(schema, False)
+    config = default_config()
+    config.normalize = False
+    graph = parse(schema_norm, config)
+    return graph

aas_test_engines/file.py

@@ -1,11 +1,12 @@
-from typing import List, Dict, TextIO, Union, Any, Set, Optional
+from typing import List, Dict, TextIO, Union, Any, Set, Optional, Generator
 import os
 import json
 from yaml import safe_load
 
 from .exception import AasTestToolsException
 from .result import AasTestResult, Level
 from .data_types import validators
+from ._file.generate import generate_graph, FlowGraph
 
 from xml.etree import ElementTree
 from json_schema_tool.schema import SchemaValidator, SchemaValidator, ValidationConfig, ParseConfig, SchemaValidationResult, KeywordValidationResult, parse_schema
@@ -18,8 +19,9 @@
 
 class AasSchema:
 
-    def __init__(self, validator: SchemaValidator):
+    def __init__(self, validator: SchemaValidator, graph: FlowGraph):
         self.validator = validator
+        self.graph = graph
 
 
 def _find_schemas() -> Dict[str, any]:
@@ -35,7 +37,8 @@ def _find_schemas() -> Dict[str, any]:
             format_validators=validators
         )
         validator = parse_schema(schema, config)
-        result[i[:-4]] = AasSchema(validator)
+        graph = generate_graph(schema)
+        result[i[:-4]] = AasSchema(validator, graph)
     return result
 
 
@@ -303,3 +306,10 @@ def check_aasx_file(file: TextIO, version: str = _DEFAULT_VERSION) -> AasTestRes
         return AasTestResult(f"Cannot read: {e}", level=Level.ERROR)
 
     return check_aasx_data(zip, version)
+
+
+def generate(version: str = _DEFAULT_VERSION) -> Generator[str, None, None]:
+    graph = _get_schema(version).graph
+    for i in graph.generate_paths():
+        sample = graph.execute(i.path)
+        yield json.dumps(sample)

requirements.txt

@@ -2,3 +2,4 @@ requests>=2.31
 pyyaml>=6.0
 json_schema_tool>=0.2
 jsonpath_ng>=1.5
+fences==1.1.0

test/test_file.py

@@ -52,6 +52,7 @@ def test_no_xml(self):
         result = file.check_xml_file(io.StringIO("no xml"))
         self.assertFalse(result.ok())
 
+
 class CheckAasxTest(TestCase):
 
     def test_not_a_zip(self):
@@ -126,3 +127,13 @@ def test_invoke(self):
         for i in s:
             print(i)
         self.assertIn(file.latest_version(), s)
+
+
+class GenerateTest(TestCase):
+
+    def test_a_few(self):
+        i = 0
+        for sample in file.generate():
+            i += 1
+            if i > 100:
+                break