Integrate new test gen into input model

Signed-off-by: Tyler Gu <[email protected]>

acto/__init__.py

@@ -0,0 +1 @@
+DEFAULT_KUBERNETES_VERSION = "v1.27.0"

acto/checker/impl/consistency.py

@@ -1,4 +1,5 @@
 """State checker"""
+
 import copy
 import json
 import re
@@ -20,6 +21,7 @@
 )
 from acto.input import InputModel
 from acto.input.get_matched_schemas import find_matched_schema
+from acto.input.property_attribute import PropertyAttribute
 from acto.k8s_util.k8sutil import canonicalize_quantity
 from acto.result import ConsistencyOracleResult, InvalidInputResult
 from acto.schema import ArraySchema, BaseSchema, ObjectSchema, extract_schema
@@ -395,7 +397,12 @@ def check(
                 )
 
                 should_compare = (
-                    should_compare or corresponding_schema.mapped
+                    should_compare
+                    or (
+                        corresponding_schema.attributes
+                        & PropertyAttribute.Mapped
+                        == PropertyAttribute.Mapped
+                    )
                 ) and must_produce_delta
 
                 # Policy: pass if any of the matched deltas is equivalent

acto/checker/impl/tests/test_state.py

@@ -44,7 +44,6 @@ def input_model_and_context_mapping() -> (
         input_model = DeterministicInputModel(
             crd=context["crd"]["body"],
             seed_input=defaultdict(lambda: defaultdict(dict)),
-            used_fields=[],
             example_dir=None,
             num_workers=1,
             num_cases=1,
@@ -57,6 +56,7 @@ def input_model_and_context_mapping() -> (
 
 
 def checker_func(s: Snapshot, prev_s: Snapshot) -> Optional[OracleResult]:
+    """Run the consistency checker and return the result."""
     api_version = s.input_cr["apiVersion"]
     checker = ConsistencyChecker(
         trial_dir="",
@@ -107,6 +107,7 @@ def checker_func(s: Snapshot, prev_s: Snapshot) -> Optional[OracleResult]:
     ),
 )
 def test_consistency_checker(test_case_id, result_dict):
+    """Test the consistency checker."""
     snapshot = load_snapshot("state", test_case_id)
     snapshot_prev = load_snapshot("state", test_case_id, load_prev=True)
     oracle_result = checker_func(snapshot, snapshot_prev)

acto/cli/schema_match.py

@@ -5,6 +5,7 @@
 from ruamel.yaml import YAML
 
 from acto.input.k8s_schemas import K8sSchemaMatcher
+from acto.schema.object import ObjectSchema
 from acto.schema.schema import extract_schema
 
 
@@ -22,7 +23,7 @@ def main():
     parser.add_argument(
         "--k8s-version",
         required=False,
-        default="1.29",
+        default="1.29.0",
         help="Kubernetes version to match the schema with",
     )
     parser.add_argument(
@@ -43,7 +44,7 @@ def main():
 
     # match the schema with Kubernetes resource schemas
     schema_matcher = K8sSchemaMatcher.from_version(args.k8s_version)
-    matches = schema_matcher.find_matched_schemas(root)
+    matches = schema_matcher.find_all_matched_schemas(root)
 
     # output the breakdown of the matched schema information
     df = pd.DataFrame(
@@ -53,6 +54,7 @@ def main():
                 "schema_path": "/".join(schema.path),
             }
             for schema, k8s_schema in matches
+            if k8s_schema.k8s_schema_name is not None
         ]
     )
 
@@ -61,14 +63,16 @@ def main():
 
     # annotate the yaml file with the matched schema information
     for schema, k8s_schema in matches:
+        if k8s_schema.k8s_schema_name is None:
+            continue
         comment = k8s_schema.k8s_schema_name
         curr = schema_yaml
         for segment in schema.path[:-1]:
             if segment == "ITEM":
                 curr = curr["items"]
             else:
                 curr = curr["properties"][segment]
-        if schema.path[-1] != "ITEM":
+        if schema.path[-1] != "ITEM" and isinstance(schema, ObjectSchema):
             curr["properties"].yaml_add_eol_comment(comment, schema.path[-1])
         else:
             curr.yaml_add_eol_comment(comment, "items")

acto/common.py

@@ -5,7 +5,7 @@
 import random
 import re
 import string
-from typing import Any, Tuple, TypeAlias, Union
+from typing import Any, Sequence, Tuple, TypeAlias, Union
 
 import deepdiff.model as deepdiff_model
 import kubernetes
@@ -22,7 +22,7 @@ class PropertyPath(pydantic.BaseModel):
 
     path: list[PathSegment]
 
-    def __init__(self, path: list[PathSegment]) -> None:
+    def __init__(self, path: Sequence[PathSegment]) -> None:
         """Override constructor to allow positional argument"""
         super().__init__(path=path)
 

acto/engine.py

@@ -16,19 +16,15 @@
 import jsonpatch
 import yaml
 
-from acto.acto_config import ACTO_CONFIG
 from acto.checker.checker_set import CheckerSet
 from acto.common import kubernetes_client, print_event
 from acto.constant import CONST
 from acto.deploy import Deploy
 from acto.input import InputModel
-from acto.input.input import DeterministicInputModel, OverSpecifiedField
-from acto.input.known_schemas.base import K8sField
-from acto.input.known_schemas.known_schema import find_all_matched_schemas_type
+from acto.input.input import DeterministicInputModel
 from acto.input.testcase import TestCase
 from acto.input.testplan import TestGroup
 from acto.input.value_with_schema import ValueWithSchema, attach_schema_to_value
-from acto.input.valuegenerator import ArrayGenerator
 from acto.kubectl_client import KubectlClient
 from acto.kubernetes_engine import base, kind
 from acto.lib.operator_config import OperatorConfig
@@ -825,72 +821,17 @@ def __init__(
         if preload_images_ is not None:
             self.context["preload_images"].update(preload_images_)
 
-        # Apply custom fields
-        if operator_config.analysis is not None:
-            used_fields = self.context["analysis_result"]["used_fields"]
-        else:
-            used_fields = None
         self.input_model: DeterministicInputModel = input_model(
             crd=self.context["crd"]["body"],
             seed_input=self.seed,
-            used_fields=used_fields,
             example_dir=operator_config.example_dir,
             num_workers=num_workers,
             num_cases=num_cases,
             mount=mount,
+            kubernetes_version=operator_config.kubernetes_version,
+            custom_module_path=operator_config.custom_module,
         )
 
-        applied_custom_k8s_fields = False
-
-        if operator_config.k8s_fields is not None:
-            module = importlib.import_module(operator_config.k8s_fields)
-            if hasattr(module, "BLACKBOX") and ACTO_CONFIG.mode == "blackbox":
-                applied_custom_k8s_fields = True
-                for k8s_field in module.BLACKBOX:
-                    self.input_model.apply_k8s_schema(k8s_field)
-            elif hasattr(module, "WHITEBOX") and ACTO_CONFIG.mode == "whitebox":
-                applied_custom_k8s_fields = True
-                for k8s_field in module.WHITEBOX:
-                    self.input_model.apply_k8s_schema(k8s_field)
-        if not applied_custom_k8s_fields:
-            # default to use the known_schema module to automatically find the mapping
-            # from CRD to K8s schema
-            logger.info(
-                "Using known_schema to find the mapping from CRD to K8s schema"
-            )
-            tuples = find_all_matched_schemas_type(self.input_model.root_schema)
-            for match_tuple in tuples:
-                logger.debug(
-                    "Found matched schema: %s -> %s",
-                    match_tuple[0].path,
-                    match_tuple[1],
-                )
-                k8s_schema = K8sField(match_tuple[0].path, match_tuple[1])
-                self.input_model.apply_k8s_schema(k8s_schema)
-
-        if operator_config.custom_fields is not None:
-            if ACTO_CONFIG.mode == "blackbox":
-                pruned_list = []
-                module = importlib.import_module(operator_config.custom_fields)
-                for custom_field in module.custom_fields:
-                    pruned_list.append(custom_field.path)
-                    self.input_model.apply_custom_field(custom_field)
-            else:
-                pruned_list = []
-                module = importlib.import_module(operator_config.custom_fields)
-                for custom_field in module.custom_fields:
-                    pruned_list.append(custom_field.path)
-                    self.input_model.apply_custom_field(custom_field)
-        else:
-            pruned_list = []
-            tuples = find_all_matched_schemas_type(self.input_model.root_schema)
-            for match_tuple in tuples:
-                custom_field = OverSpecifiedField(
-                    match_tuple[0].path,
-                    array=isinstance(match_tuple[1], ArrayGenerator),
-                )
-                self.input_model.apply_custom_field(custom_field)
-
         self.sequence_base = 20 if delta_from else 0
 
         if operator_config.custom_oracle is not None: