diff options
| author | Ján Jančár | 2025-02-17 20:38:44 +0100 |
|---|---|---|
| committer | GitHub | 2025-02-17 20:38:44 +0100 |
| commit | 27abe1d969703a87aace43334969939be1b8c9f8 (patch) | |
| tree | b8de3ec1e588b1b982753bb6798efa7d9ddd0b29 | |
| parent | 3d885d064b7758bf83321b179771a27df834dcef (diff) | |
| parent | 67ccbf3f4792fe39f9c75e0546140930d6dccd40 (diff) | |
| download | sec-certs-27abe1d969703a87aace43334969939be1b8c9f8.tar.gz sec-certs-27abe1d969703a87aace43334969939be1b8c9f8.tar.zst sec-certs-27abe1d969703a87aace43334969939be1b8c9f8.zip | |
Merge pull request #479 from crocs-muni/fix/no-dummy-path
Get rid of DUMMY_NONEXISTING_PATH
| -rw-r--r-- | src/sec_certs/constants.py | 4 | ||||
| -rw-r--r-- | src/sec_certs/dataset/auxiliary_dataset_handling.py | 23 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cc.py | 26 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cc_scheme.py | 7 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cpe.py | 8 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cve.py | 8 | ||||
| -rw-r--r-- | src/sec_certs/dataset/dataset.py | 43 | ||||
| -rw-r--r-- | src/sec_certs/dataset/fips.py | 12 | ||||
| -rw-r--r-- | src/sec_certs/dataset/fips_algorithm.py | 10 | ||||
| -rw-r--r-- | src/sec_certs/dataset/fips_iut.py | 10 | ||||
| -rw-r--r-- | src/sec_certs/dataset/fips_mip.py | 10 | ||||
| -rw-r--r-- | src/sec_certs/dataset/json_path_dataset.py | 30 | ||||
| -rw-r--r-- | src/sec_certs/dataset/protection_profile.py | 36 | ||||
| -rw-r--r-- | src/sec_certs/sample/fips_iut.py | 8 | ||||
| -rw-r--r-- | src/sec_certs/sample/fips_mip.py | 11 | ||||
| -rw-r--r-- | src/sec_certs/serialization/json.py | 125 | ||||
| -rw-r--r-- | tests/fips/test_fips_mip.py | 1 |
17 files changed, 265 insertions, 107 deletions
diff --git a/src/sec_certs/constants.py b/src/sec_certs/constants.py index ccc3887e..5626f999 100644 --- a/src/sec_certs/constants.py +++ b/src/sec_certs/constants.py @@ -1,5 +1,4 @@ import re -from pathlib import Path from typing import Final, Literal RANDOM_STATE: Final[int] = 42 @@ -7,9 +6,6 @@ REF_ANNOTATION_MODES = Literal["training", "evaluation", "production", "cross-va REF_EMBEDDING_METHOD = Literal["tf_idf", "transformer"] -# This stupid thing should die in a fire... -DUMMY_NONEXISTING_PATH = Path("/this/is/dummy/nonexisting/path") - REQUEST_TIMEOUT = 20 INCREMENTAL_NVD_UPDATE_MAX_INTERVAL_DAYS: Final[int] = 120 diff --git a/src/sec_certs/dataset/auxiliary_dataset_handling.py b/src/sec_certs/dataset/auxiliary_dataset_handling.py index d89f6520..cd67f929 100644 --- a/src/sec_certs/dataset/auxiliary_dataset_handling.py +++ b/src/sec_certs/dataset/auxiliary_dataset_handling.py @@ -10,7 +10,6 @@ from typing import Any, ClassVar import requests -from sec_certs import constants from sec_certs.configuration import config from sec_certs.dataset.cc_scheme import CCSchemeDataset from sec_certs.dataset.cpe import CPEDataset @@ -30,8 +29,15 @@ class AuxiliaryDatasetHandler(ABC): aux_datasets_dir: Path dset: Any - def __init__(self, aux_datasets_dir: str | Path) -> None: - self.aux_datasets_dir = Path(aux_datasets_dir) + def __init__(self, aux_datasets_dir: str | Path | None) -> None: + self.aux_datasets_dir = Path(aux_datasets_dir) if aux_datasets_dir is not None else None # type: ignore + + @property + def is_backed(self) -> bool: + """ + Returns whether the dataset is backed by a directory. + """ + return self.aux_datasets_dir is not None @property def root_dir(self) -> Path: @@ -44,8 +50,8 @@ class AuxiliaryDatasetHandler(ABC): def dset_path(self) -> Path: raise NotImplementedError("Not meant to be implemented by base class") - def set_local_paths(self, aux_datasets_dir: str | Path) -> None: - self.aux_datasets_dir = Path(aux_datasets_dir) + def set_local_paths(self, aux_datasets_dir: str | Path | None) -> None: + self.aux_datasets_dir = Path(aux_datasets_dir) if aux_datasets_dir is not None else None # type: ignore def process_dataset(self, download_fresh: bool = False) -> None: self.root_dir.mkdir(parents=True, exist_ok=True) @@ -178,7 +184,7 @@ class FIPSAlgorithmDatasetHandler(AuxiliaryDatasetHandler): class CCSchemeDatasetHandler(AuxiliaryDatasetHandler): def __init__( self, - aux_datasets_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, + aux_datasets_dir: str | Path | None, only_schemes: set[str] | None = None, ): super().__init__(aux_datasets_dir) @@ -209,7 +215,7 @@ class CCMaintenanceUpdateDatasetHandler(AuxiliaryDatasetHandler): def __init__( self, - aux_datasets_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, + aux_datasets_dir: str | Path | None, certs_with_updates: Iterable[CCCertificate] | None = None, ) -> None: super().__init__(aux_datasets_dir) @@ -255,9 +261,6 @@ class CCMaintenanceUpdateDatasetHandler(AuxiliaryDatasetHandler): class ProtectionProfileDatasetHandler(AuxiliaryDatasetHandler): RELATIVE_DIR: ClassVar[str] = "protection_profiles" - def __init__(self, aux_datasets_dir: str | Path = constants.DUMMY_NONEXISTING_PATH): - super().__init__(aux_datasets_dir) - @property def dset_path(self) -> Path: return self.root_dir / "dataset.json" diff --git a/src/sec_certs/dataset/cc.py b/src/sec_certs/dataset/cc.py index dc1c52b4..fa95f5cf 100644 --- a/src/sec_certs/dataset/cc.py +++ b/src/sec_certs/dataset/cc.py @@ -36,7 +36,7 @@ from sec_certs.heuristics.cc import ( from sec_certs.heuristics.common import compute_cpe_heuristics, compute_related_cves, compute_transitive_vulnerabilities from sec_certs.sample.cc import CCCertificate from sec_certs.sample.cc_maintenance_update import CCMaintenanceUpdate -from sec_certs.serialization.json import ComplexSerializableType, serialize +from sec_certs.serialization.json import ComplexSerializableType, only_backed, serialize from sec_certs.utils import helpers, sanitization from sec_certs.utils import parallel_processing as cert_processing from sec_certs.utils.profiling import staged @@ -92,7 +92,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): def __init__( self, certs: dict[str, CCCertificate] | None = None, - root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, + root_dir: str | Path | None = None, name: str | None = None, description: str = "", state: Dataset.DatasetInternalState | None = None, @@ -143,6 +143,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return df @property + @only_backed(throw=False) def reports_dir(self) -> Path: """ Returns directory that holds files associated with certification reports @@ -150,6 +151,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.certs_dir / "reports" @property + @only_backed(throw=False) def reports_pdf_dir(self) -> Path: """ Returns directory that holds PDFs associated with certification reports @@ -157,6 +159,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.reports_dir / "pdf" @property + @only_backed(throw=False) def reports_txt_dir(self) -> Path: """ Returns directory that holds TXTs associated with certification reports @@ -164,6 +167,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.reports_dir / "txt" @property + @only_backed(throw=False) def targets_dir(self) -> Path: """ Returns directory that holds files associated with security targets @@ -171,6 +175,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.certs_dir / "targets" @property + @only_backed(throw=False) def targets_pdf_dir(self) -> Path: """ Returns directory that holds PDFs associated with security targets @@ -178,6 +183,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.targets_dir / "pdf" @property + @only_backed(throw=False) def targets_txt_dir(self) -> Path: """ Returns directory that holds TXTs associated with security targets @@ -185,6 +191,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.targets_dir / "txt" @property + @only_backed(throw=False) def certificates_dir(self) -> Path: """ Returns directory that holds files associated with the certificates @@ -192,6 +199,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.certs_dir / "certificates" @property + @only_backed(throw=False) def certificates_pdf_dir(self) -> Path: """ Returns directory that holds PDFs associated with certificates @@ -199,6 +207,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.certificates_dir / "pdf" @property + @only_backed(throw=False) def certificates_txt_dir(self) -> Path: """ Returns directory that holds TXTs associated with certificates @@ -206,6 +215,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return self.certificates_dir / "txt" @property + @only_backed(throw=False) def reference_annotator_dir(self) -> Path: return self.root_dir / "reference_annotator" @@ -229,6 +239,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): } @property + @only_backed(throw=False) def active_html_tuples(self) -> list[tuple[str, Path]]: """ Returns List Tuple[str, Path] where first element is name of html file and second element is its Path. @@ -237,6 +248,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return [(x, self.web_dir / y) for y, x in self.HTML_PRODUCTS_URL.items() if "active" in y] @property + @only_backed(throw=False) def archived_html_tuples(self) -> list[tuple[str, Path]]: """ Returns List Tuple[str, Path] where first element is name of html file and second element is its Path. @@ -245,6 +257,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return [(x, self.web_dir / y) for y, x in self.HTML_PRODUCTS_URL.items() if "archived" in y] @property + @only_backed(throw=False) def active_csv_tuples(self) -> list[tuple[str, Path]]: """ Returns List Tuple[str, Path] where first element is name of csv file and second element is its Path. @@ -253,6 +266,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): return [(x, self.web_dir / y) for y, x in self.CSV_PRODUCTS_URL.items() if "active" in y] @property + @only_backed(throw=False) def archived_csv_tuples(self) -> list[tuple[str, Path]]: """ Returns List Tuple[str, Path] where first element is name of csv file and second element is its Path. @@ -262,6 +276,8 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): def _set_local_paths(self): super()._set_local_paths() + if self.root_dir is None: + return for cert in self: cert.set_local_paths( @@ -273,6 +289,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): self.certificates_txt_dir, ) + @only_backed() def process_auxiliary_datasets( self, download_fresh: bool = False, @@ -324,6 +341,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): @serialize @staged(logger, "Downloading and processing CSV and HTML files of certificates.") + @only_backed() def get_certs_from_web( self, to_download: bool = True, @@ -771,6 +789,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType): self._extract_target_keywords() self._extract_cert_keywords() + @only_backed() def extract_data(self) -> None: logger.info("Extracting various data from certification artifacts") self._extract_pdf_metadata() @@ -811,7 +830,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): def __init__( self, certs: dict[str, CCMaintenanceUpdate] | None = None, # type: ignore - root_dir: Path = constants.DUMMY_NONEXISTING_PATH, + root_dir: str | Path | None = None, name: str = "dataset name", description: str = "dataset_description", state: CCDataset.DatasetInternalState | None = None, @@ -820,6 +839,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): self.state.meta_sources_parsed = True @property + @only_backed(throw=False) def certs_dir(self) -> Path: return self.root_dir diff --git a/src/sec_certs/dataset/cc_scheme.py b/src/sec_certs/dataset/cc_scheme.py index e7f4433e..a1ee3124 100644 --- a/src/sec_certs/dataset/cc_scheme.py +++ b/src/sec_certs/dataset/cc_scheme.py @@ -4,7 +4,6 @@ import logging from collections.abc import Mapping from pathlib import Path -from sec_certs import constants from sec_certs.dataset.json_path_dataset import JSONPathDataset from sec_certs.sample.cc_scheme import CCScheme from sec_certs.serialization.json import ComplexSerializableType @@ -21,9 +20,9 @@ class CCSchemeDataset(JSONPathDataset, ComplexSerializableType): of a product name and most have a vendor/developer/manufacturer field. """ - def __init__(self, schemes: dict[str, CCScheme], json_path: str | Path = constants.DUMMY_NONEXISTING_PATH): + def __init__(self, schemes: dict[str, CCScheme], json_path: str | Path | None = None): + super().__init__(json_path) self.schemes = schemes - self.json_path = Path(json_path) @property def serialized_attributes(self) -> list[str]: @@ -51,7 +50,7 @@ class CCSchemeDataset(JSONPathDataset, ComplexSerializableType): @classmethod def from_web( cls, - json_path: str | Path = constants.DUMMY_NONEXISTING_PATH, + json_path: str | Path | None = None, only_schemes: set[str] | None = None, enhanced: bool | None = None, artifacts: bool | None = None, diff --git a/src/sec_certs/dataset/cpe.py b/src/sec_certs/dataset/cpe.py index 43db8a79..922969fa 100644 --- a/src/sec_certs/dataset/cpe.py +++ b/src/sec_certs/dataset/cpe.py @@ -10,7 +10,6 @@ from typing import Any import pandas as pd import requests -from sec_certs import constants from sec_certs.configuration import config from sec_certs.dataset.json_path_dataset import JSONPathDataset from sec_certs.sample.cpe import CPE @@ -32,11 +31,11 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): def __init__( self, cpes: dict[str, CPE] | None = None, - json_path: str | Path = constants.DUMMY_NONEXISTING_PATH, + json_path: str | Path | None = None, last_update_timestamp: datetime = datetime.fromtimestamp(0), ): + super().__init__(json_path) self.cpes = cpes if cpes is not None else {} - self.json_path = Path(json_path) self.last_update_timestamp = last_update_timestamp def __iter__(self) -> Iterator[CPE]: @@ -96,7 +95,8 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): dset.json_path = json_path dset.to_json() else: - dset.json_path = constants.DUMMY_NONEXISTING_PATH + # Clear the json_path, as it points to temporary file + dset.json_path = None return dset def enhance_with_nvd_data(self, nvd_data: dict[Any, Any]) -> None: diff --git a/src/sec_certs/dataset/cve.py b/src/sec_certs/dataset/cve.py index 1dbf8b37..9186a7f6 100644 --- a/src/sec_certs/dataset/cve.py +++ b/src/sec_certs/dataset/cve.py @@ -12,7 +12,6 @@ import pandas as pd import requests import sec_certs.configuration as config_module -from sec_certs import constants from sec_certs.dataset.json_path_dataset import JSONPathDataset from sec_certs.sample.cpe import CPE from sec_certs.sample.cve import CVE @@ -30,11 +29,11 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType): def __init__( self, cves: dict[str, CVE] | None = None, - json_path: str | Path = constants.DUMMY_NONEXISTING_PATH, + json_path: str | Path | None = None, last_update_timestamp: datetime = datetime.fromtimestamp(0), ): + super().__init__(json_path) self.cves = cves if cves is not None else {} - self.json_path = Path(json_path) self._cpe_uri_to_cve_ids_lookup: dict[str, set[str]] = {} self._cves_with_vulnerable_configurations: list[CVE] = [] self.last_update_timestamp = last_update_timestamp @@ -92,7 +91,8 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType): dset.json_path = json_path dset.to_json() else: - dset.json_path = constants.DUMMY_NONEXISTING_PATH + # Clear the json_path, as it points to temporary file + dset.json_path = None return dset def _get_cves_with_criteria_configurations(self) -> None: diff --git a/src/sec_certs/dataset/dataset.py b/src/sec_certs/dataset/dataset.py index 03535764..7add7b52 100644 --- a/src/sec_certs/dataset/dataset.py +++ b/src/sec_certs/dataset/dataset.py @@ -15,12 +15,12 @@ import pandas as pd import requests from pydantic import AnyHttpUrl -from sec_certs import constants from sec_certs.dataset.auxiliary_dataset_handling import AuxiliaryDatasetHandler from sec_certs.sample.certificate import Certificate from sec_certs.serialization.json import ( ComplexSerializableType, get_class_fullname, + only_backed, serialize, ) from sec_certs.utils import helpers @@ -52,7 +52,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): def __init__( self, certs: dict[str, CertSubType] | None = None, - root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, + root_dir: str | Path | None = None, name: str | None = None, description: str = "", state: DatasetInternalState | None = None, @@ -65,24 +65,35 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): self.name = name if name else type(self).__name__ self.description = description if description else datetime.now().strftime("%d/%m/%Y %H:%M:%S") self.state = state if state else self.DatasetInternalState() - self.root_dir = Path(root_dir) + self.root_dir = Path(root_dir) if root_dir is not None else None # type: ignore self.aux_handlers = aux_handlers if aux_handlers is not None else {} # Make sure that the auxiliary handlers (if supplied by the user) have the correct root_dir self._set_local_paths() @property + def is_backed(self) -> bool: + """ + Returns whether the dataset is backed by a directory. + """ + return self.root_dir is not None + + @property def root_dir(self) -> Path: """ Directory that will hold the serialized dataset files. """ - return self._root_dir + return self._root_dir # type: ignore @root_dir.setter - def root_dir(self, new_dir: str | Path) -> None: + def root_dir(self, new_dir: str | Path | None) -> None: """ This setter will only set the root dir and all internal paths so that they point to the new root dir. No data is being moved around. """ + if new_dir is None: + self._root_dir = None + return + new_dir = Path(new_dir) if new_dir.is_file(): raise ValueError(f"Root dir of {get_class_fullname(self)} cannot be a file.") @@ -91,6 +102,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): self._set_local_paths() @property + @only_backed(throw=False) def web_dir(self) -> Path: """ Path to certification-artifacts posted on web. @@ -98,6 +110,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): return self.root_dir / "web" @property + @only_backed(throw=False) def auxiliary_datasets_dir(self) -> Path: """ Path to directory with auxiliary datasets. @@ -105,6 +118,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): return self.root_dir / "auxiliary_datasets" @property + @only_backed(throw=False) def certs_dir(self) -> Path: """ Returns directory that holds files associated with certificates @@ -112,6 +126,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): return self.root_dir / "certs" @property + @only_backed(throw=False) def json_path(self) -> Path: return self.root_dir / (self.name + ".json") @@ -148,7 +163,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): archive_url: AnyHttpUrl | None = None, snapshot_url: AnyHttpUrl | None = None, progress_bar_desc: str | None = None, - path: None | str | Path = None, + path: str | Path | None = None, auxiliary_datasets: bool = False, artifacts: bool = False, ) -> DatasetSubType: @@ -215,7 +230,8 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): if path: dset.move_dataset(path) else: - dset.root_dir = constants.DUMMY_NONEXISTING_PATH + # Clear the path, as it points to temporary file + dset._root_dir = None if auxiliary_datasets: dset.process_auxiliary_datasets(download_fresh=True) return dset @@ -251,10 +267,13 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): return dset def _set_local_paths(self) -> None: - if hasattr(self, "aux_handlers"): + if self.root_dir is None: + return + if hasattr(self, "aux_handlers") and self.aux_handlers: for handler in self.aux_handlers.values(): handler.set_local_paths(self.auxiliary_datasets_dir) + @only_backed() def move_dataset(self, new_root_dir: str | Path) -> None: """ Moves all dataset files to `new_root_dir` and adjusts all paths internally. Deletes the artifacts from the original location. @@ -270,6 +289,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): shutil.rmtree(self.root_dir) self.root_dir = new_root_dir + @only_backed() def copy_dataset(self, new_root_dir: str | Path) -> None: """ Copies all dataset files to `new_root_dir` and adjusts all paths internally. Keeps the artifacts from the original location. @@ -296,6 +316,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): @staged(logger, "Processing auxiliary datasets") @serialize + @only_backed() def process_auxiliary_datasets(self, download_fresh: bool = False, **kwargs) -> None: """ Processes all auxiliary datasets (CPE, CVE, ...) that are required during computation. @@ -305,6 +326,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): handler.process_dataset(download_fresh) self.state.auxiliary_datasets_processed = True + @only_backed() def load_auxiliary_datasets(self) -> None: logger.info("Loading auxiliary datasets into memory.") for handler in self.aux_handlers.values(): @@ -317,6 +339,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): ) @serialize + @only_backed() def download_all_artifacts(self, fresh: bool = True) -> None: """ Downloads all artifacts related to certification in the given scheme. @@ -337,6 +360,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): raise NotImplementedError("Not meant to be implemented by the base class.") @serialize + @only_backed() def convert_all_pdfs(self, fresh: bool = True) -> None: """ Converts all pdf artifacts to txt, given the certification scheme. @@ -355,6 +379,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): raise NotImplementedError("Not meant to be implemented by the base class.") @serialize + @only_backed() def analyze_certificates(self) -> None: """ Does two things: @@ -381,10 +406,12 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): self.compute_heuristics() @abstractmethod + @only_backed() def extract_data(self) -> None: raise NotImplementedError("Not meant to be implemented by the base class.") @serialize + @only_backed() def compute_heuristics(self) -> None: logger.info("Computing various heuristics from the certificates.") self.load_auxiliary_datasets() diff --git a/src/sec_certs/dataset/fips.py b/src/sec_certs/dataset/fips.py index cbe4b6b2..a64ca3d7 100644 --- a/src/sec_certs/dataset/fips.py +++ b/src/sec_certs/dataset/fips.py @@ -24,7 +24,7 @@ from sec_certs.dataset.dataset import Dataset from sec_certs.heuristics.common import compute_cpe_heuristics, compute_related_cves, compute_transitive_vulnerabilities from sec_certs.heuristics.fips import compute_references from sec_certs.sample.fips import FIPSCertificate -from sec_certs.serialization.json import ComplexSerializableType, serialize +from sec_certs.serialization.json import ComplexSerializableType, only_backed, serialize from sec_certs.utils import helpers from sec_certs.utils import parallel_processing as cert_processing from sec_certs.utils.helpers import fips_dgst @@ -59,7 +59,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): def __init__( self, certs: dict[str, FIPSCertificate] | None = None, - root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, + root_dir: str | Path | None = None, name: str | None = None, description: str = "", state: Dataset.DatasetInternalState | None = None, @@ -81,18 +81,22 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): } @property + @only_backed(throw=False) def policies_dir(self) -> Path: return self.certs_dir / "policies" @property + @only_backed(throw=False) def policies_pdf_dir(self) -> Path: return self.policies_dir / "pdf" @property + @only_backed(throw=False) def policies_txt_dir(self) -> Path: return self.policies_dir / "txt" @property + @only_backed(throw=False) def module_dir(self) -> Path: return self.certs_dir / "modules" @@ -128,6 +132,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): compute_transitive_vulnerabilities(self.certs) @serialize + @only_backed() def extract_data(self) -> None: logger.info("Extracting various data from certification artifacts.") for cert in self: @@ -235,11 +240,14 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): def _set_local_paths(self) -> None: super()._set_local_paths() + if self.root_dir is None: + return for cert in self: cert.set_local_paths(self.policies_pdf_dir, self.policies_txt_dir, self.module_dir) @serialize @staged(logger, "Downloading and processing certificates.") + @only_backed() def get_certs_from_web(self, to_download: bool = True, keep_metadata: bool = True) -> None: self.web_dir.mkdir(parents=True, exist_ok=True) diff --git a/src/sec_certs/dataset/fips_algorithm.py b/src/sec_certs/dataset/fips_algorithm.py index c9a8b3c5..70a9f25d 100644 --- a/src/sec_certs/dataset/fips_algorithm.py +++ b/src/sec_certs/dataset/fips_algorithm.py @@ -21,11 +21,9 @@ logger = logging.getLogger(__name__) class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType): - def __init__( - self, algs: dict[str, FIPSAlgorithm] | None = None, json_path: str | Path = constants.DUMMY_NONEXISTING_PATH - ): + def __init__(self, algs: dict[str, FIPSAlgorithm] | None = None, json_path: str | Path | None = None): + super().__init__(json_path) self.algs = algs if algs is not None else {} - self.json_path = Path(json_path) self.alg_number_to_algs: dict[str, set[FIPSAlgorithm]] = {} self._build_lookup_dicts() @@ -55,11 +53,11 @@ class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType): return isinstance(other, FIPSAlgorithmDataset) and self.algs == other.algs @classmethod - def from_web(cls, json_path: str | Path = constants.DUMMY_NONEXISTING_PATH) -> FIPSAlgorithmDataset: + def from_web(cls, json_path: str | Path | None = None) -> FIPSAlgorithmDataset: with TemporaryDirectory() as tmp_dir: htmls = FIPSAlgorithmDataset.download_alg_list_htmls(Path(tmp_dir)) algs = set(itertools.chain.from_iterable(FIPSAlgorithmDataset.parse_algorithms_from_html(x) for x in htmls)) - return cls({x.dgst: x for x in algs}, json_path) + return cls({x.dgst: x for x in algs}, json_path=json_path) @staticmethod def download_alg_list_htmls(output_dir: Path) -> list[Path]: diff --git a/src/sec_certs/dataset/fips_iut.py b/src/sec_certs/dataset/fips_iut.py index 494358f6..45986c11 100644 --- a/src/sec_certs/dataset/fips_iut.py +++ b/src/sec_certs/dataset/fips_iut.py @@ -7,7 +7,6 @@ from tempfile import NamedTemporaryFile import requests -from sec_certs import constants from sec_certs.configuration import config from sec_certs.dataset.dataset import logger from sec_certs.dataset.json_path_dataset import JSONPathDataset @@ -19,11 +18,10 @@ from sec_certs.utils.tqdm import tqdm @dataclass class IUTDataset(JSONPathDataset, ComplexSerializableType): snapshots: list[IUTSnapshot] - _json_path: Path - def __init__(self, snapshots: list[IUTSnapshot], json_path: str | Path = constants.DUMMY_NONEXISTING_PATH): + def __init__(self, snapshots: list[IUTSnapshot], json_path: str | Path | None = None): + super().__init__(json_path) self.snapshots = snapshots - self.json_path = Path(json_path) def __iter__(self) -> Iterator[IUTSnapshot]: yield from self.snapshots @@ -59,8 +57,8 @@ class IUTDataset(JSONPathDataset, ComplexSerializableType): Get the IUTDataset from sec-certs.org """ iut_resp = requests.get(config.fips_iut_dataset) - if iut_resp.status_code != 200: + if iut_resp.status_code != requests.codes.ok: raise ValueError(f"Getting IUT dataset failed: {iut_resp.status_code}") - with NamedTemporaryFile() as tmpfile: + with NamedTemporaryFile(suffix=".json") as tmpfile: tmpfile.write(iut_resp.content) return cls.from_json(tmpfile.name) diff --git a/src/sec_certs/dataset/fips_mip.py b/src/sec_certs/dataset/fips_mip.py index a5af3f4a..41ab4646 100644 --- a/src/sec_certs/dataset/fips_mip.py +++ b/src/sec_certs/dataset/fips_mip.py @@ -9,7 +9,6 @@ from tempfile import NamedTemporaryFile import requests -from sec_certs import constants from sec_certs.configuration import config from sec_certs.dataset.dataset import logger from sec_certs.dataset.json_path_dataset import JSONPathDataset @@ -21,11 +20,10 @@ from sec_certs.utils.tqdm import tqdm @dataclass class MIPDataset(JSONPathDataset, ComplexSerializableType): snapshots: list[MIPSnapshot] - _json_path: Path - def __init__(self, snapshots: list[MIPSnapshot], json_path: str | Path = constants.DUMMY_NONEXISTING_PATH): + def __init__(self, snapshots: list[MIPSnapshot], json_path: str | Path | None = None): + super().__init__(json_path) self.snapshots = snapshots - self.json_path = Path(json_path) def __iter__(self) -> Iterator[MIPSnapshot]: yield from self.snapshots @@ -61,9 +59,9 @@ class MIPDataset(JSONPathDataset, ComplexSerializableType): Get the MIPDataset from sec-certs.org """ mip_resp = requests.get(config.fips_mip_dataset) - if mip_resp.status_code != 200: + if mip_resp.status_code != requests.codes.ok: raise ValueError(f"Getting MIP dataset failed: {mip_resp.status_code}") - with NamedTemporaryFile() as tmpfile: + with NamedTemporaryFile(suffix=".json") as tmpfile: tmpfile.write(mip_resp.content) return cls.from_json(tmpfile.name) diff --git a/src/sec_certs/dataset/json_path_dataset.py b/src/sec_certs/dataset/json_path_dataset.py index bfd71ddd..76df73f2 100644 --- a/src/sec_certs/dataset/json_path_dataset.py +++ b/src/sec_certs/dataset/json_path_dataset.py @@ -5,33 +5,49 @@ import shutil from abc import ABC from pathlib import Path -from sec_certs.serialization.json import ComplexSerializableType, get_class_fullname +from sec_certs.serialization.json import ComplexSerializableType, get_class_fullname, only_backed logger = logging.getLogger(__name__) class JSONPathDataset(ComplexSerializableType, ABC): - _json_path: Path + _json_path: Path | None + + def __init__(self, json_path: str | Path | None = None): + super().__init__() + self.json_path = Path(json_path) if json_path is not None else None + + @property + def is_backed(self) -> bool: + """ + Returns whether the dataset is backed by a JSON file. + """ + return self.json_path is not None @property - def json_path(self) -> Path: + def json_path(self) -> Path | None: return self._json_path @json_path.setter - def json_path(self, new_path: str | Path) -> None: + def json_path(self, new_path: str | Path | None) -> None: + if new_path is None: + self._json_path = None + return + new_path = Path(new_path) if new_path.is_dir(): raise ValueError(f"Json path of {get_class_fullname(self)} cannot be a directory.") self._json_path = new_path + @only_backed() def move_dataset(self, new_json_path: str | Path) -> None: - logger.info(f"Moving {get_class_fullname(self)} dataset to {new_json_path}") + logger.info(f"Moving {get_class_fullname(self)} dataset to {new_json_path}.") new_json_path = Path(new_json_path) new_json_path.parent.mkdir(parents=True, exist_ok=True) - if self.json_path.exists(): - shutil.move(str(self.json_path), str(new_json_path)) + if self.json_path and self.json_path.exists(): + shutil.move(self.json_path, new_json_path) self.json_path = new_json_path else: self.json_path = new_json_path diff --git a/src/sec_certs/dataset/protection_profile.py b/src/sec_certs/dataset/protection_profile.py index b4200795..40113c87 100644 --- a/src/sec_certs/dataset/protection_profile.py +++ b/src/sec_certs/dataset/protection_profile.py @@ -10,7 +10,7 @@ from sec_certs.configuration import config from sec_certs.dataset.auxiliary_dataset_handling import AuxiliaryDatasetHandler from sec_certs.dataset.dataset import Dataset, logger from sec_certs.sample.protection_profile import ProtectionProfile -from sec_certs.serialization.json import ComplexSerializableType, serialize +from sec_certs.serialization.json import ComplexSerializableType, only_backed, serialize from sec_certs.utils import helpers from sec_certs.utils import parallel_processing as cert_processing from sec_certs.utils.profiling import staged @@ -39,7 +39,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy def __init__( self, certs: dict[str, ProtectionProfile] | None = None, - root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, + root_dir: str | Path | None = None, name: str | None = None, description: str = "", state: Dataset.DatasetInternalState | None = None, @@ -48,10 +48,12 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy super().__init__(certs, root_dir, name, description, state, aux_handlers) @property + @only_backed(throw=False) def json_path(self) -> Path: return self.root_dir / "dataset.json" @property + @only_backed(throw=False) def reports_dir(self) -> Path: """ Path to protection profile reports. @@ -59,6 +61,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy return self.root_dir / "reports" @property + @only_backed(throw=False) def pps_dir(self) -> Path: """ Path to actual protection profiles. @@ -66,6 +69,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy return self.root_dir / "pps" @property + @only_backed(throw=False) def reports_pdf_dir(self) -> Path: """ Path to pdfs of protection profile reports. @@ -73,6 +77,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy return self.reports_dir / "pdf" @property + @only_backed(throw=False) def reports_txt_dir(self) -> Path: """ Path to txts of protection profile reports. @@ -80,6 +85,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy return self.reports_dir / "txt" @property + @only_backed(throw=False) def pps_pdf_dir(self) -> Path: """ Path to pdfs of protection profiles @@ -87,22 +93,29 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy return self.pps_dir / "pdf" @property + @only_backed(throw=False) def pps_txt_dir(self) -> Path: """ Path to txts of protection profiles. """ return self.pps_dir / "txt" - def _compute_heuristics_body(self): - logger.info("Protection profile dataset has no heuristics to compute, skipping.") - @property + @only_backed(throw=False) def web_dir(self) -> Path: """ Path to directory with html sources downloaded from commoncriteriaportal.org """ return self.root_dir / "web" + def _set_local_paths(self): + super()._set_local_paths() + if self.root_dir is None: + return + + for cert in self: + cert.set_local_paths(self.reports_pdf_dir, self.pps_pdf_dir, self.reports_txt_dir, self.pps_txt_dir) + HTML_URL = { "pp_active.html": constants.CC_PORTAL_BASE_URL + "/pps/index.cfm", "pp_archived.html": constants.CC_PORTAL_BASE_URL + "/pps/index.cfm?archived=1", @@ -110,19 +123,23 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy } @property + @only_backed(throw=False) def active_html_tuples(self) -> list[tuple[str, Path]]: return [(x, self.web_dir / y) for y, x in self.HTML_URL.items() if "active" in y] @property + @only_backed(throw=False) def archived_html_tuples(self) -> list[tuple[str, Path]]: return [(x, self.web_dir / y) for y, x in self.HTML_URL.items() if "archived" in y] @property + @only_backed(throw=False) def collaborative_html_tuples(self) -> list[tuple[str, Path]]: return [(x, self.web_dir / y) for y, x in self.HTML_URL.items() if "collaborative" in y] @serialize @staged(logger, "Downloading and processing CSV and HTML files of certificates.") + @only_backed() def get_certs_from_web( self, to_download: bool = True, @@ -299,6 +316,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy progress_bar_desc="Downloading PDFs of actual Protection Profiles.", ) + @only_backed() def extract_data(self): """ Extracts pdf metadata and keywords from converted text documents. @@ -357,12 +375,10 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy ) self.update_with_certs(processed_certs) - def _set_local_paths(self): - super()._set_local_paths() - - for cert in self: - cert.set_local_paths(self.reports_pdf_dir, self.pps_pdf_dir, self.reports_txt_dir, self.pps_txt_dir) + def _compute_heuristics_body(self): + logger.info("Protection profile dataset has no heuristics to compute, skipping.") + @only_backed() def process_auxiliary_datasets(self, **kwargs) -> None: """ Dummy method to adhere to `Dataset` interface. `ProtectionProfile` dataset has currently no auxiliary datasets. diff --git a/src/sec_certs/sample/fips_iut.py b/src/sec_certs/sample/fips_iut.py index 6d979dde..963758c7 100644 --- a/src/sec_certs/sample/fips_iut.py +++ b/src/sec_certs/sample/fips_iut.py @@ -148,7 +148,7 @@ class IUTSnapshot(ComplexSerializableType): Get an IUT snapshot from the FIPS website right now. """ iut_resp = requests.get(constants.FIPS_IUT_URL) - if iut_resp.status_code != 200: + if iut_resp.status_code != requests.codes.ok: raise ValueError(f"Getting IUT snapshot failed: {iut_resp.status_code}") snapshot_date = to_utc(datetime.now()) @@ -157,7 +157,7 @@ class IUTSnapshot(ComplexSerializableType): @classmethod def from_web(cls) -> IUTSnapshot: """ - Fetch a fresh snapshot from sec-certs.org, if the `preferred_source_remote_datasets` config + Fetch a fresh IUT snapshot from sec-certs.org, if the `preferred_source_remote_datasets` config entry is equal to "sec-certs". Otherwise, the same as `from_nist_web`. @@ -166,8 +166,8 @@ class IUTSnapshot(ComplexSerializableType): return cls.from_nist_web() else: iut_resp = requests.get(config.fips_iut_latest_snapshot) - if iut_resp.status_code != 200: + if iut_resp.status_code != requests.codes.ok: raise ValueError(f"Getting IUT snapshot failed: {iut_resp.status_code}") - with NamedTemporaryFile() as tmpfile: + with NamedTemporaryFile(suffix=".json") as tmpfile: tmpfile.write(iut_resp.content) return cls.from_json(tmpfile.name) diff --git a/src/sec_certs/sample/fips_mip.py b/src/sec_certs/sample/fips_mip.py index c7d84ad1..684263e2 100644 --- a/src/sec_certs/sample/fips_mip.py +++ b/src/sec_certs/sample/fips_mip.py @@ -251,7 +251,7 @@ class MIPSnapshot(ComplexSerializableType): Get a MIP snapshot from the FIPS website right now. """ mip_resp = requests.get(constants.FIPS_MIP_URL) - if mip_resp.status_code != 200: + if mip_resp.status_code != requests.codes.ok: raise ValueError(f"Getting MIP snapshot failed: {mip_resp.status_code}") snapshot_date = to_utc(datetime.now()) @@ -260,14 +260,17 @@ class MIPSnapshot(ComplexSerializableType): @classmethod def from_web(cls) -> MIPSnapshot: """ - Get a MIP snapshot from the FIPS website right now. + Fetch a fresh MIP snapshot from sec-certs.org, if the `preferred_source_remote_datasets` config + entry is equal to "sec-certs". + + Otherwise, the same as `from_nist_web`. """ if config.preferred_source_remote_datasets == "origin": return cls.from_nist_web() else: mip_resp = requests.get(config.fips_mip_latest_snapshot) - if mip_resp.status_code != 200: + if mip_resp.status_code != requests.codes.ok: raise ValueError(f"Getting MIP snapshot failed: {mip_resp.status_code}") - with NamedTemporaryFile() as tmpfile: + with NamedTemporaryFile(suffix=".json") as tmpfile: tmpfile.write(mip_resp.content) return cls.from_json(tmpfile.name) diff --git a/src/sec_certs/serialization/json.py b/src/sec_certs/serialization/json.py index dad0ebb7..c843638c 100644 --- a/src/sec_certs/serialization/json.py +++ b/src/sec_certs/serialization/json.py @@ -3,15 +3,17 @@ from __future__ import annotations import copy import gzip import json +import logging from collections.abc import Callable from datetime import date, datetime from functools import wraps from pathlib import Path -from typing import Any, TypeVar - -from sec_certs import constants +from typing import Any, TypeVar, cast T = TypeVar("T", bound="ComplexSerializableType") +TCallable = TypeVar("TCallable", bound=Callable[..., Any]) + +logger = logging.getLogger(__name__) class SerializationError(Exception): @@ -19,13 +21,21 @@ class SerializationError(Exception): class ComplexSerializableType: + """ + A class that can be serialized to json and thus a dictionary. + + Direct inheritance from this class is required for the class to be serializable. + Only the `serialized_attributes` are serialized. If `__slots__` is defined, only those attributes are serialized. + + .. note:: + The `to_dict` and `from_dict` should be overridden if non-trivial types of attributes need to be serialized. + """ + __slots__: tuple[str] def __init__(self, *args, **kwargs): pass - # Ideally, the serialized_fields would be an class variable referencing itself, but that it virtually impossible - # to achieve without using metaclasses. Not to complicate the code, we choose instance variable. @property def serialized_attributes(self) -> list[str]: if hasattr(self, "__slots__") and self.__slots__: @@ -50,29 +60,38 @@ class ComplexSerializableType: """ Serializes `ComplexSerializableType` instance to json file. :param str | Path | None output_path: path where the file will be stored. If None, `obj.json_path` access is attempted, defaults to None - :param bool compress: if True, will be compress with gzip, defaults to False + :param bool compress: if True, will be compressed with gzip, defaults to False """ if not output_path and (not hasattr(self, "json_path") or not self.json_path): # type: ignore raise SerializationError( - f"The object {self} of type {self.__class__} does not have json_path attribute set but to_json() was called without an argument." + f"The object {self} of type {get_class_fullname(self)} does not have json_path attribute set but to_json() was called without an argument." ) if not output_path: output_path = self.json_path # type: ignore - if self.json_path == constants.DUMMY_NONEXISTING_PATH: # type: ignore - raise SerializationError(f"json_path attribute for '{get_class_fullname(self)}' was not yet set.") - if hasattr(self, "root_dir") and self.root_dir == constants.DUMMY_NONEXISTING_PATH: # type: ignore - raise SerializationError(f"root_dir attribute for '{get_class_fullname(self)}' was not yet set.") + if self.json_path is None: # type: ignore + raise SerializationError(f"json_path attribute for {get_class_fullname(self)} was not yet set.") + if hasattr(self, "root_dir") and self.root_dir is None: # type: ignore + raise SerializationError(f"root_dir attribute for {get_class_fullname(self)} was not yet set.") + + if not output_path: + raise SerializationError("Output path for json must be set.") - if Path(output_path).is_dir(): # type: ignore - raise SerializationError("output path for json cannot be directory.") + path = Path(output_path) + if path.is_dir(): + raise SerializationError("Output path for json cannot be a directory.") - # false positive MyPy warning, cannot be None if compress: - with gzip.open(str(output_path), "wt", encoding="utf-8") as handle: # type: ignore - json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) + if path.suffix != ".gz": + raise SerializationError(f"Expected path to a compressed file (.gz), got {path.suffix}.") + + with gzip.open(path, "wt", encoding="utf-8") as handle: + json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) # type: ignore else: - with Path(output_path).open("w") as handle: # type: ignore - json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) + if path.suffix != ".json": + raise SerializationError(f"Expected path to a json file (.json), got {path.suffix}.") + + with path.open("wt") as handle: + json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) # type: ignore @classmethod def from_json(cls: type[T], input_path: str | Path, is_compressed: bool = False) -> T: @@ -82,16 +101,29 @@ class ComplexSerializableType: :param bool is_compressed: if True, will decompress .gz first, defaults to False :return T: the deserialized object """ + path = Path(input_path) if is_compressed: - with gzip.open(str(input_path), "rt", encoding="utf-8") as handle: + if path.suffix != ".gz": + raise SerializationError(f"Expected path to a compressed file (.gz), got {path.suffix}.") + + with gzip.open(path, "rt", encoding="utf-8") as handle: return json.load(handle, cls=CustomJSONDecoder) else: - with Path(input_path).open("r") as handle: + if path.suffix != ".json": + raise SerializationError(f"Expected path to a json file (.json), got {path.suffix}.") + + with path.open("r") as handle: return json.load(handle, cls=CustomJSONDecoder) -# Decorator for serialization -def serialize(func: Callable): +def serialize(func: Callable) -> Callable: + """ + Decorator to be used on instance methods of ComplexSerializableType child classes. + The decorated method will be serialized to json after execution. + + Adds the `update_json` keyword argument to the decorated method. If set to False, the json will not be updated. + """ + @wraps(func) def _serialize(*args, **kwargs): if not args or not issubclass(type(args[0]), ComplexSerializableType): @@ -99,7 +131,7 @@ def serialize(func: Callable): "@serialize decorator is to be used only on instance methods of ComplexSerializableType child classes." ) - if hasattr(args[0], "_root_dir") and args[0]._root_dir == constants.DUMMY_NONEXISTING_PATH: + if hasattr(args[0], "root_dir") and args[0].root_dir is None: raise SerializationError( "The invoked method requires dataset serialization. Cannot serialize without root_dir set. You can set it with obj.root_dir = ..." ) @@ -113,7 +145,44 @@ def serialize(func: Callable): return _serialize +def only_backed(throw: bool = True): + """ + Decorator to be used on instance methods of ComplexSerializableType child classes. + The decorated method will only be executed if the `root_dir` attribute is set. + + :param bool throw: if True, will raise ValueError if `root_dir` is not set, defaults to True + Otherwise, just logs a warning and returns None. + """ + + def deco(func: TCallable) -> TCallable: + @wraps(func) + def _only_backed(*args, **kwargs): + if args[0].root_dir is None: + if throw: + raise ValueError(f"Method {func.__name__} can only be called on backed dataset.") + else: + logger.warning(f"Method {func.__name__} can only be called on backed dataset.") + return None + else: + return func(*args, **kwargs) + + return cast(TCallable, _only_backed) + + return deco + + def get_class_fullname(obj: Any) -> str: + """ + Returns the full name of the class of the object. + + Example: + >>> get_class_fullname(datetime.now()) + 'datetime.datetime' + + + :param Any obj: object to get the class name from + :return str: full name of the class + """ klass = obj if isinstance(obj, type) else obj.__class__ module = klass.__module__ if module == "builtins": @@ -122,6 +191,10 @@ def get_class_fullname(obj: Any) -> str: class CustomJSONEncoder(json.JSONEncoder): + """ + Custom JSONEncoder. + """ + def default(self, obj): if isinstance(obj, ComplexSerializableType): return {**{"_type": get_class_fullname(obj)}, **obj.to_dict()} @@ -142,8 +215,10 @@ class CustomJSONEncoder(json.JSONEncoder): class CustomJSONDecoder(json.JSONDecoder): """ - Custom JSONDecoder. Any complex object that should be de-serializable must inherit directly from class - ComplexSerializableType (nested inheritance does not currently work (because x.__subclassess__() prints only direct + Custom JSONDecoder. + + Any complex object that should be de-serializable must inherit directly from class + `ComplexSerializableType` (nested inheritance does not currently work (because x.__subclassess__() prints only direct subclasses. Any such class must implement methods to_dict() and from_dict(). These are used to drive serialization. """ diff --git a/tests/fips/test_fips_mip.py b/tests/fips/test_fips_mip.py index c15836ef..b8ac805f 100644 --- a/tests/fips/test_fips_mip.py +++ b/tests/fips/test_fips_mip.py @@ -33,6 +33,7 @@ def test_mip_dataset_from_dumps(data_dir: Path): assert len(dset) == 3 +@pytest.mark.xfail(reason="May fail due to network issues.") def test_mip_flows(): dset = MIPDataset.from_web() assert dset.compute_flows() |
