Fix argument mismatch and pydantic constraint error (#300)

* Fix argument mismatch and pydantic constraint error

* Migrate away from pydantic min_items

* Fix on-disk system state filenames

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

---------

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

acto/__main__.py

@@ -141,7 +141,6 @@
 acto = Acto(workdir_path=args.workdir_path,
             operator_config=config,
             cluster_runtime=args.cluster_runtime,
-            enable_analysis=False,
             preload_images_=args.preload_images,
             context_file=context_cache,
             helper_crd=args.helper_crd,

acto/checker/checker_set.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+from typing import Optional
 
 from acto.checker.impl.crash import CrashChecker
 from acto.checker.impl.health import HealthChecker
@@ -25,7 +26,7 @@ def __init__(
         trial_dir: str,
         input_model: InputModel,
         oracle_handle: OracleHandle,
-        checker_generators: list = None,
+        checker_generators: Optional[list] = None,
     ):
         checker_generators = [
             CrashChecker,
@@ -87,7 +88,7 @@ def check(
             )
 
         generation_result_path = os.path.join(
-            self.trial_dir, f"generation-{generation}-runtime.json"
+            self.trial_dir, f"generation-{generation:03d}-runtime.json"
         )
         with open(generation_result_path, "w", encoding="utf-8") as f:
             json.dump(run_result.to_dict(), f, cls=ActoEncoder, indent=4)

acto/checker/impl/state.py

@@ -232,7 +232,7 @@ def __init__(
 
     def write_delta_log(self, generation: int, input_delta, system_delta):
         """Write the input delta and system delta to a log file"""
-        delta_log_path = f"{self.trial_dir}/delta-{generation}.log"
+        delta_log_path = f"{self.trial_dir}/delta-{generation:03d}.log"
         with open(delta_log_path, "w", encoding="utf-8") as f:
             f.write("---------- INPUT DELTA  ----------\n")
             f.write(json.dumps(input_delta, cls=ActoEncoder, indent=6))

acto/checker/test_checker.py

@@ -4,7 +4,7 @@
 import sys
 
 from acto.checker.checker_set import CheckerSet
-from acto.common import RunResult, RecoveryResult, ErrorResult, Oracle
+from acto.common import ErrorResult, Oracle, RecoveryResult, RunResult
 from acto.input import InputModel
 from acto.serialization import ActoEncoder
 from acto.snapshot import Snapshot
@@ -74,12 +74,12 @@ def check_trial_worker(workqueue: multiprocessing.Queue, checker_class: type,
 
             alarm = False
             for generation in range(0, 20):
-                mutated_filename = '%s/mutated-%d.yaml' % (trial_dir, generation)
-                operator_log_path = "%s/operator-%d.log" % (trial_dir, generation)
-                system_state_path = "%s/system-state-%03d.json" % (trial_dir, generation)
-                events_log_path = "%s/events-%d.json" % (trial_dir, generation)
-                cli_output_path = "%s/cli-output-%d.log" % (trial_dir, generation)
-                runtime_result_path = "%s/generation-%d-runtime.json" % (trial_dir, generation)
+                events_log_path = f"{trial_dir}/events-{generation:03d}.json"
+                mutated_filename = f"{trial_dir}/mutated-{generation:03d}.yaml"
+                operator_log_path = f"{trial_dir}/operator-{generation:03d}.log"
+                system_state_path = f"{trial_dir}/system-state-{generation:03d}.json"
+                cli_output_path = f"{trial_dir}/cli-output-{generation:03d}.log"
+                runtime_result_path = f"{trial_dir}/generation-{generation:03d}-runtime.json"
 
                 if not os.path.exists(mutated_filename):
                     break

acto/lib/operator_config.py

@@ -45,7 +45,7 @@ class DeployConfig(BaseModel, extra="forbid"):
     """Configuration for deploying the operator"""
     steps: List[DeployStep] = Field(
         description="Steps to deploy the operator",
-        min_length=1,)
+        min_length=1)
 
 
 class AnalysisConfig(BaseModel, extra="forbid"):

acto/post_process/post_process.py

@@ -1,82 +1,96 @@
+"""Post process the testrun results"""
 import glob
 import hashlib
 import json
 import os
-import sys
 from abc import abstractmethod
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 import yaml
 
-sys.path.append('.')
-sys.path.append('..')
-
-from acto.common import RunResult, oracle_result_from_dict
+from acto.common import RunResult
 from acto.lib.operator_config import OperatorConfig
 
 
 class Step:
-    '''A step in a trial
-    
+    """A step in a trial
+
     Attributes:
         input (Dict): input to the operator
         operator_log (str): operator log
         system_state (Dict): system state
         cli_output (str): cli output
         runtime_result (Dict): runtime result
-    '''
-
-    def __init__(self, trial_dir: str, gen: int, input: Dict, operator_log: str, system_state: Dict,
-                 cli_output: str, runtime_result: Dict):
+    """
+
+    def __init__(
+        self,
+        trial_dir: str,
+        gen: int,
+        input_cr: Dict,
+        operator_log: list[str],
+        system_state: Dict,
+        cli_output: str,
+        runtime_result: Dict,
+    ):
         self._trial_dir = trial_dir
         self._gen = gen
-        self._input = input
-        self._input_digest = hashlib.md5(json.dumps(input,
-                                                    sort_keys=True).encode("utf-8")).hexdigest()
+        self._input = input_cr
+        self._input_digest = hashlib.md5(
+            json.dumps(input_cr, sort_keys=True).encode("utf-8")
+        ).hexdigest()
         self._operator_log = operator_log
         self._system_state = system_state
         self._cli_output = cli_output
         self._runtime_result = RunResult.from_dict(runtime_result)
 
     @property
     def trial_dir(self) -> str:
+        """Return trial directory"""
         return self._trial_dir
 
     @property
     def gen(self) -> int:
+        """Return generation"""
         return self._gen
 
     @property
     def input(self) -> Dict:
+        """Return input"""
         return self._input
 
     @property
     def input_digest(self) -> str:
+        """Return input digest"""
         return self._input_digest
 
     @property
-    def operator_log(self) -> str:
+    def operator_log(self) -> list[str]:
+        """Return operator log"""
         return self._operator_log
 
     @property
     def system_state(self) -> Dict:
+        """Return system state"""
         return self._system_state
 
     @property
     def cli_output(self) -> str:
+        """Return cli output"""
         return self._cli_output
 
     @property
     def runtime_result(self) -> RunResult:
+        """Return runtime result"""
         return self._runtime_result
 
 
 def read_trial_dir(trial_dir: str) -> Dict[str, Step]:
-    '''Read a trial directory and return a list of steps'''
+    """Read a trial directory and return a list of steps"""
 
     steps: Dict[str, Step] = {}
     for generation in range(0, 20):
-        if not os.path.exists('%s/mutated-%d.yaml' % (trial_dir, generation)):
+        if not os.path.exists(f"{trial_dir}/mutated-{generation:03d}.yaml"):
             break
 
         step = construct_step(trial_dir, generation)
@@ -88,61 +102,93 @@ def read_trial_dir(trial_dir: str) -> Dict[str, Step]:
     return steps
 
 
-def construct_step(trial_dir, generation) -> Step:
-    events_log_path = "%s/events-%d.json" % (trial_dir, generation)
-    mutated_filename = '%s/mutated-%d.yaml' % (trial_dir, generation)
-    operator_log_path = "%s/operator-%d.log" % (trial_dir, generation)
-    system_state_path = "%s/system-state-%03d.json" % (trial_dir, generation)
-    cli_output_path = "%s/cli-output-%d.log" % (trial_dir, generation)
-    runtime_result_path = "%s/generation-%d-runtime.json" % (trial_dir, generation)
+def construct_step(trial_dir, generation) -> Optional[Step]:
+    """Construct a step from a trial directory and generation"""
+    events_log_path = f"{trial_dir}/events-{generation:03d}.json"
+    mutated_filename = f"{trial_dir}/mutated-{generation:03d}.yaml"
+    operator_log_path = f"{trial_dir}/operator-{generation:03d}.log"
+    system_state_path = f"{trial_dir}/system-state-{generation:03d}.json"
+    cli_output_path = f"{trial_dir}/cli-output-{generation:03d}.log"
+    runtime_result_path = (
+        f"{trial_dir}/generation-{generation:03d}-runtime.json"
+    )
 
     if not os.path.exists(operator_log_path):
         return None
 
     if not os.path.exists(runtime_result_path):
         return None
 
-    with open(mutated_filename, 'r') as input_file, \
-            open(operator_log_path, 'r') as operator_log_file, \
-            open(system_state_path, 'r') as system_state_file, \
-            open(events_log_path, 'r') as events_log, \
-            open(cli_output_path, 'r') as cli_output, \
-            open(runtime_result_path, 'r') as runtime_result_file:
-
-        input = yaml.load(input_file, Loader=yaml.FullLoader)
+    with open(mutated_filename, "r", encoding="utf-8") as input_file, open(
+        operator_log_path,
+        "r",
+        encoding="utf-8",
+    ) as operator_log_file, open(
+        system_state_path,
+        "r",
+        encoding="utf-8",
+    ) as system_state_file, open(
+        events_log_path,
+        "r",
+        encoding="utf-8",
+    ) as _, open(
+        cli_output_path,
+        "r",
+        encoding="utf-8",
+    ) as cli_output, open(
+        runtime_result_path,
+        "r",
+        encoding="utf-8",
+    ) as runtime_result_file:
+        input_cr = yaml.load(input_file, Loader=yaml.FullLoader)
         operator_log = operator_log_file.read().splitlines()
         system_state = json.load(system_state_file)
         cli_result = json.load(cli_output)
         runtime_result = json.load(runtime_result_file)
 
-        return Step(trial_dir, generation, input, operator_log, system_state, cli_result,
-                    runtime_result)
+        return Step(
+            trial_dir,
+            generation,
+            input_cr,
+            operator_log,
+            system_state,
+            cli_result,
+            runtime_result,
+        )
 
 
 class PostProcessor(object):
-
+    """Post processor base class"""
     def __init__(self, testrun_dir: str, config: OperatorConfig):
         # Set config and context
         self.config = config
-        context_cache = os.path.join(os.path.dirname(config.seed_custom_resource), 'context.json')
-        with open(context_cache, 'r') as context_fin:
+        context_cache = os.path.join(
+            os.path.dirname(config.seed_custom_resource), "context.json"
+        )
+        with open(context_cache, "r", encoding="utf-8") as context_fin:
             self._context = json.load(context_fin)
-            self._context['preload_images'] = set(self._context['preload_images'])
+            self._context["preload_images"] = set(
+                self._context["preload_images"]
+            )
 
         # Initliaze trial dirs
         self._trials: List[str] = []
-        self._trial_to_steps: Dict[str,
-                                   Dict[str, Step]] = {}  # trial_dir -> steps, key by trial_dir string
-        trial_dirs = glob.glob(testrun_dir + '/*')
+        self._trial_to_steps: Dict[
+            str, Dict[str, Step]
+        ] = {}  # trial_dir -> steps, key by trial_dir string
+        trial_dirs = glob.glob(testrun_dir + "/*")
         for trial_dir in trial_dirs:
             if not os.path.isdir(trial_dir):
                 continue
             else:
                 self._trials.append(trial_dir)
-                self._trial_to_steps[os.path.basename(trial_dir)] = read_trial_dir(trial_dir)
+                self._trial_to_steps[
+                    os.path.basename(trial_dir)
+                ] = read_trial_dir(trial_dir)
 
     @property
     def trials(self) -> List[str]:
+        """Return trials"""
         return self._trials
 
     @trials.setter
@@ -151,12 +197,15 @@ def trials(self, value):
 
     @property
     def trial_to_steps(self) -> Dict[str, Dict[str, Step]]:
+        """Return the dictionary for trial to steps"""
         return self._trial_to_steps
 
     @property
     def context(self) -> Dict:
+        """Return context"""
         return self._context
 
     @abstractmethod
-    def post_process(self):
-        raise NotImplementedError
+    def post_process(self, *args, **kwargs):
+        """Post process the testrun results"""
+        raise NotImplementedError

test/integration_tests/test_data/crdbop-920/delta-002.log

@@ -0,0 +1,55 @@
+---------- INPUT DELTA  ----------
+{
+      "values_changed": {
+            "root['spec']['ingress']['sql']['tls'][0]['secretName']": {
+                  "prev": "ACTOKEY",
+                  "curr": "",
+                  "path": [
+                        "spec",
+                        "ingress",
+                        "sql",
+                        "tls",
+                        0,
+                        "secretName"
+                  ]
+            }
+      }
+}
+---------- SYSTEM DELTA ----------
+{
+      "daemonset_pods": {},
+      "daemon_set": {},
+      "pod": {},
+      "deployment_pods": {},
+      "stateful_set": {},
+      "deployment": {},
+      "config_map": {},
+      "service": {},
+      "service_account": {},
+      "pvc": {},
+      "cronjob": {},
+      "ingress": {},
+      "network_policy": {},
+      "pod_disruption_budget": {},
+      "secret": {},
+      "endpoints": {},
+      "job": {},
+      "role": {},
+      "role_binding": {},
+      "custom_resource_spec": {
+            "dictionary_item_removed": {
+                  "root['ingress']['sql']['tls'][0]['secretName']": {
+                        "prev": "ACTOKEY",
+                        "curr": "NotPresent",
+                        "path": [
+                              "ingress",
+                              "sql",
+                              "tls",
+                              0,
+                              "secretName"
+                        ]
+                  }
+            }
+      },
+      "custom_resource_status": {}
+}