feat(class_generator): Add support for parsing user code from generat… (

#2121)

* feat(class_generator): Add support for parsing user code from generated files

* Fix case when file have no extra code from the user

* fix tests

class_generator/class_generator.py

@@ -298,6 +298,19 @@ def render_jinja_template(template_dict: Dict[Any, Any], template_dir: str, temp
     return rendered
 
 
+def parse_user_code_from_file(file_path: str) -> str:
+    with open(file_path) as fd:
+        data = fd.read()
+
+    line = "    # End of generated code"
+    if line in data:
+        _end_of_generated_code_index = data.index(line)
+        _user_code = data[_end_of_generated_code_index + len(line) :]
+        return _user_code
+
+    return ""
+
+
 def generate_resource_file_from_dict(
     resource_dict: Dict[str, Any],
     overwrite: bool = False,
@@ -318,7 +331,7 @@ def generate_resource_file_from_dict(
     )
 
     formatted_kind_str = convert_camel_case_to_snake_case(string_=resource_dict["kind"])
-    _file_suffix = f"{'_' + output_file_suffix if output_file_suffix else ''}"
+    _file_suffix: str = f"{'_' + output_file_suffix if output_file_suffix else ''}"
 
     if add_tests:
         overwrite = True
@@ -334,8 +347,14 @@ def generate_resource_file_from_dict(
     else:
         _output_file = os.path.join(base_dir, f"{formatted_kind_str}{_file_suffix}.py")
 
+    _output_file_exists: bool = os.path.exists(_output_file)
+    _user_code: str = ""
+
+    if _output_file_exists:
+        _user_code = parse_user_code_from_file(file_path=_output_file)
+
     orig_filename = _output_file
-    if os.path.exists(_output_file):
+    if _output_file_exists:
         if overwrite:
             LOGGER.warning(f"Overwriting {_output_file}")
 
@@ -345,11 +364,13 @@ def generate_resource_file_from_dict(
             _output_file = temp_output_file
 
     if dry_run:
+        if _user_code:
+            rendered += _user_code
         _code = Syntax(code=rendered, lexer="python", line_numbers=True)
         Console().print(_code)
 
     else:
-        write_and_format_rendered(filepath=_output_file, data=rendered)
+        write_and_format_rendered(filepath=_output_file, data=rendered, user_code=_user_code)
 
     return orig_filename, _output_file
 
@@ -566,10 +587,13 @@ def class_generator(
     return generated_files
 
 
-def write_and_format_rendered(filepath: str, data: str) -> None:
+def write_and_format_rendered(filepath: str, data: str, user_code: str = "") -> None:
     with open(filepath, "w") as fd:
         fd.write(data)
 
+        if user_code:
+            fd.write(user_code)
+
     for op in ("format", "check"):
         run_command(
             command=shlex.split(f"poetry run ruff {op} {filepath}"),

class_generator/manifests/class_generator_template.j2

@@ -92,3 +92,4 @@ class {{ kind }}({{ base_class }}):
           {% endif %}
           {% endfor %}
     {% endif %}
+    # End of generated code

class_generator/tests/manifests/APIServer/api_server.py

@@ -90,3 +90,5 @@ def to_dict(self) -> None:
 
             if self.tls_security_profile:
                 _spec["tlsSecurityProfile"] = self.tls_security_profile
+
+    # End of generated code

class_generator/tests/manifests/ClusterOperator/cluster_operator.py

@@ -17,3 +17,5 @@ def __init__(
         **kwargs: Any,
     ) -> None:
         super().__init__(**kwargs)
+
+    # End of generated code

class_generator/tests/manifests/ConfigMap/config_map.py

@@ -57,3 +57,5 @@ def to_dict(self) -> None:
 
             if self.immutable is not None:
                 self.res["immutable"] = self.immutable
+
+    # End of generated code

class_generator/tests/manifests/DNS/dns_config_openshift_io.py

@@ -69,3 +69,5 @@ def to_dict(self) -> None:
 
             if self.public_zone:
                 _spec["publicZone"] = self.public_zone
+
+    # End of generated code

class_generator/tests/manifests/DNS/dns_operator_openshift_io.py

@@ -118,3 +118,5 @@ def to_dict(self) -> None:
 
             if self.upstream_resolvers:
                 _spec["upstreamResolvers"] = self.upstream_resolvers
+
+    # End of generated code

class_generator/tests/manifests/Deployment/deployment.py

@@ -102,3 +102,5 @@ def to_dict(self) -> None:
 
             if self.strategy:
                 _spec["strategy"] = self.strategy
+
+    # End of generated code

class_generator/tests/manifests/Image/image_caching_internal_knative_dev.py

@@ -61,3 +61,5 @@ def to_dict(self) -> None:
 
             if self.service_account_name:
                 _spec["serviceAccountName"] = self.service_account_name
+
+    # End of generated code

class_generator/tests/manifests/Image/image_config_openshift_io.py

@@ -73,3 +73,5 @@ def to_dict(self) -> None:
 
             if self.registry_sources:
                 _spec["registrySources"] = self.registry_sources
+
+    # End of generated code

class_generator/tests/manifests/Image/image_image_openshift_io.py

@@ -130,3 +130,5 @@ def to_dict(self) -> None:
 
             if self.signatures:
                 self.res["signatures"] = self.signatures
+
+    # End of generated code

class_generator/tests/manifests/ImageContentSourcePolicy/image_content_source_policy.py

@@ -53,3 +53,5 @@ def to_dict(self) -> None:
 
             if self.repository_digest_mirrors:
                 _spec["repositoryDigestMirrors"] = self.repository_digest_mirrors
+
+    # End of generated code

class_generator/tests/manifests/Machine/machine.py

@@ -86,3 +86,5 @@ def to_dict(self) -> None:
 
             if self.taints:
                 _spec["taints"] = self.taints
+
+    # End of generated code

class_generator/tests/manifests/NMState/nm_state.py

@@ -93,3 +93,5 @@ def to_dict(self) -> None:
 
             if self.tolerations:
                 _spec["tolerations"] = self.tolerations
+
+    # End of generated code

class_generator/tests/manifests/Pod/pod.py

@@ -452,3 +452,5 @@ def to_dict(self) -> None:
 
             if self.volumes:
                 _spec["volumes"] = self.volumes
+
+    # End of generated code

class_generator/tests/manifests/Secret/secret.py

@@ -65,3 +65,5 @@ def to_dict(self) -> None:
 
             if self.type:
                 self.res["type"] = self.type
+
+    # End of generated code

class_generator/tests/manifests/ServiceMeshMember/service_mesh_member.py

@@ -38,3 +38,5 @@ def to_dict(self) -> None:
             _spec = self.res["spec"]
 
             _spec["controlPlaneRef"] = self.control_plane_ref
+
+    # End of generated code