From 925dbcbb2a7ab7d1b0353f3cd16de41e296288fe Mon Sep 17 00:00:00 2001 From: etannant Date: Wed, 7 Oct 2026 14:58:02 -0700 Subject: [PATCH] AMDSMI CPER utf-8 serialization issue fix --- nodescraper/models/datamodel.py | 30 +++++++++++----- test/unit/framework/test_datamodel.py | 51 ++++++++++++++++++++++++++- 2 files changed, 71 insertions(+), 10 deletions(-) diff --git a/nodescraper/models/datamodel.py b/nodescraper/models/datamodel.py index 45c21049..1d6c0826 100644 --- a/nodescraper/models/datamodel.py +++ b/nodescraper/models/datamodel.py @@ -24,12 +24,11 @@ # ############################################################################### import io -import json import os import tarfile from typing import Any, TypeVar, Union -from pydantic import BaseModel, field_validator +from pydantic import BaseModel, ConfigDict, ValidationInfo, field_validator from nodescraper.utils import get_unique_filename @@ -37,16 +36,24 @@ class FileModel(BaseModel): + """Binary file payload. JSON uses URL-safe base64 for ``file_contents``.""" + + model_config = ConfigDict(ser_json_bytes="base64", val_json_bytes="base64") + file_contents: bytes file_name: str @field_validator("file_contents", mode="before") @classmethod - def file_contents_conformer(cls, value: Union[io.BytesIO, str, bytes]) -> bytes: + def file_contents_conformer( + cls, value: Union[io.BytesIO, str, bytes], info: ValidationInfo + ) -> Union[str, bytes]: if isinstance(value, io.BytesIO): return value.getvalue() - if isinstance(value, str): + if isinstance(value, str) and info.mode != "json": + # Constructor/Python input is text. return value.encode("utf-8") + # JSON strings stay base64 for val_json_bytes. return value def log_model(self, log_path: str) -> None: @@ -77,15 +84,21 @@ def log_model(self, log_path: str): get_unique_filename(log_path, f"{self.__class__.__name__.lower()}.json"), ) - exlude_fields = set() + # Write binary FileModel payloads as files; keep them in JSON as base64. for key in self.__class__.model_fields: data = getattr(self, key) if isinstance(data, FileModel): data.log_model(log_path) - exlude_fields.add(key) + elif ( + isinstance(data, list) + and data + and all(isinstance(item, FileModel) for item in data) + ): + for item in data: + item.log_model(log_path) with open(log_name, "w", encoding="utf-8") as log_file: - log_file.write(self.model_dump_json(indent=2, exclude=exlude_fields)) + log_file.write(self.model_dump_json(indent=2)) def merge_data(self, input_data: "DataModel") -> None: """Merge data into current data""" @@ -122,8 +135,7 @@ def import_model(cls: type[TDataModel], model_input: Union[dict[str, Any], str]) # Build from json file else: with open(model_input, "r", encoding="utf-8") as input_file: - data = json.load(input_file) - return cls(**data) + return cls.model_validate_json(input_file.read()) raise ValueError("Invalid input for model data") diff --git a/test/unit/framework/test_datamodel.py b/test/unit/framework/test_datamodel.py index ed6c2567..c2bdaa99 100644 --- a/test/unit/framework/test_datamodel.py +++ b/test/unit/framework/test_datamodel.py @@ -2,11 +2,12 @@ from __future__ import annotations +import base64 import json import os from pathlib import Path -from nodescraper.models.datamodel import DataModel +from nodescraper.models.datamodel import DataModel, FileModel class FolderDataModel(DataModel): @@ -17,6 +18,11 @@ def build_from_folder(cls, folder_path: str) -> "FolderDataModel": return cls(value=os.path.basename(folder_path)) +class CperListDataModel(DataModel): + value: str = "ok" + cper_data: list[FileModel] = [] + + def test_import_model_from_dict(): """Baseline: a dict is passed straight to the model constructor.""" assert FolderDataModel.import_model({"value": "abc"}).value == "abc" @@ -36,3 +42,46 @@ def test_import_model_from_directory_uses_build_from_folder(tmp_path: Path): folder.mkdir() assert FolderDataModel.import_model(str(folder)).value == "collected" + + +def test_filemodel_json_roundtrips_binary_cper_bytes(): + """Non-UTF-8 CPER bytes serialize as base64 and reload intact as bytes.""" + cper_bytes = b"CPER\x00\x01\xff\xff\xff\xff" + model = FileModel(file_contents=cper_bytes, file_name="corrected-1.cper") + dumped = model.model_dump_json() + payload = json.loads(dumped) + assert payload["file_contents"] == base64.urlsafe_b64encode(cper_bytes).decode("ascii") + + reloaded = FileModel.model_validate_json(dumped) + assert isinstance(reloaded.file_contents, bytes) + assert reloaded.file_contents == cper_bytes + + +def test_filemodel_python_strings_are_utf8_not_base64(): + """A plain string that is also valid base64 must be stored as UTF-8 text.""" + text = "VGVzdA==" + model = FileModel(file_contents=text, file_name="t.txt") + assert model.file_contents == text.encode("utf-8") + assert FileModel(file_contents="hello", file_name="t.txt").file_contents == b"hello" + + +def test_log_model_keeps_cper_list_in_json(tmp_path: Path): + """CPER-like bytes must not crash log_model and remain in the JSON dump.""" + cper_bytes = b"CPER\x00\x01\xff\xff\xff\xff" + model = CperListDataModel( + value="ok", + cper_data=[FileModel(file_contents=cper_bytes, file_name="corrected-1.cper")], + ) + + model.log_model(str(tmp_path)) + + cper_path = tmp_path / "corrected-1.cper" + assert cper_path.read_bytes() == cper_bytes + + json_files = list(tmp_path.glob("*.json")) + assert len(json_files) == 1 + payload = json.loads(json_files[0].read_text(encoding="utf-8")) + assert payload["value"] == "ok" + assert len(payload["cper_data"]) == 1 + assert payload["cper_data"][0]["file_name"] == "corrected-1.cper" + assert base64.urlsafe_b64decode(payload["cper_data"][0]["file_contents"]) == cper_bytes