Compare commits
38 Commits
0.0.1
...
0.0.1.post
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
616a53578c | ||
|
|
d20712a72f | ||
|
|
2837b6439f | ||
|
|
b4d6ed6130 | ||
|
|
cd69fc22a8 | ||
|
|
9aec2d66cd | ||
|
|
af81ec5fd3 | ||
|
|
26e32a004f | ||
|
|
b7cbc50f79 | ||
|
|
86eee2e378 | ||
|
|
3e0defc574 | ||
|
|
f6e581381d | ||
|
|
981c5201a9 | ||
|
|
ab11052c8f | ||
|
|
4f5dade949 | ||
|
|
cce260bc5e | ||
|
|
915a4332ee | ||
|
|
4dca3eb9d1 | ||
|
|
e11c541139 | ||
|
|
637b50b325 | ||
|
|
f45c9cc8f3 | ||
|
|
95b0c298ce | ||
|
|
04a4cf7b36 | ||
|
|
f42a839cff | ||
|
|
7f3a4ef545 | ||
|
|
608c8a1010 | ||
|
|
210781f086 | ||
|
|
df966ccac4 | ||
|
|
29827b51bc | ||
|
|
7440731135 | ||
|
|
87682c2c9c | ||
|
|
0eef35e36f | ||
|
|
4c17436cea | ||
|
|
a90ab4885b | ||
|
|
192dcc74f8 | ||
|
|
171a2f1617 | ||
|
|
001ffbbbf1 | ||
|
|
8ab6c8e179 |
2
.project
2
.project
@@ -1,6 +1,6 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<projectDescription>
|
||||
<name>{{project_name}}</name>
|
||||
<name>dabmodel</name>
|
||||
<comment></comment>
|
||||
<projects>
|
||||
</projects>
|
||||
|
||||
@@ -1,13 +1,20 @@
|
||||
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
|
||||
<?eclipse-pydev version="1.0"?><pydev_project>
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_INTERPRETER">Default</pydev_property>
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_VERSION">python interpreter</pydev_property>
|
||||
|
||||
<pydev_pathproperty name="org.python.pydev.PROJECT_SOURCE_PATH">
|
||||
<path>/${PROJECT_DIR_NAME}/src</path>
|
||||
<path>/${PROJECT_DIR_NAME}</path>
|
||||
</pydev_pathproperty>
|
||||
|
||||
</pydev_project>
|
||||
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
|
||||
<?eclipse-pydev version="1.0"?><pydev_project>
|
||||
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_INTERPRETER">Python311</pydev_property>
|
||||
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_VERSION">python interpreter</pydev_property>
|
||||
|
||||
|
||||
<pydev_pathproperty name="org.python.pydev.PROJECT_SOURCE_PATH">
|
||||
|
||||
<path>/${PROJECT_DIR_NAME}/src</path>
|
||||
|
||||
<path>/${PROJECT_DIR_NAME}</path>
|
||||
|
||||
</pydev_pathproperty>
|
||||
|
||||
|
||||
</pydev_project>
|
||||
|
||||
@@ -34,7 +34,9 @@ classifiers = [
|
||||
]
|
||||
dependencies = [
|
||||
'importlib-metadata; python_version<"3.9"',
|
||||
'packaging'
|
||||
'packaging',
|
||||
'frozendict',
|
||||
'typeguard'
|
||||
]
|
||||
dynamic = ["version"]
|
||||
|
||||
@@ -78,7 +80,7 @@ test = ["chacha_cicd_helper"]
|
||||
coverage-check = ["chacha_cicd_helper"]
|
||||
complexity-check = ["chacha_cicd_helper"]
|
||||
quality-check = ["chacha_cicd_helper"]
|
||||
type-check = ["chacha_cicd_helper"]
|
||||
type-check = ["chacha_cicd_helper","types-pytz"]
|
||||
doc-gen = ["chacha_cicd_helper"]
|
||||
|
||||
# [project.scripts]
|
||||
|
||||
17
src/dabmodel/LAMFields/Constraint.py
Normal file
17
src/dabmodel/LAMFields/Constraint.py
Normal file
@@ -0,0 +1,17 @@
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
T_Field = TypeVar("T_Field")
|
||||
|
||||
|
||||
class Constraint(Generic[T_Field]):
|
||||
"""Constraint class
|
||||
Base class for Field's constraints
|
||||
"""
|
||||
|
||||
_bound_type: type
|
||||
|
||||
def __init__(self): ...
|
||||
|
||||
def check(self, value: T_Field) -> bool:
|
||||
"""Check if a Constraint is completed"""
|
||||
return True
|
||||
41
src/dabmodel/LAMFields/FrozenLAMField.py
Normal file
41
src/dabmodel/LAMFields/FrozenLAMField.py
Normal file
@@ -0,0 +1,41 @@
|
||||
from typing import Generic, TypeVar, Any
|
||||
|
||||
from .LAMField import LAMField
|
||||
from .Constraint import Constraint
|
||||
from ..tools import LAMdeepfreeze
|
||||
|
||||
T_Field = TypeVar("T_Field")
|
||||
|
||||
|
||||
class FrozenLAMField(Generic[T_Field]):
|
||||
"""FrozenLAMField class
|
||||
a read-only proxy of a Field
|
||||
"""
|
||||
|
||||
def __init__(self, inner_field: LAMField[T_Field]):
|
||||
self._inner_field = inner_field
|
||||
|
||||
@property
|
||||
def doc(self) -> str:
|
||||
"""Returns Field's documentation (frozen)"""
|
||||
return LAMdeepfreeze(self._inner_field.doc)
|
||||
|
||||
@property
|
||||
def constraints(self) -> tuple[Constraint]:
|
||||
"""Returns Field's constraint (frozen)"""
|
||||
return LAMdeepfreeze(self._inner_field.constraints)
|
||||
|
||||
@property
|
||||
def default_value(self) -> Any:
|
||||
"""Returns Field's default value (frozen)"""
|
||||
return self._inner_field.default_value
|
||||
|
||||
@property
|
||||
def value(self) -> Any:
|
||||
"""Returns Field's value (frosen)"""
|
||||
return self._inner_field.value
|
||||
|
||||
@property
|
||||
def annotations(self) -> Any:
|
||||
"""Returns Field's annotation (frozen)"""
|
||||
return LAMdeepfreeze(self._inner_field.annotations)
|
||||
38
src/dabmodel/LAMFields/LAMCompatible.py
Normal file
38
src/dabmodel/LAMFields/LAMCompatible.py
Normal file
@@ -0,0 +1,38 @@
|
||||
from typing import Generic, TypeVar, Self, Hashable, Any
|
||||
from abc import ABC, abstractmethod
|
||||
from ..tools import JSONType
|
||||
|
||||
TV_LAMCompatbile = TypeVar("TV_LABCompatbile", bound="LABCompatible")
|
||||
|
||||
|
||||
class LAMCompatible(Generic[TV_LAMCompatbile], ABC):
|
||||
"""Any type that can safely live inside a LABField."""
|
||||
|
||||
@classmethod
|
||||
def lam_validate_annotation(cls, annotation: Any) -> None:
|
||||
"""
|
||||
Validate the type annotation (e.g., SpecialList[int]).
|
||||
Raise if it's not compatible.
|
||||
"""
|
||||
return # default: do nothing (simple types don’t need it)
|
||||
|
||||
@abstractmethod
|
||||
def lam_validate(self) -> None:
|
||||
"""Raise if the value is invalid for this type."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def lam_freeze(self) -> Hashable:
|
||||
"""Return an immutable/hashable representation of this value."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def lam_to_plain(self) -> JSONType:
|
||||
"""Return a plain serializable form (str, dict, etc.)."""
|
||||
...
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def lam_from_plain(cls, plain: JSONType) -> Self:
|
||||
"""Return an Object from a plain serializable form (str, dict, etc.)."""
|
||||
...
|
||||
62
src/dabmodel/LAMFields/LAMField.py
Normal file
62
src/dabmodel/LAMFields/LAMField.py
Normal file
@@ -0,0 +1,62 @@
|
||||
from typing import Generic, TypeVar, Optional, Any
|
||||
|
||||
from .LAMFieldInfo import LAMFieldInfo
|
||||
from .Constraint import Constraint
|
||||
from ..tools import LAMdeepfreeze
|
||||
|
||||
TV_LABField = TypeVar("TV_LABField")
|
||||
|
||||
|
||||
class LAMField(Generic[TV_LABField]):
|
||||
"""This class describe a Field in Schema"""
|
||||
|
||||
def __init__(self, name: str, v: Optional[TV_LABField], a: Any, i: LAMFieldInfo):
|
||||
self._name: str = name
|
||||
self._source: Optional[type] = None
|
||||
self._default_value: Optional[TV_LABField] = v
|
||||
self._value: Optional[TV_LABField] = v
|
||||
self._annotations: Any = a
|
||||
self._info: LAMFieldInfo = i
|
||||
self._constraints: list[Constraint[Any]] = i.constraints
|
||||
|
||||
def add_source(self, s: type) -> None:
|
||||
"""Adds source Appliance to the Field"""
|
||||
self._source = s
|
||||
|
||||
@property
|
||||
def doc(self) -> str:
|
||||
"""Returns Field's documentation"""
|
||||
return self._info.doc
|
||||
|
||||
def add_constraint(self, c: Constraint) -> None:
|
||||
"""Adds constraint to the Field"""
|
||||
self._constraints.append(c)
|
||||
|
||||
@property
|
||||
def constraints(self) -> list[Constraint]:
|
||||
"""Returns Field's constraint"""
|
||||
return self._info.constraints
|
||||
|
||||
@property
|
||||
def default_value(self) -> Any:
|
||||
"""Returns Field's default value (frozen)"""
|
||||
return LAMdeepfreeze(self._default_value)
|
||||
|
||||
def update_value(self, v: Optional[TV_LABField] = None) -> None:
|
||||
"""Updates Field's value"""
|
||||
self._value = v
|
||||
|
||||
@property
|
||||
def value(self) -> Any:
|
||||
"""Returns Field's value (frozen)"""
|
||||
return LAMdeepfreeze(self._value)
|
||||
|
||||
@property
|
||||
def raw_value(self) -> Optional[TV_LABField]:
|
||||
"""Returns Field's value"""
|
||||
return self._value
|
||||
|
||||
@property
|
||||
def annotations(self) -> Any:
|
||||
"""Returns Field's annotation"""
|
||||
return self._annotations
|
||||
26
src/dabmodel/LAMFields/LAMFieldInfo.py
Normal file
26
src/dabmodel/LAMFields/LAMFieldInfo.py
Normal file
@@ -0,0 +1,26 @@
|
||||
from typing import Optional, Any
|
||||
from .Constraint import Constraint
|
||||
|
||||
|
||||
class LAMFieldInfo:
|
||||
"""This Class allows to describe a Field in Appliance class"""
|
||||
|
||||
def __init__(
|
||||
self, *, doc: str = "", constraints: Optional[list[Constraint]] = None
|
||||
):
|
||||
self._doc: str = doc
|
||||
self._constraints: list[Constraint]
|
||||
if constraints is None:
|
||||
self._constraints = []
|
||||
else:
|
||||
self._constraints = constraints
|
||||
|
||||
@property
|
||||
def doc(self) -> str:
|
||||
"""Returns Field's documentation"""
|
||||
return self._doc
|
||||
|
||||
@property
|
||||
def constraints(self) -> list[Constraint[Any]]:
|
||||
"""Returns Field's constraints"""
|
||||
return self._constraints
|
||||
0
src/dabmodel/LAMFields/__init__.py
Normal file
0
src/dabmodel/LAMFields/__init__.py
Normal file
@@ -11,4 +11,33 @@ Main module __init__ file.
|
||||
"""
|
||||
|
||||
from .__metadata__ import __version__, __Summuary__, __Name__
|
||||
from .test_module import test_function
|
||||
|
||||
|
||||
from .LAMFields.LAMField import LAMField
|
||||
from .LAMFields.LAMFieldInfo import LAMFieldInfo
|
||||
from .LAMFields.FrozenLAMField import FrozenLAMField
|
||||
from .LAMFields.LAMCompatible import LAMCompatible
|
||||
from .appliance import Appliance
|
||||
from .feature import Feature
|
||||
from .element import Element
|
||||
|
||||
|
||||
from .exception import (
|
||||
DABModelException,
|
||||
MultipleInheritanceForbidden,
|
||||
BrokenInheritance,
|
||||
ReadOnlyField,
|
||||
NewFieldForbidden,
|
||||
NotAnnotatedField,
|
||||
ReadOnlyFieldAnnotation,
|
||||
InvalidFieldValue,
|
||||
InvalidFieldAnnotation,
|
||||
IncompletelyAnnotatedField,
|
||||
ImportForbidden,
|
||||
FunctionForbidden,
|
||||
InvalidFeatureInheritance,
|
||||
FeatureNotBound,
|
||||
UnsupportedFieldType,
|
||||
)
|
||||
|
||||
__all__ = [name for name in globals() if not name.startswith("_")]
|
||||
|
||||
@@ -15,20 +15,26 @@ import warnings
|
||||
|
||||
try: # pragma: no cover
|
||||
__version__ = version("dabmodel")
|
||||
except PackageNotFoundError: # pragma: no cover
|
||||
warnings.warn("can not read __version__, assuming local test context, setting it to ?.?.?")
|
||||
except PackageNotFoundError: # pragma: no cover
|
||||
warnings.warn(
|
||||
"can not read __version__, assuming local test context, setting it to ?.?.?"
|
||||
)
|
||||
__version__ = "?.?.?"
|
||||
|
||||
try: # pragma: no cover
|
||||
dist = distribution("dabmodel")
|
||||
__Summuary__ = dist.metadata["Summary"]
|
||||
except PackageNotFoundError: # pragma: no cover
|
||||
warnings.warn('can not read dist.metadata["Summary"], assuming local test context, setting it to <dabmodel description>')
|
||||
warnings.warn(
|
||||
'can not read dist.metadata["Summary"], assuming local test context, setting it to <dabmodel description>'
|
||||
)
|
||||
__Summuary__ = "dabmodel description"
|
||||
|
||||
try: # pragma: no cover
|
||||
dist = distribution("dabmodel")
|
||||
__Name__ = dist.metadata["Name"]
|
||||
except PackageNotFoundError: # pragma: no cover
|
||||
warnings.warn('can not read dist.metadata["Name"], assuming local test context, setting it to <dabmodel>')
|
||||
warnings.warn(
|
||||
'can not read dist.metadata["Name"], assuming local test context, setting it to <dabmodel>'
|
||||
)
|
||||
__Name__ = "dabmodel"
|
||||
|
||||
9
src/dabmodel/appliance.py
Normal file
9
src/dabmodel/appliance.py
Normal file
@@ -0,0 +1,9 @@
|
||||
from .element import Element
|
||||
from .meta.appliance import _MetaAppliance
|
||||
|
||||
|
||||
class Appliance(metaclass=_MetaAppliance):
|
||||
"""BaseFeature class
|
||||
Base class for Appliance.
|
||||
An appliance is a server configuration / image that is built using appliance's code and Fields.
|
||||
"""
|
||||
7
src/dabmodel/element.py
Normal file
7
src/dabmodel/element.py
Normal file
@@ -0,0 +1,7 @@
|
||||
from .meta.element import _MetaElement, IElement
|
||||
|
||||
|
||||
class Element(IElement, metaclass=_MetaElement):
|
||||
"""Element class
|
||||
Base class to apply metaclass and set common Fields.
|
||||
"""
|
||||
109
src/dabmodel/exception.py
Normal file
109
src/dabmodel/exception.py
Normal file
@@ -0,0 +1,109 @@
|
||||
class DABModelException(Exception):
|
||||
"""DABModelException Exception class
|
||||
Base Exception for DABModelException class
|
||||
"""
|
||||
|
||||
|
||||
class FunctionForbidden(DABModelException): ...
|
||||
|
||||
|
||||
class ExternalCodeForbidden(FunctionForbidden): ...
|
||||
|
||||
|
||||
class ClosureForbidden(FunctionForbidden): ...
|
||||
|
||||
|
||||
class ReservedFieldName(Exception):
|
||||
"""DABModelException Exception class
|
||||
Base Exception for DABModelException class
|
||||
"""
|
||||
|
||||
|
||||
class MultipleInheritanceForbidden(DABModelException):
|
||||
"""MultipleInheritanceForbidden Exception class
|
||||
Multiple inheritance is forbidden when using dabmodel
|
||||
"""
|
||||
|
||||
|
||||
class BrokenInheritance(DABModelException):
|
||||
"""BrokenInheritance Exception class
|
||||
inheritance chain is broken
|
||||
"""
|
||||
|
||||
|
||||
class ReadOnlyField(DABModelException):
|
||||
"""ReadOnlyField Exception class
|
||||
The used Field is ReadOnly
|
||||
"""
|
||||
|
||||
|
||||
class NewFieldForbidden(DABModelException):
|
||||
"""NewFieldForbidden Exception class
|
||||
Field creation is forbidden
|
||||
"""
|
||||
|
||||
|
||||
class InvalidFieldAnnotation(DABModelException):
|
||||
"""InvalidFieldAnnotation Exception class
|
||||
The field annotation is invalid
|
||||
"""
|
||||
|
||||
|
||||
class InvalidInitializerType(DABModelException):
|
||||
"""InvalidInitializerType Exception class
|
||||
The initializer is not a valid type
|
||||
"""
|
||||
|
||||
|
||||
class NotAnnotatedField(InvalidFieldAnnotation):
|
||||
"""NotAnnotatedField Exception class
|
||||
The Field is not Annotated
|
||||
"""
|
||||
|
||||
|
||||
class IncompletelyAnnotatedField(InvalidFieldAnnotation):
|
||||
"""IncompletelyAnnotatedField Exception class
|
||||
The field annotation is incomplete
|
||||
"""
|
||||
|
||||
|
||||
class UnsupportedFieldType(InvalidFieldAnnotation):
|
||||
"""UnsupportedFieldType Exception class
|
||||
The field type is unsupported
|
||||
"""
|
||||
|
||||
|
||||
class ReadOnlyFieldAnnotation(DABModelException):
|
||||
"""ReadOnlyFieldAnnotation Exception class
|
||||
Field annotation connot be modified
|
||||
"""
|
||||
|
||||
|
||||
class InvalidFieldValue(DABModelException):
|
||||
"""InvalidFieldValue Exception class
|
||||
The Field value is invalid
|
||||
"""
|
||||
|
||||
|
||||
class NonExistingField(DABModelException):
|
||||
"""NonExistingField Exception class
|
||||
The given Field is non existing
|
||||
"""
|
||||
|
||||
|
||||
class ImportForbidden(DABModelException):
|
||||
"""ImportForbidden Exception class
|
||||
Imports are forbidden
|
||||
"""
|
||||
|
||||
|
||||
class InvalidFeatureInheritance(DABModelException):
|
||||
"""InvalidFeatureInheritance Exception class
|
||||
Features of same name in child appliance need to be from same type
|
||||
"""
|
||||
|
||||
|
||||
class FeatureNotBound(DABModelException):
|
||||
"""FeatureNotBound Exception class
|
||||
a Feature must be bound to the appliance (or parent)
|
||||
"""
|
||||
12
src/dabmodel/feature.py
Normal file
12
src/dabmodel/feature.py
Normal file
@@ -0,0 +1,12 @@
|
||||
from .element import Element
|
||||
from .meta.feature import _MetaFeature
|
||||
|
||||
|
||||
class Feature(metaclass=_MetaFeature):
|
||||
"""Feature class
|
||||
Base class for Appliance's Features.
|
||||
Features are optional traits of an appliance.
|
||||
"""
|
||||
|
||||
_BoundAppliance: "Optional[Type[BaseAppliance]]" = None
|
||||
Enabled: bool = False
|
||||
0
src/dabmodel/meta/__init__.py
Normal file
0
src/dabmodel/meta/__init__.py
Normal file
213
src/dabmodel/meta/appliance.py
Normal file
213
src/dabmodel/meta/appliance.py
Normal file
@@ -0,0 +1,213 @@
|
||||
from typing import Any, Type
|
||||
from copy import copy
|
||||
|
||||
from typeguard import check_type, CollectionCheckStrategy, TypeCheckError
|
||||
|
||||
from ..tools import LAMdeepfreeze
|
||||
from ..LAMFields.LAMField import LAMField
|
||||
from ..LAMFields.FrozenLAMField import FrozenLAMField
|
||||
from .element import _MetaElement
|
||||
from ..feature import Feature
|
||||
from ..exception import (
|
||||
InvalidFieldValue,
|
||||
InvalidFeatureInheritance,
|
||||
FeatureNotBound,
|
||||
)
|
||||
|
||||
|
||||
class _MetaAppliance(_MetaElement):
|
||||
"""_MetaAppliance class
|
||||
Appliance specific metaclass code
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def check_class(
|
||||
mcs: type["meta"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions: dict[str, Any],
|
||||
) -> None:
|
||||
"""
|
||||
Appliance-specific pre-check: ensure the `features` slot exists in schema.
|
||||
|
||||
Copies the parent's `features` mapping when inheriting to keep it per-class.
|
||||
"""
|
||||
super().check_class(name, bases, namespace, extensions) # type: ignore[misc]
|
||||
if "features" not in namespace["__LAMSchema__"]:
|
||||
namespace["__LAMSchema__"]["features"] = {}
|
||||
else:
|
||||
namespace["__LAMSchema__"]["features"] = copy(
|
||||
namespace["__LAMSchema__"]["features"]
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def process_class_fields(
|
||||
mcs: type["meta"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any],
|
||||
extensions: dict[str, Any],
|
||||
):
|
||||
"""
|
||||
Like meta.process_class_fields but also stages Feature declarations.
|
||||
|
||||
Initializes:
|
||||
extensions["new_features"], extensions["modified_features"]
|
||||
then defers to the base scanner for regular fields.
|
||||
"""
|
||||
extensions["new_features"] = {}
|
||||
extensions["modified_features"] = {}
|
||||
super().process_class_fields(name, bases, namespace, extensions) # type: ignore[misc]
|
||||
|
||||
@classmethod
|
||||
def process_new_field(
|
||||
mcs: type["meta"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any],
|
||||
_fname: str,
|
||||
_fvalue: Any,
|
||||
extensions: dict[str, Any],
|
||||
): # pylint: disable=unused-argument
|
||||
"""
|
||||
Intercept Feature declarations.
|
||||
|
||||
- If `_fname` already exists in parent's `features`, enforce same type;
|
||||
stage into `modified_features`.
|
||||
- Else, if `_fvalue` is a Feature *class*, stage into `new_features`.
|
||||
- Otherwise, it is a regular field: delegate to meta.process_new_field.
|
||||
"""
|
||||
if _fname in namespace["__LAMSchema__"]["features"].keys():
|
||||
if not issubclass(_fvalue, namespace["__LAMSchema__"]["features"][_fname]):
|
||||
raise InvalidFeatureInheritance(
|
||||
f"Feature {_fname} is not an instance of {bases[0]}.{_fname}"
|
||||
)
|
||||
extensions["modified_features"][_fname] = _fvalue
|
||||
elif isinstance(_fvalue, type) and issubclass(_fvalue, Feature):
|
||||
extensions["new_features"][_fname] = _fvalue
|
||||
else:
|
||||
super().process_new_field(name, bases, namespace, _fname, _fvalue, extensions) # type: ignore[misc]
|
||||
|
||||
@classmethod
|
||||
def commit_fields(
|
||||
mcs: type["meta"],
|
||||
cls,
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions: dict[str, Any],
|
||||
):
|
||||
"""
|
||||
Commit regular fields (via meta) and then bind staged Feature classes.
|
||||
|
||||
For each new/modified feature:
|
||||
- bind it to `cls` (sets the feature's `_BoundAppliance`),
|
||||
- register it under `cls.__LAMSchema__["features"]`.
|
||||
"""
|
||||
super().commit_fields(cls, name, bases, namespace, extensions) # type: ignore[misc]
|
||||
|
||||
for _ftname, _ftvalue in extensions["modified_features"].items():
|
||||
_ftvalue._BoundAppliance = cls # pylint: disable=protected-access
|
||||
cls.__LAMSchema__["features"][_ftname] = _ftvalue
|
||||
for _ftname, _ftvalue in extensions["new_features"].items():
|
||||
_ftvalue._BoundAppliance = cls # pylint: disable=protected-access
|
||||
cls.__LAMSchema__["features"][_ftname] = _ftvalue
|
||||
|
||||
def finalize_instance(cls: Type, obj, extensions: dict[str, Any]):
|
||||
"""
|
||||
Instantiate and attach all features declared (or overridden) in the instance schema.
|
||||
Handles:
|
||||
- Declared features (plain class)
|
||||
- Subclass replacements
|
||||
- Dict overrides (class + patch dict)
|
||||
"""
|
||||
for fname, fdef in obj.__LAMSchema__.get("features", {}).items():
|
||||
# Case 1: plain class or subclass
|
||||
if isinstance(fdef, type) and issubclass(fdef, Feature):
|
||||
inst = fdef()
|
||||
object.__setattr__(obj, fname, inst)
|
||||
|
||||
# Case 2: (class, dict) → dict overrides
|
||||
elif isinstance(fdef, tuple) and len(fdef) == 2:
|
||||
feat_cls, overrides = fdef
|
||||
inst = feat_cls()
|
||||
for field_name, new_val in overrides.items():
|
||||
if field_name not in feat_cls.__LAMSchema__:
|
||||
raise InvalidFieldValue(
|
||||
f"Feature '{fname}' has no field '{field_name}'"
|
||||
)
|
||||
field = feat_cls.__LAMSchema__[field_name]
|
||||
try:
|
||||
check_type(
|
||||
new_val,
|
||||
field.annotations,
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
f"Invalid value for {fname}.{field_name}: "
|
||||
f"expected {field.annotations}, got {new_val!r}"
|
||||
) from exp
|
||||
object.__setattr__(inst, field_name, LAMdeepfreeze(new_val))
|
||||
inst.__LAMSchema__[field_name] = FrozenLAMField(
|
||||
LAMField(field_name, new_val, field.annotations, field._info)
|
||||
)
|
||||
object.__setattr__(obj, fname, inst)
|
||||
|
||||
else:
|
||||
raise InvalidFieldValue(
|
||||
f"Invalid feature definition stored for '{fname}': {fdef!r}"
|
||||
)
|
||||
|
||||
def apply_overrides(cls, obj, extensions, *args, **kwargs):
|
||||
"""
|
||||
Support for runtime field and feature overrides.
|
||||
|
||||
Fields:
|
||||
MyApp(name="foo")
|
||||
|
||||
Features:
|
||||
MyApp(F1=MyF1) # inheritance / replacement
|
||||
MyApp(F1={"val": 42, ...}) # dict override of existing feature
|
||||
"""
|
||||
|
||||
# --- feature overrides ---
|
||||
for k, v in list(kwargs.items()):
|
||||
if k in cls.__LAMSchema__.get("features", {}):
|
||||
base_feat_cls = cls.__LAMSchema__["features"][k]
|
||||
|
||||
# Case 1: subclass replacement (inheritance)
|
||||
if isinstance(v, type) and issubclass(v, base_feat_cls):
|
||||
bound = getattr(v, "_BoundAppliance", None)
|
||||
if bound is None or not issubclass(cls, bound):
|
||||
raise FeatureNotBound(
|
||||
f"Feature {v.__name__} is not bound to {cls.__name__}"
|
||||
)
|
||||
# record subclass into instance schema
|
||||
obj.__LAMSchema__["features"][k] = v
|
||||
kwargs.pop(k)
|
||||
|
||||
# Case 2: dict override
|
||||
elif isinstance(v, dict):
|
||||
# store (class, override_dict) for finalize_instance
|
||||
obj.__LAMSchema__["features"][k] = (base_feat_cls, v)
|
||||
kwargs.pop(k)
|
||||
|
||||
else:
|
||||
raise InvalidFieldValue(
|
||||
f"Feature override for '{k}' must be a Feature subclass or dict, got {type(v)}"
|
||||
)
|
||||
|
||||
# --- new features not declared at class level ---
|
||||
for k, v in list(kwargs.items()):
|
||||
if isinstance(v, type) and issubclass(v, Feature):
|
||||
bound = getattr(v, "_BoundAppliance", None)
|
||||
if bound is None or not issubclass(cls, bound):
|
||||
raise FeatureNotBound(
|
||||
f"Feature {v.__name__} is not bound to {cls.__name__}"
|
||||
)
|
||||
obj.__LAMSchema__["features"][k] = v
|
||||
kwargs.pop(k)
|
||||
|
||||
super().apply_overrides(obj, extensions, *args, **kwargs)
|
||||
797
src/dabmodel/meta/element.py
Normal file
797
src/dabmodel/meta/element.py
Normal file
@@ -0,0 +1,797 @@
|
||||
from typing import (
|
||||
Optional,
|
||||
TypeVar,
|
||||
Union,
|
||||
get_origin,
|
||||
get_args,
|
||||
List,
|
||||
Dict,
|
||||
Any,
|
||||
Tuple,
|
||||
Set,
|
||||
Annotated,
|
||||
FrozenSet,
|
||||
Callable,
|
||||
Type,
|
||||
)
|
||||
|
||||
from types import UnionType, FunctionType, SimpleNamespace
|
||||
from copy import deepcopy, copy
|
||||
|
||||
import math
|
||||
import inspect, ast, textwrap
|
||||
|
||||
from typeguard import check_type, TypeCheckError, CollectionCheckStrategy
|
||||
|
||||
from ..tools import LAMdeepfreeze
|
||||
from ..LAMFields.LAMField import LAMField
|
||||
from ..LAMFields.LAMFieldInfo import LAMFieldInfo
|
||||
from ..LAMFields.FrozenLAMField import FrozenLAMField
|
||||
|
||||
|
||||
from ..exception import (
|
||||
MultipleInheritanceForbidden,
|
||||
BrokenInheritance,
|
||||
ReadOnlyField,
|
||||
NewFieldForbidden,
|
||||
NotAnnotatedField,
|
||||
ReadOnlyFieldAnnotation,
|
||||
InvalidFieldValue,
|
||||
InvalidFieldAnnotation,
|
||||
IncompletelyAnnotatedField,
|
||||
ImportForbidden,
|
||||
FunctionForbidden,
|
||||
NonExistingField,
|
||||
InvalidInitializerType,
|
||||
UnsupportedFieldType,
|
||||
)
|
||||
|
||||
ALLOWED_ANNOTATIONS: dict[str, Any] = {
|
||||
"Union": Union,
|
||||
"Optional": Optional,
|
||||
"List": List,
|
||||
"Dict": Dict,
|
||||
"Tuple": Tuple,
|
||||
"Set": Set,
|
||||
"FrozenSet": FrozenSet,
|
||||
"Annotated": Annotated,
|
||||
# builtins:
|
||||
"int": int,
|
||||
"str": str,
|
||||
"float": float,
|
||||
"bool": bool,
|
||||
"complex": complex,
|
||||
"bytes": bytes,
|
||||
"None": type(None),
|
||||
"list": list,
|
||||
"dict": dict,
|
||||
"set": set,
|
||||
"frozenset": frozenset,
|
||||
"tuple": tuple,
|
||||
}
|
||||
ALLOWED_MODEL_FIELDS_TYPES: tuple[type[Any], ...] = (
|
||||
str,
|
||||
int,
|
||||
float,
|
||||
complex,
|
||||
bool,
|
||||
bytes,
|
||||
)
|
||||
|
||||
|
||||
ALLOWED_HELPERS_MATH = SimpleNamespace(
|
||||
sqrt=math.sqrt,
|
||||
floor=math.floor,
|
||||
ceil=math.ceil,
|
||||
trunc=math.trunc,
|
||||
fabs=math.fabs,
|
||||
copysign=math.copysign,
|
||||
hypot=math.hypot,
|
||||
exp=math.exp,
|
||||
log=math.log,
|
||||
log10=math.log10,
|
||||
sin=math.sin,
|
||||
cos=math.cos,
|
||||
tan=math.tan,
|
||||
atan2=math.atan2,
|
||||
radians=math.radians,
|
||||
degrees=math.degrees,
|
||||
)
|
||||
|
||||
ALLOWED_HELPERS_DEFAULT: dict[str, object] = {
|
||||
"math": ALLOWED_HELPERS_MATH,
|
||||
"print": print,
|
||||
# Numbers & reducers (pure, deterministic)
|
||||
"abs": abs,
|
||||
"round": round,
|
||||
"min": min,
|
||||
"max": max,
|
||||
"sum": sum,
|
||||
# Introspection-free basics
|
||||
"len": len,
|
||||
"sorted": sorted,
|
||||
# Basic constructors (for copy-on-write patterns)
|
||||
"tuple": tuple,
|
||||
"list": list,
|
||||
"dict": dict,
|
||||
"set": set,
|
||||
# Simple casts if they need to normalize types
|
||||
"int": int,
|
||||
"float": float,
|
||||
"str": str,
|
||||
"bool": bool,
|
||||
"bytes": bytes,
|
||||
"complex": complex,
|
||||
# Easy iteration helpers (optional but handy)
|
||||
"range": range,
|
||||
}
|
||||
|
||||
|
||||
class IElement:
|
||||
pass
|
||||
|
||||
|
||||
def _check_initializer_safety(func) -> None:
|
||||
"""
|
||||
Preliminary structural check for __initializer__.
|
||||
|
||||
Policy (minimal):
|
||||
- Forbid 'import' / 'from ... import ...' inside the initializer body.
|
||||
- Forbid nested function definitions (closures/helpers) in the body.
|
||||
- Allow lambdas.
|
||||
- No restrictions on calls here (keep it simple).
|
||||
- Optionally forbid closures (free vars) for determinism.
|
||||
"""
|
||||
try:
|
||||
src = inspect.getsource(func)
|
||||
except OSError as exc:
|
||||
# If source isn't available (rare), fail closed (or skip if you prefer)
|
||||
raise FunctionForbidden("Cannot inspect __initializer__ source") from exc
|
||||
|
||||
src = textwrap.dedent(src)
|
||||
mod = ast.parse(src)
|
||||
|
||||
# Find the FunctionDef node that corresponds to this initializer
|
||||
init_node = next(
|
||||
(
|
||||
n
|
||||
for n in mod.body
|
||||
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
and n.name == func.__name__
|
||||
),
|
||||
None,
|
||||
)
|
||||
if init_node is None:
|
||||
# Fallback: if not found, analyze nothing further to avoid false positives
|
||||
return
|
||||
|
||||
for node in ast.walk(ast.Module(body=init_node.body, type_ignores=[])):
|
||||
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
raise ImportForbidden("imports disabled in __initializer")
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
raise FunctionForbidden("Nested defs are forbidden in __initializer")
|
||||
# if isinstance(node, ast.Lambda):
|
||||
# raise FunctionForbidden("Lambdas are forbidden in __initializer")
|
||||
|
||||
# Optional: forbid closures (keeps determinism; allows lambdas that don't capture)
|
||||
if func.__code__.co_freevars:
|
||||
# Inspect captured free vars
|
||||
closure_vars = inspect.getclosurevars(func)
|
||||
captured = {**closure_vars.globals, **closure_vars.nonlocals}
|
||||
for name, val in captured.items():
|
||||
if isinstance(val, type) and issubclass(val, IElement):
|
||||
continue
|
||||
if isinstance(val, (int, str, float, bool, type(None))):
|
||||
continue
|
||||
raise FunctionForbidden(
|
||||
f"Closures are forbidden in __initializer__ (captured: {name}={val!r})"
|
||||
)
|
||||
|
||||
|
||||
def _blocked_import(*args, **kwargs):
|
||||
raise ImportForbidden("imports disabled in __initializer")
|
||||
|
||||
|
||||
def _resolve_annotation(ann):
|
||||
if isinstance(ann, str):
|
||||
# Safe eval against a **whitelist** only
|
||||
return eval( # pylint: disable=eval-used
|
||||
ann, {"__builtins__": {}}, ALLOWED_ANNOTATIONS
|
||||
)
|
||||
return ann
|
||||
|
||||
|
||||
def _peel_annotated(t: Any) -> Any:
|
||||
# If you ever allow Annotated[T, ...], peel to T
|
||||
while True:
|
||||
origin = get_origin(t)
|
||||
if origin is None:
|
||||
return t
|
||||
name = (
|
||||
getattr(origin, "__name__", "")
|
||||
or getattr(origin, "__qualname__", "")
|
||||
or str(origin)
|
||||
)
|
||||
if "Annotated" in name:
|
||||
args = get_args(t)
|
||||
t = args[0] if args else t
|
||||
else:
|
||||
return t
|
||||
|
||||
|
||||
def _check_annotation_definition( # pylint: disable=too-complex,too-many-return-statements
|
||||
_type,
|
||||
):
|
||||
|
||||
print(f"_type={_type}")
|
||||
_type = _peel_annotated(_type)
|
||||
_origin = get_origin(_type) or _type
|
||||
_args = get_args(_type)
|
||||
|
||||
# handle Optional[] and Union[None,...]
|
||||
if (_origin is Union or _origin is UnionType) and type(None) in _args:
|
||||
return all(
|
||||
_check_annotation_definition(_)
|
||||
for _ in get_args(_type)
|
||||
if _ is not type(None)
|
||||
)
|
||||
|
||||
# handle other Union[...]
|
||||
if _origin is Union or _origin is UnionType:
|
||||
return all(_check_annotation_definition(_) for _ in _args)
|
||||
|
||||
# handle Dict[...]
|
||||
if _origin is dict:
|
||||
print("a")
|
||||
if len(_args) != 2:
|
||||
raise IncompletelyAnnotatedField(
|
||||
f"Dict Annotation requires 2 inner definitions: {_type}"
|
||||
)
|
||||
if not _peel_annotated(_args[0]) in ALLOWED_MODEL_FIELDS_TYPES:
|
||||
raise IncompletelyAnnotatedField(
|
||||
f"Dict Key must be simple builtin: {_type}"
|
||||
)
|
||||
return _check_annotation_definition(_args[1])
|
||||
|
||||
# handle Tuple[]
|
||||
if _origin is tuple:
|
||||
if len(_args) == 0:
|
||||
raise IncompletelyAnnotatedField(
|
||||
f"Annotation requires inner definition: {_type}"
|
||||
)
|
||||
if len(_args) == 2 and _args[1] is Ellipsis:
|
||||
return _check_annotation_definition(_args[0])
|
||||
return all(_check_annotation_definition(_) for _ in _args)
|
||||
|
||||
# handle Set[],Tuple[],FrozenSet[],List[]
|
||||
if _origin in [set, frozenset, tuple, list]:
|
||||
if len(_args) == 0:
|
||||
raise IncompletelyAnnotatedField(
|
||||
f"Annotation requires inner definition: {_type}"
|
||||
)
|
||||
return all(_check_annotation_definition(_) for _ in _args)
|
||||
|
||||
if isinstance(_origin, type) and issubclass(_origin, IElement):
|
||||
return
|
||||
|
||||
if _type in ALLOWED_MODEL_FIELDS_TYPES:
|
||||
return
|
||||
raise UnsupportedFieldType(_type)
|
||||
|
||||
|
||||
class ModelSpecView:
|
||||
"""ModelSpecView class
|
||||
A class that will act as fake BaseElement proxy to allow setting values"""
|
||||
|
||||
__slots__ = ("_vals", "_types", "_touched", "_name", "_module")
|
||||
|
||||
def __init__(
|
||||
self, values: dict[str, Any], types_map: dict[str, type], name: str, module: str
|
||||
):
|
||||
self._name: str
|
||||
self._vals: dict[str, Any]
|
||||
self._types: dict[str, type]
|
||||
self._touched: set
|
||||
self._module: str
|
||||
object.__setattr__(self, "_vals", dict(values))
|
||||
object.__setattr__(self, "_types", types_map)
|
||||
object.__setattr__(self, "_name", name)
|
||||
object.__setattr__(self, "_module", module)
|
||||
|
||||
@property
|
||||
def __name__(self) -> str:
|
||||
"""returns proxified class' name"""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def __module__(self) -> str:
|
||||
"""returns proxified module's name"""
|
||||
return self._module
|
||||
|
||||
@__module__.setter
|
||||
def __module__(self, value: str):
|
||||
pass
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
"""internal proxy getattr"""
|
||||
if name not in self._types:
|
||||
raise AttributeError(f"Unknown field {name}")
|
||||
return self._vals[name]
|
||||
|
||||
def __setattr__(self, name: str, value: Any):
|
||||
"""internal proxy setattr"""
|
||||
if name not in self._types:
|
||||
raise NonExistingField(f"Cannot set unknown field {name}")
|
||||
T = self._types[name]
|
||||
|
||||
try:
|
||||
check_type(
|
||||
value,
|
||||
T,
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
f"Field <{name}> value is not of expected type {T}."
|
||||
) from exp
|
||||
|
||||
self._vals[name] = value
|
||||
|
||||
def export(self) -> dict:
|
||||
"""exports all proxified values"""
|
||||
return dict(self._vals)
|
||||
|
||||
|
||||
T_Meta = TypeVar("T_Meta", bound="_MetaElement")
|
||||
T_BE = TypeVar("T_BE", bound="BaseElement")
|
||||
|
||||
|
||||
class _MetaElement(type):
|
||||
"""metaclass to use to build BaseElement"""
|
||||
|
||||
modified_fields: Dict[str, Any] = {}
|
||||
new_fields: Dict[str, LAMField[Any]] = {}
|
||||
initializer: Optional[Callable[..., Any]] = None
|
||||
__LAMSchema__: dict[str, Any] = {}
|
||||
|
||||
@classmethod
|
||||
def check_class(
|
||||
mcs: type["_MetaElement"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions: dict[str, Any],
|
||||
) -> None:
|
||||
"""
|
||||
Early class-build hook.
|
||||
|
||||
Validates the inheritance shape, initializes an empty schema for root classes,
|
||||
copies the parent schema for subclasses, and ensures all annotated fields
|
||||
have a default (inserting `None` when missing).
|
||||
|
||||
This runs before the class object is created.
|
||||
"""
|
||||
# print("__NEW__ Defining:", name, "with keys:", list(namespace))
|
||||
|
||||
if len(bases) > 1:
|
||||
raise MultipleInheritanceForbidden(
|
||||
"Multiple inheritance is not supported by dabmodel"
|
||||
)
|
||||
if len(bases) == 0: # base class (Appliance,Feature)
|
||||
namespace["__LAMSchema__"] = {}
|
||||
elif (
|
||||
len(bases) == 1 and bases[0] is IElement
|
||||
): # special case for Element (hidden layer IElement)
|
||||
namespace["__LAMSchema__"] = {}
|
||||
else: # standard inheritance
|
||||
# check class tree origin
|
||||
if "__LAMSchema__" not in dir(bases[0]):
|
||||
raise BrokenInheritance(
|
||||
"__LAMSchema__ not found in base class, broken inheritance chain."
|
||||
)
|
||||
# copy inherited schema
|
||||
namespace["__LAMSchema__"] = copy(bases[0].__LAMSchema__)
|
||||
|
||||
# force field without default value to be instantiated (with None)
|
||||
if "__annotations__" in namespace:
|
||||
for _funknown in [
|
||||
_ for _ in namespace["__annotations__"] if _ not in namespace.keys()
|
||||
]:
|
||||
namespace[_funknown] = None
|
||||
|
||||
@classmethod
|
||||
def process_class_fields( # pylint: disable=too-complex,too-many-branches
|
||||
mcs: type["_MetaElement"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any],
|
||||
extensions: dict[str, Any],
|
||||
):
|
||||
"""
|
||||
Scan the class namespace and partition fields.
|
||||
|
||||
Detects:
|
||||
- modified fields (shadowing parent values),
|
||||
- new fields (present in annotations),
|
||||
- the optional `__initializer` classmethod (in mangled or unmangled form).
|
||||
|
||||
Validates annotations and types and removes processed items from `namespace`
|
||||
so they won't become normal attributes. Results are staged into:
|
||||
mcs.new_fields, mcs.modified_fields, mcs.initializer
|
||||
to be committed later.
|
||||
"""
|
||||
# iterating new and modified fields
|
||||
mcs.modified_fields = {}
|
||||
mcs.new_fields = {}
|
||||
mcs.initializer = None
|
||||
initializer_name: Optional[str] = None
|
||||
for _fname, _fvalue in namespace.items():
|
||||
if _fname == f"_{name}__initializer" or (
|
||||
name.startswith("_") and _fname == "__initializer"
|
||||
):
|
||||
if not isinstance(_fvalue, classmethod):
|
||||
raise InvalidInitializerType(
|
||||
"__initializer should be a classmethod"
|
||||
)
|
||||
mcs.initializer = _fvalue.__func__
|
||||
if name.startswith("_"):
|
||||
initializer_name = "__initializer"
|
||||
else:
|
||||
initializer_name = f"_{name}__initializer"
|
||||
elif _fname.startswith("_"):
|
||||
pass
|
||||
elif isinstance(_fvalue, classmethod):
|
||||
pass
|
||||
else:
|
||||
print(f"Parsing Field: {_fname} / {_fvalue}")
|
||||
if (
|
||||
len(bases) == 1 and _fname in namespace["__LAMSchema__"].keys()
|
||||
): # Modified fields
|
||||
mcs.process_modified_field(
|
||||
name, bases, namespace, _fname, _fvalue, extensions
|
||||
)
|
||||
else: # New fieds
|
||||
mcs.process_new_field(
|
||||
name, bases, namespace, _fname, _fvalue, extensions
|
||||
)
|
||||
# removing modified fields from class (will add them back later)
|
||||
for _fname in mcs.new_fields:
|
||||
del namespace[_fname]
|
||||
for _fname in mcs.modified_fields:
|
||||
del namespace[_fname]
|
||||
if mcs.initializer is not None and initializer_name is not None:
|
||||
del namespace[initializer_name]
|
||||
|
||||
@classmethod
|
||||
def process_modified_field(
|
||||
mcs: type["_MetaElement"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any],
|
||||
_fname: str,
|
||||
_fvalue: Any,
|
||||
extensions: dict[str, Any],
|
||||
): # pylint: disable=unused-argument
|
||||
"""
|
||||
Handle a *modified* field declared by a subclass.
|
||||
|
||||
Forbids annotation changes, validates the new default value against
|
||||
the inherited annotation, and stages the new default into `mcs.modified_fields`.
|
||||
"""
|
||||
if "__annotations__" in namespace and _fname in namespace["__annotations__"]:
|
||||
raise ReadOnlyFieldAnnotation(
|
||||
f"annotations cannot be modified on derived classes {_fname}"
|
||||
)
|
||||
try:
|
||||
check_type(
|
||||
_fvalue,
|
||||
namespace["__LAMSchema__"][_fname].annotations,
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
f"Field <{_fname}> New Field value is not of expected type {bases[0].__annotations__[_fname]}."
|
||||
) from exp
|
||||
mcs.modified_fields[_fname] = _fvalue
|
||||
|
||||
@classmethod
|
||||
def process_new_field(
|
||||
mcs: type["_MetaElement"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any],
|
||||
_fname: str,
|
||||
_fvalue: Any,
|
||||
extensions: dict[str, Any],
|
||||
): # pylint: disable=unused-argument
|
||||
"""
|
||||
Handle a *new* field declared on the class.
|
||||
|
||||
Resolves string annotations against a whitelist, validates `Annotated[...]`
|
||||
payloads (allowing only LAMFieldInfo), checks the default value type,
|
||||
and stages the field as a `LAMField` in `mcs.new_fields`.
|
||||
"""
|
||||
print(f"New field: {_fname}")
|
||||
|
||||
# check if field is annotated
|
||||
if (
|
||||
"__annotations__" not in namespace
|
||||
or _fname not in namespace["__annotations__"]
|
||||
):
|
||||
raise NotAnnotatedField(
|
||||
f"Every dabmodel Fields must be annotated ({_fname})"
|
||||
)
|
||||
|
||||
# check if annotation is allowed
|
||||
if isinstance(namespace["__annotations__"][_fname], str):
|
||||
namespace["__annotations__"][_fname] = _resolve_annotation(
|
||||
namespace["__annotations__"][_fname]
|
||||
)
|
||||
|
||||
try:
|
||||
_check_annotation_definition(namespace["__annotations__"][_fname])
|
||||
except InvalidFieldAnnotation:
|
||||
raise
|
||||
except Exception as ex:
|
||||
raise InvalidFieldAnnotation(
|
||||
f"Field <{_fname}> has not an allowed or valid annotation.", ex
|
||||
)
|
||||
|
||||
_finfo: LAMFieldInfo = LAMFieldInfo()
|
||||
origin = get_origin(namespace["__annotations__"][_fname])
|
||||
tname = (
|
||||
getattr(origin, "__name__", "")
|
||||
or getattr(origin, "__qualname__", "")
|
||||
or str(origin)
|
||||
)
|
||||
if "Annotated" in tname:
|
||||
args = get_args(namespace["__annotations__"][_fname])
|
||||
if args:
|
||||
if len(args) > 2:
|
||||
raise InvalidFieldAnnotation(
|
||||
f"Field <{_fname}> had invalid Annotated value."
|
||||
)
|
||||
if len(args) == 2 and not isinstance(args[1], LAMFieldInfo):
|
||||
raise InvalidFieldAnnotation(
|
||||
"Only LAMFieldInfo object is allowed as Annotated data."
|
||||
)
|
||||
|
||||
_finfo = args[1]
|
||||
|
||||
# check if value is valid
|
||||
try:
|
||||
check_type(
|
||||
_fvalue,
|
||||
namespace["__annotations__"][_fname],
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
f"Value of Field <{_fname}> is not of expected type {namespace['__annotations__'][_fname]}."
|
||||
) from exp
|
||||
print(f"!!VAL: {_fvalue}")
|
||||
mcs.new_fields[_fname] = LAMField(
|
||||
_fname, _fvalue, namespace["__annotations__"][_fname], _finfo
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def apply_initializer(
|
||||
mcs: type["_MetaElement"],
|
||||
cls,
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions: dict[str, Any],
|
||||
):
|
||||
"""
|
||||
Apply the optional `__initializer` classmethod to compute derived defaults.
|
||||
|
||||
The initializer runs in a restricted, import-blocked environment using a
|
||||
`ModelSpecView` proxy that enforces type checking on assignments.
|
||||
On success, the computed values are validated and written back into the
|
||||
class schema's DABFields.
|
||||
"""
|
||||
if mcs.initializer is not None:
|
||||
_check_initializer_safety(mcs.initializer)
|
||||
init_fieldvalues = {}
|
||||
init_fieldtypes = {}
|
||||
for _fname, _fvalue in cls.__LAMSchema__.items():
|
||||
if isinstance(_fvalue, LAMField):
|
||||
init_fieldvalues[_fname] = deepcopy(_fvalue.raw_value)
|
||||
init_fieldtypes[_fname] = _fvalue.annotations
|
||||
fakecls = ModelSpecView(
|
||||
init_fieldvalues, init_fieldtypes, cls.__name__, cls.__module__
|
||||
)
|
||||
safe_globals = {
|
||||
"__builtins__": {"__import__": _blocked_import},
|
||||
**ALLOWED_HELPERS_DEFAULT,
|
||||
}
|
||||
# if mcs.initializer.__code__.co_freevars:
|
||||
# raise FunctionForbidden("__initializer must not use closures")
|
||||
safe_initializer = FunctionType(
|
||||
mcs.initializer.__code__,
|
||||
safe_globals,
|
||||
name=mcs.initializer.__name__,
|
||||
argdefs=mcs.initializer.__defaults__,
|
||||
closure=mcs.initializer.__closure__,
|
||||
)
|
||||
safe_initializer(fakecls) # pylint: disable=not-callable
|
||||
for _fname, _fvalue in fakecls.export().items():
|
||||
try:
|
||||
check_type(
|
||||
_fvalue,
|
||||
cls.__LAMSchema__[_fname].annotations,
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
f"Value of Field <{_fname}> is not of expected type {namespace['__annotations__'][_fname]}."
|
||||
) from exp
|
||||
cls.__LAMSchema__[_fname] = deepcopy(cls.__LAMSchema__[_fname])
|
||||
cls.__LAMSchema__[_fname].update_value(_fvalue)
|
||||
|
||||
def __new__(
|
||||
mcs: type["_MetaElement"],
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any],
|
||||
) -> Type:
|
||||
"""BaseElement new class"""
|
||||
extensions: dict[str, Any] = {}
|
||||
mcs.check_class(name, bases, namespace, extensions)
|
||||
mcs.process_class_fields(name, bases, namespace, extensions)
|
||||
|
||||
_cls = super().__new__(mcs, name, bases, namespace)
|
||||
|
||||
mcs.commit_fields(_cls, name, bases, namespace, extensions)
|
||||
mcs.apply_initializer(_cls, name, bases, namespace, extensions)
|
||||
_cls.install_instance_guard(extensions)
|
||||
|
||||
return _cls
|
||||
|
||||
@classmethod
|
||||
def commit_fields(
|
||||
mcs: type["_MetaElement"],
|
||||
cls,
|
||||
name: str,
|
||||
bases: tuple[type[Any], ...],
|
||||
namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions: dict[str, Any],
|
||||
):
|
||||
"""
|
||||
Commit staged fields into the class schema (`__LAMSchema__`).
|
||||
|
||||
- For modified fields: copy the parent's LAMField, update its value.
|
||||
- For new fields: set the freshly built LAMField and record its source.
|
||||
"""
|
||||
for _fname, _fvalue in mcs.modified_fields.items():
|
||||
cls.__LAMSchema__[_fname] = deepcopy(bases[0].__LAMSchema__[_fname])
|
||||
cls.__LAMSchema__[_fname].update_value(_fvalue)
|
||||
|
||||
for _fname, _fvalue in mcs.new_fields.items():
|
||||
_fvalue.add_source(cls)
|
||||
cls.__LAMSchema__[_fname] = _fvalue
|
||||
|
||||
def __call__(cls: Type, *args: Any, **kw: Any): # intentionally untyped
|
||||
"""BaseElement new instance"""
|
||||
obj = super().__call__(*args)
|
||||
|
||||
extensions: dict[str, Any] = {}
|
||||
|
||||
cls.populate_instance( # pylint: disable=no-value-for-parameter
|
||||
obj, extensions, *args, **kw
|
||||
)
|
||||
cls.freeze_instance_schema( # pylint: disable=no-value-for-parameter
|
||||
obj, extensions, *args, **kw
|
||||
)
|
||||
cls.apply_overrides( # pylint: disable=no-value-for-parameter
|
||||
obj, extensions, *args, **kw
|
||||
)
|
||||
cls.finalize_instance(obj, extensions) # pylint: disable=no-value-for-parameter
|
||||
|
||||
return obj
|
||||
|
||||
def populate_instance(
|
||||
cls: Type, obj: Any, extensions: dict[str, Any], *args: Any, **kw: Any
|
||||
):
|
||||
"""
|
||||
Populate the new instance with field values from the class schema.
|
||||
|
||||
Copies each LAMField.value to an instance attribute (deep-frozen view).
|
||||
"""
|
||||
for _fname, _fvalue in cls.__LAMSchema__.items():
|
||||
if isinstance(_fvalue, LAMField):
|
||||
object.__setattr__(obj, _fname, _fvalue.value)
|
||||
|
||||
def freeze_instance_schema(
|
||||
cls: Type, obj: Any, extensions: dict[str, Any], *args: Any, **kw: Any
|
||||
):
|
||||
"""
|
||||
Freeze the instance's schema by wrapping DABFields into FrozenLAMField.
|
||||
|
||||
Creates a per-instance `__LAMSchema__` dict where each field is read-only.
|
||||
"""
|
||||
inst_schema = copy(obj.__LAMSchema__)
|
||||
for _fname, _fvalue in cls.__LAMSchema__.items():
|
||||
if isinstance(_fvalue, LAMField):
|
||||
inst_schema[_fname] = FrozenLAMField(_fvalue)
|
||||
|
||||
if "features" in inst_schema:
|
||||
inst_schema["features"] = dict(inst_schema["features"])
|
||||
|
||||
object.__setattr__(obj, "__LAMSchema__", inst_schema)
|
||||
|
||||
def apply_overrides(cls, obj, extensions, *args, **kwargs):
|
||||
"""
|
||||
Hook for runtime overrides at instance creation.
|
||||
|
||||
Invoked after the schema has been frozen but before finalize_instance.
|
||||
Subclasses of _MetaElement can override this to support things like:
|
||||
|
||||
- Field overrides: MyApp(field=value)
|
||||
"""
|
||||
|
||||
# --- field overrides (unchanged) ---
|
||||
for k, v in list(kwargs.items()):
|
||||
if k in cls.__LAMSchema__: # regular field
|
||||
field = cls.__LAMSchema__[k]
|
||||
try:
|
||||
check_type(
|
||||
v,
|
||||
field.annotations,
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
f"Invalid value for field '{k}': expected {field.annotations}, got {v!r}"
|
||||
) from exp
|
||||
|
||||
object.__setattr__(obj, k, LAMdeepfreeze(v))
|
||||
obj.__LAMSchema__[k] = FrozenLAMField(
|
||||
LAMField(k, v, field.annotations, field._info)
|
||||
)
|
||||
kwargs.pop(k)
|
||||
|
||||
if kwargs:
|
||||
unknown = ", ".join(sorted(kwargs.keys()))
|
||||
raise InvalidFieldValue(f"Unknown parameters: {unknown}")
|
||||
|
||||
def finalize_instance(cls: Type, obj: Any, extensions: dict[str, Any]):
|
||||
"""
|
||||
Finalization hook invoked at the end of instance construction.
|
||||
|
||||
Subclasses of the metaclass override this to attach runtime components
|
||||
to the instance. (Example: BaseMetaAppliance instantiates bound Features
|
||||
and sets them as attributes on the appliance instance.)
|
||||
"""
|
||||
|
||||
def install_instance_guard(cls: Type, extensions: dict[str, Any]):
|
||||
"""
|
||||
Install the runtime `__setattr__` guard on the class.
|
||||
|
||||
After instances are constructed, prevents:
|
||||
- creating new public fields,
|
||||
- reassigning existing fields post-initialization.
|
||||
|
||||
Private/dunder attributes are exempt to allow internal bookkeeping.
|
||||
"""
|
||||
orig_setattr = getattr(cls, "__setattr__")
|
||||
|
||||
# cls.orig_setattr = orig_setattr
|
||||
|
||||
def guarded_setattr(_self, key: str, value: Any):
|
||||
if key.startswith("_"): # allow private and dunder attrs
|
||||
return orig_setattr(_self, key, value)
|
||||
# block writes after init if key is readonly
|
||||
if key in _self.__LAMSchema__.keys():
|
||||
if key in _self.__dict__:
|
||||
raise ReadOnlyField(f"{key} is read-only")
|
||||
# elif key in _self.__LAMSchema__["features"].keys():
|
||||
# if key in _self.__dict__:
|
||||
# raise ReadOnlyField(f"{key} is read-only")
|
||||
else:
|
||||
raise NewFieldForbidden("creating new fields is not allowed")
|
||||
|
||||
return orig_setattr(_self, key, value)
|
||||
|
||||
setattr(cls, "__setattr__", guarded_setattr)
|
||||
19
src/dabmodel/meta/feature.py
Normal file
19
src/dabmodel/meta/feature.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from typing import Type, Any
|
||||
from .element import _MetaElement
|
||||
from ..exception import FeatureNotBound
|
||||
|
||||
|
||||
class _MetaFeature(_MetaElement):
|
||||
"""_MetaFeature class
|
||||
Feature specific metaclass code
|
||||
"""
|
||||
|
||||
def __call__(cls: Type, *args: Any, **kw: Any): # intentionally untyped
|
||||
"""BaseFeature new instance"""
|
||||
|
||||
if cls._BoundAppliance is None:
|
||||
raise FeatureNotBound()
|
||||
|
||||
obj = super().__call__(*args, **kw)
|
||||
|
||||
return obj
|
||||
@@ -1,43 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# dabmodel (c) by chacha
|
||||
#
|
||||
# dabmodel is licensed under a
|
||||
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
|
||||
#
|
||||
# You should have received a copy of the license along with this
|
||||
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
|
||||
|
||||
"""Phasellus tellus lectus, volutpat eu dapibus ut, suscipit vel augue.
|
||||
|
||||
Tips:
|
||||
Aliquam non leo vel libero sagittis viverra. Quisque lobortis nunc sit amet augue euismod laoreet.
|
||||
Note:
|
||||
Maecenas volutpat porttitor pretium. Aliquam suscipit quis nisi non imperdiet.
|
||||
Note:
|
||||
Vivamus et efficitur lorem, eget imperdiet tortor. Integer vel interdum sem.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING: # Only imports the below statements during type checking
|
||||
pass
|
||||
|
||||
def test_function(testvar: int) -> int:
|
||||
""" A test function that return testvar+1 and print "Hello world !"
|
||||
|
||||
Proin eget sapien eget ipsum efficitur mollis nec ac nibh.
|
||||
|
||||
Note:
|
||||
Morbi id lectus maximus, condimentum nunc eget, porta felis. In tristique velit tortor.
|
||||
|
||||
Args:
|
||||
testvar: any integer
|
||||
|
||||
Returns:
|
||||
testvar+1
|
||||
"""
|
||||
print("Hello world !")
|
||||
return testvar+1
|
||||
36
src/dabmodel/tools.py
Normal file
36
src/dabmodel/tools.py
Normal file
@@ -0,0 +1,36 @@
|
||||
"""library's internal tools"""
|
||||
|
||||
from typing import Union, List, Any, Dict
|
||||
from uuid import UUID
|
||||
from datetime import datetime
|
||||
import json
|
||||
|
||||
from frozendict import deepfreeze
|
||||
|
||||
JSONPrimitive = Union[str, int, float, bool, None]
|
||||
JSONType = Union[JSONPrimitive, List[Any], Dict[str, Any]] # recursive in practice
|
||||
|
||||
|
||||
class LAMJSONEncoder(json.JSONEncoder):
|
||||
"""allows to JSON encode non supported data type"""
|
||||
|
||||
def default(self, o):
|
||||
if isinstance(o, UUID):
|
||||
# if the o is uuid, we simply return the value of uuid
|
||||
return o.hex
|
||||
if isinstance(o, datetime):
|
||||
return str(o)
|
||||
return json.JSONEncoder.default(self, o)
|
||||
|
||||
|
||||
def LAMdeepfreeze(value):
|
||||
"""recursive freeze helper function"""
|
||||
if isinstance(value, dict):
|
||||
return deepfreeze(value)
|
||||
if isinstance(value, set):
|
||||
return frozenset(LAMdeepfreeze(v) for v in value)
|
||||
if isinstance(value, list):
|
||||
return tuple(LAMdeepfreeze(v) for v in value)
|
||||
if isinstance(value, tuple):
|
||||
return tuple(LAMdeepfreeze(v) for v in value)
|
||||
return value
|
||||
2019
test/test_appliance.py
Normal file
2019
test/test_appliance.py
Normal file
File diff suppressed because it is too large
Load Diff
293
test/test_element.py
Normal file
293
test/test_element.py
Normal file
@@ -0,0 +1,293 @@
|
||||
# dabmodel (c) by chacha
|
||||
#
|
||||
# dabmodel is licensed under a
|
||||
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
|
||||
#
|
||||
# You should have received a copy of the license along with this
|
||||
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
|
||||
|
||||
import unittest
|
||||
|
||||
import sys
|
||||
import subprocess
|
||||
from os import chdir, environ
|
||||
from pathlib import Path
|
||||
|
||||
print(__name__)
|
||||
print(__package__)
|
||||
|
||||
from src import dabmodel as dm
|
||||
|
||||
testdir_path = Path(__file__).parent.resolve()
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
|
||||
class ElementTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
print("\n->", unittest.TestCase.id(self))
|
||||
|
||||
def test_element_simple(self):
|
||||
class E(dm.Element):
|
||||
ivalue: int = 43
|
||||
strvalue: str = "test"
|
||||
fvalue: float = 1.4322
|
||||
ar_int: list[int] = [1, 54, 65]
|
||||
ar_int2: list[int] = [1, 54, 65]
|
||||
|
||||
class A(dm.Appliance):
|
||||
elem: E = E(ivalue=45, strvalue="coucou", ar_int=[5, 7])
|
||||
|
||||
a = A()
|
||||
|
||||
self.assertIsInstance(a.elem, E)
|
||||
self.assertIsInstance(a.elem.ivalue, int)
|
||||
self.assertEqual(a.elem.ivalue, 45)
|
||||
self.assertIsInstance(a.elem.strvalue, str)
|
||||
self.assertEqual(a.elem.strvalue, "coucou")
|
||||
self.assertIsInstance(a.elem.fvalue, float)
|
||||
self.assertEqual(a.elem.fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elem.ar_int, tuple)
|
||||
self.assertEqual(a.elem.ar_int, (5, 7))
|
||||
self.assertIsInstance(a.elem.ar_int2, tuple)
|
||||
self.assertEqual(a.elem.ar_int2, (1, 54, 65))
|
||||
|
||||
def test_element_in_container(self):
|
||||
class E(dm.Element):
|
||||
ivalue: int = 43
|
||||
strvalue: str = "test"
|
||||
fvalue: float = 1.4322
|
||||
ar_int: list[int] = [1, 54, 65]
|
||||
ar_int2: list[int] = [1, 54, 65]
|
||||
|
||||
class A(dm.Appliance):
|
||||
elems: list[E] = [
|
||||
E(ivalue=45, strvalue="coucou", ar_int=[5, 7]),
|
||||
E(ivalue=46, strvalue="coucou2", ar_int=[50, 7]),
|
||||
]
|
||||
|
||||
a = A()
|
||||
|
||||
self.assertIsInstance(a.elems, tuple)
|
||||
self.assertEqual(len(a.elems), 2)
|
||||
|
||||
self.assertIsInstance(a.elems[0], E)
|
||||
self.assertIsInstance(a.elems[0].ivalue, int)
|
||||
self.assertEqual(a.elems[0].ivalue, 45)
|
||||
self.assertIsInstance(a.elems[0].strvalue, str)
|
||||
self.assertEqual(a.elems[0].strvalue, "coucou")
|
||||
self.assertIsInstance(a.elems[0].fvalue, float)
|
||||
self.assertEqual(a.elems[0].fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elems[0].ar_int, tuple)
|
||||
self.assertEqual(a.elems[0].ar_int, (5, 7))
|
||||
self.assertIsInstance(a.elems[0].ar_int2, tuple)
|
||||
self.assertEqual(a.elems[0].ar_int2, (1, 54, 65))
|
||||
|
||||
self.assertIsInstance(a.elems[1], E)
|
||||
self.assertIsInstance(a.elems[1].ivalue, int)
|
||||
self.assertEqual(a.elems[1].ivalue, 46)
|
||||
self.assertIsInstance(a.elems[1].strvalue, str)
|
||||
self.assertEqual(a.elems[1].strvalue, "coucou2")
|
||||
self.assertIsInstance(a.elems[1].fvalue, float)
|
||||
self.assertEqual(a.elems[1].fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elems[1].ar_int, tuple)
|
||||
self.assertEqual(a.elems[1].ar_int, (50, 7))
|
||||
self.assertIsInstance(a.elems[1].ar_int2, tuple)
|
||||
self.assertEqual(a.elems[1].ar_int2, (1, 54, 65))
|
||||
|
||||
def test_element_frozen(self):
|
||||
class E(dm.Element):
|
||||
ivalue: int = 43
|
||||
strvalue: str = "test"
|
||||
fvalue: float = 1.4322
|
||||
ar_int: list[int] = [1, 54, 65]
|
||||
ar_int2: list[int] = [1, 54, 65]
|
||||
|
||||
class A(dm.Appliance):
|
||||
elems: list[E] = [
|
||||
E(ivalue=45, strvalue="coucou", ar_int=[5, 7]),
|
||||
E(ivalue=46, strvalue="coucou2", ar_int=[50, 7]),
|
||||
]
|
||||
elem: E = E()
|
||||
|
||||
a = A()
|
||||
|
||||
with self.assertRaises(AttributeError):
|
||||
a.elems.add(E())
|
||||
|
||||
with self.assertRaises(dm.ReadOnlyField):
|
||||
a.elem.ivalue = 1
|
||||
|
||||
with self.assertRaises(dm.ReadOnlyField):
|
||||
a.elems[0].ivalue = 1
|
||||
|
||||
def test_element_inheritance(self):
|
||||
class E(dm.Element):
|
||||
ivalue: int = 43
|
||||
strvalue: str = "test"
|
||||
fvalue: float = 1.4322
|
||||
ar_int: list[int] = [1, 54, 65]
|
||||
ar_int2: list[int] = [1, 54, 65]
|
||||
|
||||
class E2(E):
|
||||
ivalue2: int = 43
|
||||
|
||||
class A(dm.Appliance):
|
||||
elems: list[E] = [
|
||||
E(ivalue=45, strvalue="coucou", ar_int=[5, 7]),
|
||||
E2(ivalue=46, strvalue="coucou2", ar_int=[50, 7], ivalue2=32),
|
||||
]
|
||||
elem: E = E()
|
||||
elem2: E2 = E2(ivalue=7, ivalue2=33)
|
||||
|
||||
a = A()
|
||||
|
||||
self.assertIsInstance(a.elems, tuple)
|
||||
self.assertEqual(len(a.elems), 2)
|
||||
|
||||
self.assertIsInstance(a.elems[0], E)
|
||||
self.assertIsInstance(a.elems[0].ivalue, int)
|
||||
self.assertEqual(a.elems[0].ivalue, 45)
|
||||
self.assertIsInstance(a.elems[0].strvalue, str)
|
||||
self.assertEqual(a.elems[0].strvalue, "coucou")
|
||||
self.assertIsInstance(a.elems[0].fvalue, float)
|
||||
self.assertEqual(a.elems[0].fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elems[0].ar_int, tuple)
|
||||
self.assertEqual(a.elems[0].ar_int, (5, 7))
|
||||
self.assertIsInstance(a.elems[0].ar_int2, tuple)
|
||||
self.assertEqual(a.elems[0].ar_int2, (1, 54, 65))
|
||||
|
||||
self.assertIsInstance(a.elems[1], E2)
|
||||
self.assertIsInstance(a.elems[1].ivalue, int)
|
||||
self.assertEqual(a.elems[1].ivalue, 46)
|
||||
self.assertIsInstance(a.elems[1].ivalue2, int)
|
||||
self.assertEqual(a.elems[1].ivalue2, 32)
|
||||
self.assertIsInstance(a.elems[1].strvalue, str)
|
||||
self.assertEqual(a.elems[1].strvalue, "coucou2")
|
||||
self.assertIsInstance(a.elems[1].fvalue, float)
|
||||
self.assertEqual(a.elems[1].fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elems[1].ar_int, tuple)
|
||||
self.assertEqual(a.elems[1].ar_int, (50, 7))
|
||||
self.assertIsInstance(a.elems[1].ar_int2, tuple)
|
||||
self.assertEqual(a.elems[1].ar_int2, (1, 54, 65))
|
||||
|
||||
self.assertIsInstance(a.elem, E)
|
||||
self.assertIsInstance(a.elem.ivalue, int)
|
||||
self.assertEqual(a.elem.ivalue, 43)
|
||||
self.assertIsInstance(a.elem.strvalue, str)
|
||||
self.assertEqual(a.elem.strvalue, "test")
|
||||
self.assertIsInstance(a.elem.fvalue, float)
|
||||
self.assertEqual(a.elem.fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elem.ar_int, tuple)
|
||||
self.assertEqual(a.elem.ar_int, (1, 54, 65))
|
||||
self.assertIsInstance(a.elem.ar_int2, tuple)
|
||||
self.assertEqual(a.elem.ar_int2, (1, 54, 65))
|
||||
|
||||
self.assertIsInstance(a.elem2, E2)
|
||||
self.assertIsInstance(a.elem2.ivalue, int)
|
||||
self.assertEqual(a.elem2.ivalue, 7)
|
||||
self.assertIsInstance(a.elem2.ivalue2, int)
|
||||
self.assertEqual(a.elem2.ivalue2, 33)
|
||||
self.assertIsInstance(a.elem2.strvalue, str)
|
||||
self.assertEqual(a.elem2.strvalue, "test")
|
||||
self.assertIsInstance(a.elem2.fvalue, float)
|
||||
self.assertEqual(a.elem2.fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elem2.ar_int, tuple)
|
||||
self.assertEqual(a.elem2.ar_int, (1, 54, 65))
|
||||
self.assertIsInstance(a.elem2.ar_int2, tuple)
|
||||
self.assertEqual(a.elem2.ar_int2, (1, 54, 65))
|
||||
|
||||
def test_element_initializer(self):
|
||||
class E(dm.Element):
|
||||
ivalue: int = 43
|
||||
strvalue: str = "test"
|
||||
fvalue: float = 1.4322
|
||||
ar_int: list[int] = [1, 54, 65]
|
||||
ar_int2: list[int] = [1, 54, 65]
|
||||
|
||||
class A(dm.Appliance):
|
||||
elem: E = E(ivalue=45, strvalue="coucou", ar_int=[5, 7])
|
||||
|
||||
@classmethod
|
||||
def __initializer(self):
|
||||
self.elem = E(ivalue=12, strvalue="coucou", ar_int=[5, 7])
|
||||
|
||||
a = A()
|
||||
|
||||
self.assertIsInstance(a.elem, E)
|
||||
self.assertIsInstance(a.elem.ivalue, int)
|
||||
self.assertEqual(a.elem.ivalue, 12)
|
||||
self.assertIsInstance(a.elem.strvalue, str)
|
||||
self.assertEqual(a.elem.strvalue, "coucou")
|
||||
self.assertIsInstance(a.elem.fvalue, float)
|
||||
self.assertEqual(a.elem.fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elem.ar_int, tuple)
|
||||
self.assertEqual(a.elem.ar_int, (5, 7))
|
||||
self.assertIsInstance(a.elem.ar_int2, tuple)
|
||||
self.assertEqual(a.elem.ar_int2, (1, 54, 65))
|
||||
|
||||
def test_element_in_container_initializer(self):
|
||||
class E(dm.Element):
|
||||
ivalue: int = 43
|
||||
strvalue: str = "test"
|
||||
fvalue: float = 1.4322
|
||||
ar_int: list[int] = [1, 54, 65]
|
||||
ar_int2: list[int] = [1, 54, 65]
|
||||
|
||||
class A(dm.Appliance):
|
||||
elems: list[E] = [E(ivalue=45, strvalue="coucou", ar_int=[5, 7])]
|
||||
|
||||
class B(A):
|
||||
@classmethod
|
||||
def __initializer(cls):
|
||||
cls.elems.append(E(ivalue=46, strvalue="coucou2", ar_int=[50, 7]))
|
||||
|
||||
a = A()
|
||||
b = B()
|
||||
|
||||
self.assertIsInstance(a.elems, tuple)
|
||||
self.assertEqual(len(a.elems), 1)
|
||||
|
||||
self.assertIsInstance(a.elems[0], E)
|
||||
self.assertIsInstance(a.elems[0].ivalue, int)
|
||||
self.assertEqual(a.elems[0].ivalue, 45)
|
||||
self.assertIsInstance(a.elems[0].strvalue, str)
|
||||
self.assertEqual(a.elems[0].strvalue, "coucou")
|
||||
self.assertIsInstance(a.elems[0].fvalue, float)
|
||||
self.assertEqual(a.elems[0].fvalue, 1.4322)
|
||||
self.assertIsInstance(a.elems[0].ar_int, tuple)
|
||||
self.assertEqual(a.elems[0].ar_int, (5, 7))
|
||||
self.assertIsInstance(a.elems[0].ar_int2, tuple)
|
||||
self.assertEqual(a.elems[0].ar_int2, (1, 54, 65))
|
||||
|
||||
self.assertIsInstance(b.elems, tuple)
|
||||
self.assertEqual(len(b.elems), 2)
|
||||
|
||||
self.assertIsInstance(b.elems[0], E)
|
||||
self.assertIsInstance(b.elems[0].ivalue, int)
|
||||
self.assertEqual(b.elems[0].ivalue, 45)
|
||||
self.assertIsInstance(b.elems[0].strvalue, str)
|
||||
self.assertEqual(b.elems[0].strvalue, "coucou")
|
||||
self.assertIsInstance(b.elems[0].fvalue, float)
|
||||
self.assertEqual(b.elems[0].fvalue, 1.4322)
|
||||
self.assertIsInstance(b.elems[0].ar_int, tuple)
|
||||
self.assertEqual(b.elems[0].ar_int, (5, 7))
|
||||
self.assertIsInstance(b.elems[0].ar_int2, tuple)
|
||||
self.assertEqual(b.elems[0].ar_int2, (1, 54, 65))
|
||||
|
||||
self.assertIsInstance(b.elems[1], E)
|
||||
self.assertIsInstance(b.elems[1].ivalue, int)
|
||||
self.assertEqual(b.elems[1].ivalue, 46)
|
||||
self.assertIsInstance(b.elems[1].strvalue, str)
|
||||
self.assertEqual(b.elems[1].strvalue, "coucou2")
|
||||
self.assertIsInstance(b.elems[1].fvalue, float)
|
||||
self.assertEqual(b.elems[1].fvalue, 1.4322)
|
||||
self.assertIsInstance(b.elems[1].ar_int, tuple)
|
||||
self.assertEqual(b.elems[1].ar_int, (50, 7))
|
||||
self.assertIsInstance(b.elems[1].ar_int2, tuple)
|
||||
self.assertEqual(b.elems[1].ar_int2, (1, 54, 65))
|
||||
|
||||
|
||||
# ---------- main ----------
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
730
test/test_feature.py
Normal file
730
test/test_feature.py
Normal file
@@ -0,0 +1,730 @@
|
||||
# dabmodel (c) by chacha
|
||||
#
|
||||
# dabmodel is licensed under a
|
||||
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
|
||||
#
|
||||
# You should have received a copy of the license along with this
|
||||
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
|
||||
|
||||
import unittest
|
||||
|
||||
from os import chdir
|
||||
from pathlib import Path
|
||||
|
||||
from typing import (
|
||||
Any,
|
||||
Annotated,
|
||||
)
|
||||
|
||||
|
||||
print(__name__)
|
||||
print(__package__)
|
||||
|
||||
from src import dabmodel as dm
|
||||
|
||||
|
||||
testdir_path = Path(__file__).parent.resolve()
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
|
||||
def test_initializer_safe_testfc():
|
||||
eval("print('hi')")
|
||||
|
||||
|
||||
class FeatureTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
print("\n->", unittest.TestCase.id(self))
|
||||
|
||||
def immutable_vars__test_field(
|
||||
self, obj: Any, name: str, default_value: Any, test_value: Any
|
||||
):
|
||||
# field is not in the class
|
||||
self.assertNotIn(name, dir(obj.__class__))
|
||||
# field is in the object
|
||||
self.assertIn(name, dir(obj))
|
||||
# field is in the schema
|
||||
self.assertIn(name, obj.__LAMSchema__.keys())
|
||||
# field is readable
|
||||
self.assertEqual(getattr(obj, name), default_value)
|
||||
# field is read only
|
||||
with self.assertRaises(dm.ReadOnlyField):
|
||||
setattr(obj, name, test_value)
|
||||
|
||||
def test_simple(self):
|
||||
"""Testing first appliance feature, and Field types (simple)"""
|
||||
|
||||
# class can be created
|
||||
class Appliance1(dm.Appliance):
|
||||
VarStrOuter: str = "testvalue APPLIANCE"
|
||||
|
||||
class Feature1(dm.Feature):
|
||||
VarStrInner: str = "testvalue FEATURE"
|
||||
|
||||
app1 = Appliance1()
|
||||
|
||||
self.assertIsInstance(Appliance1.__LAMSchema__["VarStrOuter"], dm.LAMField)
|
||||
self.assertIsInstance(app1.__LAMSchema__["VarStrOuter"], dm.FrozenLAMField)
|
||||
self.assertIn("Feature1", app1.__LAMSchema__["features"])
|
||||
self.assertIn(
|
||||
"VarStrInner", app1.__LAMSchema__["features"]["Feature1"].__LAMSchema__
|
||||
)
|
||||
self.assertIsInstance(
|
||||
app1.__LAMSchema__["features"]["Feature1"].__LAMSchema__["VarStrInner"],
|
||||
dm.LAMField,
|
||||
)
|
||||
self.assertTrue(hasattr(app1, "Feature1"))
|
||||
self.assertIsInstance(
|
||||
app1.Feature1.__LAMSchema__["VarStrInner"], dm.FrozenLAMField
|
||||
)
|
||||
self.assertTrue(hasattr(app1.Feature1, "VarStrInner"))
|
||||
|
||||
def test_inheritance(self):
|
||||
"""Testing first appliance feature, and Field types (simple)"""
|
||||
|
||||
# class can be created
|
||||
class Appliance1(dm.Appliance):
|
||||
VarStrOuter: str = "testvalue APPLIANCE1"
|
||||
|
||||
class Feature1(dm.Feature):
|
||||
VarStrInner: str = "testvalue FEATURE1"
|
||||
VarInt: int = 42
|
||||
|
||||
print(dir(Appliance1))
|
||||
|
||||
class Appliance2(Appliance1):
|
||||
VarStrOuter = "testvalue APPLIANCE2"
|
||||
|
||||
class Feature2(dm.Feature):
|
||||
VarStrInner: str = "testvalue FEATURE2"
|
||||
|
||||
print(dir(Appliance2))
|
||||
|
||||
class Appliance3(Appliance2):
|
||||
VarStrOuter = "testvalue APPLIANCE3"
|
||||
|
||||
class Feature1(Appliance1.Feature1):
|
||||
VarStrInner = "testvalue FEATURE1 modded"
|
||||
|
||||
class Feature3(dm.Feature):
|
||||
VarStrInner: str = "testvalue FEATURE3"
|
||||
|
||||
print(dir(Appliance3))
|
||||
|
||||
app1 = Appliance1()
|
||||
app2 = Appliance2()
|
||||
app3 = Appliance3()
|
||||
|
||||
self.assertIsInstance(Appliance1.__LAMSchema__["VarStrOuter"], dm.LAMField)
|
||||
self.assertIsInstance(app1.__LAMSchema__["VarStrOuter"], dm.FrozenLAMField)
|
||||
self.assertIn("Feature1", app1.__LAMSchema__["features"])
|
||||
self.assertIn(
|
||||
"VarStrInner", app1.__LAMSchema__["features"]["Feature1"].__LAMSchema__
|
||||
)
|
||||
self.assertIsInstance(
|
||||
app1.__LAMSchema__["features"]["Feature1"].__LAMSchema__["VarStrInner"],
|
||||
dm.LAMField,
|
||||
)
|
||||
self.assertTrue(hasattr(app1, "Feature1"))
|
||||
self.assertIsInstance(
|
||||
app1.Feature1.__LAMSchema__["VarStrInner"], dm.FrozenLAMField
|
||||
)
|
||||
self.assertTrue(hasattr(app1.Feature1, "VarStrInner"))
|
||||
self.assertEqual(app1.VarStrOuter, "testvalue APPLIANCE1")
|
||||
self.assertEqual(app1.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app1.Feature1.VarInt, 42)
|
||||
self.assertEqual(app2.VarStrOuter, "testvalue APPLIANCE2")
|
||||
self.assertEqual(app2.Feature2.VarStrInner, "testvalue FEATURE2")
|
||||
self.assertEqual(app3.VarStrOuter, "testvalue APPLIANCE3")
|
||||
self.assertEqual(app3.Feature1.VarStrInner, "testvalue FEATURE1 modded")
|
||||
self.assertEqual(app3.Feature1.VarInt, 42)
|
||||
self.assertEqual(app3.Feature3.VarStrInner, "testvalue FEATURE3")
|
||||
|
||||
def test_inheritance2(self):
|
||||
"""Testing first appliance feature, and Field types (simple)"""
|
||||
|
||||
# class can be created
|
||||
class Appliance1(dm.Appliance):
|
||||
class Feature1(dm.Feature):
|
||||
VarStrInner: str = "testvalue FEATURE1"
|
||||
|
||||
# check cannot REdefine a feature from Feature
|
||||
with self.assertRaises(dm.InvalidFeatureInheritance):
|
||||
|
||||
class Appliance2(Appliance1):
|
||||
class Feature1(dm.Feature): ...
|
||||
|
||||
class Appliance2b(Appliance1):
|
||||
class Feature1(Appliance1.Feature1): ...
|
||||
|
||||
# check only REdefine a feature from highest parent
|
||||
with self.assertRaises(dm.InvalidFeatureInheritance):
|
||||
|
||||
class Appliance3(Appliance2b):
|
||||
class Feature1(Appliance1.Feature1): ...
|
||||
|
||||
class Appliance3b(Appliance2b):
|
||||
class Feature1(Appliance2b.Feature1): ...
|
||||
|
||||
app1 = Appliance1()
|
||||
app2 = Appliance2b()
|
||||
app3 = Appliance3b()
|
||||
|
||||
self.assertEqual(app1.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app2.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app3.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
|
||||
class Appliance4(Appliance3b):
|
||||
class Feature1(Appliance3b.Feature1):
|
||||
VarStrInner = "testvalue FEATURE4"
|
||||
|
||||
self.assertEqual(app1.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app2.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app3.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
|
||||
app4 = Appliance4()
|
||||
|
||||
self.assertEqual(app1.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app2.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app3.Feature1.VarStrInner, "testvalue FEATURE1")
|
||||
self.assertEqual(app4.Feature1.VarStrInner, "testvalue FEATURE4")
|
||||
|
||||
def test_register(self):
|
||||
"""Testing first appliance feature, and Field types (simple)"""
|
||||
|
||||
# class can be created
|
||||
class Appliance1(dm.Appliance):
|
||||
pass
|
||||
|
||||
class Feature1(dm.Feature):
|
||||
_BoundAppliance = Appliance1
|
||||
VarStrInner: str = "testvalue FEATURE1"
|
||||
|
||||
app = Appliance1(feat1=Feature1)
|
||||
self.assertEqual(app.feat1.VarStrInner, "testvalue FEATURE1")
|
||||
|
||||
# check it does not leak accross instances
|
||||
|
||||
app = Appliance1(feat2=Feature1)
|
||||
self.assertEqual(app.feat2.VarStrInner, "testvalue FEATURE1")
|
||||
|
||||
with self.assertRaises(AttributeError):
|
||||
app.feat1
|
||||
|
||||
def test_register_notbound(self):
|
||||
"""Testing first appliance feature, and Field types (simple)"""
|
||||
|
||||
# class can be created
|
||||
class Appliance1(dm.Appliance):
|
||||
pass
|
||||
|
||||
class Feature1(dm.Feature):
|
||||
VarStrInner: str = "testvalue FEATURE1"
|
||||
|
||||
with self.assertRaises(dm.FeatureNotBound):
|
||||
Appliance1(feat1=Feature1)
|
||||
|
||||
def test_register_defect(self):
|
||||
|
||||
class Feature1(dm.Feature):
|
||||
pass
|
||||
|
||||
with self.assertRaises(dm.FeatureNotBound):
|
||||
Feature1()
|
||||
|
||||
def test_new_field_forbidden(self):
|
||||
class App(dm.Appliance):
|
||||
x: int = 1
|
||||
|
||||
app = App()
|
||||
with self.assertRaises(dm.NewFieldForbidden):
|
||||
app.y = 2
|
||||
|
||||
def test_inherit_declared(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
|
||||
class MyF1(App.F1):
|
||||
val = 2
|
||||
val2: str = "toto"
|
||||
|
||||
app = App(F1=MyF1)
|
||||
self.assertIsInstance(app.F1, MyF1)
|
||||
self.assertEqual(app.F1.val, 2)
|
||||
self.assertEqual(app.F1.val2, "toto")
|
||||
|
||||
def test_override_declared(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
val2: str = "toto"
|
||||
|
||||
app = App(F1={"val": 42, "val2": "tata"})
|
||||
self.assertEqual(app.F1.val, 42)
|
||||
self.assertEqual(app.F1.val2, "tata")
|
||||
|
||||
def test_dict_override_type_error(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
|
||||
# wrong type for val → must raise InvalidFieldValue
|
||||
with self.assertRaises(dm.InvalidFieldValue):
|
||||
App(F1={"val": "not-an-int"})
|
||||
|
||||
def test_dict_override_nonexisting_field(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
|
||||
# field does not exist → must raise
|
||||
with self.assertRaises(dm.InvalidFieldValue):
|
||||
App(F1={"doesnotexist": 123})
|
||||
|
||||
def test_inheritance_with_extra_fields(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
|
||||
class MyF1(App.F1):
|
||||
val = 2
|
||||
extra: str = "hello"
|
||||
|
||||
app = App(F1=MyF1)
|
||||
self.assertEqual(app.F1.val, 2)
|
||||
self.assertEqual(app.F1.extra, "hello")
|
||||
|
||||
def test_not_bound_runtime_attach_fails(self):
|
||||
class App(dm.Appliance):
|
||||
pass
|
||||
|
||||
class UnboundFeature(dm.Feature):
|
||||
x: int = 1
|
||||
|
||||
# attaching an unbound feature should raise
|
||||
with self.assertRaises(dm.FeatureNotBound):
|
||||
App(Unbound=UnboundFeature)
|
||||
|
||||
def test_override_does_not_leak_between_instances(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
|
||||
app1 = App(F1={"val": 99})
|
||||
app2 = App()
|
||||
self.assertEqual(app1.F1.val, 99)
|
||||
self.assertEqual(app2.F1.val, 1)
|
||||
|
||||
def test_deepfreeze_nested_mixed_tuple_list(self):
|
||||
class App(dm.Appliance):
|
||||
data: tuple[list[int], tuple[int, list[int]]] = ([1, 2], (3, [4, 5]))
|
||||
|
||||
app = App()
|
||||
|
||||
# Top-level: must be tuple
|
||||
self.assertIsInstance(app.data, tuple)
|
||||
|
||||
# First element of tuple: should have been frozen to tuple, not list
|
||||
self.assertIsInstance(app.data[0], tuple)
|
||||
|
||||
# Nested second element: itself a tuple
|
||||
self.assertIsInstance(app.data[1], tuple)
|
||||
|
||||
# Deepest element: inner list should also be frozen to tuple
|
||||
self.assertIsInstance(app.data[1][1], tuple)
|
||||
|
||||
# Check immutability
|
||||
with self.assertRaises(TypeError):
|
||||
app.data[0] += (99,) # tuples are immutable
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
app.data[1][1] += (42,) # inner tuple also immutable
|
||||
|
||||
def test_inacurate_type(self):
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance1(dm.Appliance):
|
||||
SomeVar: list = []
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance2(dm.Appliance):
|
||||
SomeVar: list[Any] = []
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance3(dm.Appliance):
|
||||
SomeVar: list[object] = []
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance4(dm.Appliance):
|
||||
SomeVar: dict = {}
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance5(dm.Appliance):
|
||||
SomeVar: dict[str, Any] = {}
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance6(dm.Appliance):
|
||||
SomeVar: dict[Any, Any] = {}
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance7(dm.Appliance):
|
||||
SomeVar: dict[Any, str] = {}
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldAnnotation):
|
||||
|
||||
class Appliance8(dm.Appliance):
|
||||
SomeVar: dict[str, object] = {}
|
||||
|
||||
def test_runtime_attach_bound_success(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
|
||||
class Extra(App.F1): # stays bound to App
|
||||
val = 7
|
||||
|
||||
app = App(Extra=Extra)
|
||||
self.assertTrue(hasattr(app, "Extra"))
|
||||
self.assertIsInstance(app.Extra, Extra)
|
||||
self.assertEqual(app.Extra.val, 7)
|
||||
|
||||
def test_cant_override_inherited_annotation(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
val: int = 1
|
||||
|
||||
with self.assertRaises(dm.ReadOnlyFieldAnnotation):
|
||||
|
||||
class Extra(App.F1):
|
||||
val: str = "test"
|
||||
|
||||
def test_fields_are_frozen_after_override(self):
|
||||
class App(dm.Appliance):
|
||||
class F(dm.Feature):
|
||||
nums: list[int] = [1, 2]
|
||||
tag: str = "x"
|
||||
|
||||
# dict override
|
||||
app1 = App(F={"nums": [9], "tag": "y"})
|
||||
self.assertEqual(app1.F.nums, (9,))
|
||||
self.assertEqual(app1.F.tag, "y")
|
||||
with self.assertRaises(AttributeError):
|
||||
app1.F.nums.append(3) # tuple
|
||||
|
||||
# subclass override
|
||||
class F2(App.F):
|
||||
nums = [4, 5]
|
||||
|
||||
app2 = App(F=F2)
|
||||
self.assertEqual(app2.F.nums, (4, 5))
|
||||
with self.assertRaises(dm.ReadOnlyField):
|
||||
app2.F.nums += (6,) # still immutable
|
||||
|
||||
def test_dict_partial_override_keeps_other_defaults(self):
|
||||
class App(dm.Appliance):
|
||||
class F(dm.Feature):
|
||||
a: int = 1
|
||||
b: str = "k"
|
||||
|
||||
app = App(F={"b": "z"})
|
||||
self.assertEqual(app.F.a, 1) # default remains
|
||||
self.assertEqual(app.F.b, "z") # overridden
|
||||
|
||||
def test_override_linear_chain(self):
|
||||
# Base appliance defines Feat1
|
||||
class A(dm.Appliance):
|
||||
class Feat1(dm.Feature):
|
||||
x: int = 1
|
||||
|
||||
# ✅ Appliance B overrides Feat1 by subclassing A.Feat1
|
||||
class B(A):
|
||||
class Feat1(A.Feat1):
|
||||
y: int = 2
|
||||
|
||||
self.assertTrue(issubclass(B.Feat1, A.Feat1))
|
||||
|
||||
# ✅ Appliance C overrides Feat1 again by subclassing B.Feat1 (not A.Feat1)
|
||||
class C(B):
|
||||
class Feat1(B.Feat1):
|
||||
z: int = 3
|
||||
|
||||
self.assertTrue(issubclass(C.Feat1, B.Feat1))
|
||||
self.assertTrue(issubclass(C.Feat1, A.Feat1))
|
||||
|
||||
# ❌ Bad: D tries to override with a *fresh* Feature, not subclass of B.Feat1
|
||||
with self.assertRaises(dm.InvalidFeatureInheritance):
|
||||
|
||||
class D(B):
|
||||
class Feat1(dm.Feature):
|
||||
fail: str = "oops"
|
||||
|
||||
# ❌ Bad: E tries to override with ancestor (A.Feat1) instead of B.Feat1
|
||||
with self.assertRaises(dm.InvalidFeatureInheritance):
|
||||
|
||||
class E(B):
|
||||
class Feat1(A.Feat1):
|
||||
fail: str = "oops"
|
||||
|
||||
# ✅ New feature name in child is always fine
|
||||
class F(B):
|
||||
class Feat2(dm.Feature):
|
||||
other: str = "ok"
|
||||
|
||||
self.assertTrue(hasattr(F, "Feat2"))
|
||||
|
||||
def test_override_chain_runtime_replacement(self):
|
||||
# Build a linear chain: A -> B -> C for feature 'Feat1'
|
||||
class A(dm.Appliance):
|
||||
class Feat1(dm.Feature):
|
||||
x: int = 1
|
||||
|
||||
class B(A):
|
||||
class Feat1(A.Feat1):
|
||||
y: int = 2
|
||||
|
||||
class C(B):
|
||||
class Feat1(B.Feat1):
|
||||
z: int = 3
|
||||
|
||||
# ✅ OK: at instantiation of C, replacing Feat1 with a subclass of the LATEST (C.Feat1)
|
||||
class CFeat1Plus(C.Feat1):
|
||||
w: int = 4
|
||||
|
||||
c_ok = C(Feat1=CFeat1Plus)
|
||||
self.assertIsInstance(c_ok.Feat1, CFeat1Plus)
|
||||
self.assertEqual(
|
||||
(c_ok.Feat1.x, c_ok.Feat1.y, c_ok.Feat1.z, c_ok.Feat1.w), (1, 2, 3, 4)
|
||||
)
|
||||
|
||||
# ❌ Not OK: replacing with a subclass of the ancestor (A.Feat1) — must target latest (C.Feat1)
|
||||
class AFeat1Alt(A.Feat1):
|
||||
pass
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldValue):
|
||||
C(Feat1=AFeat1Alt)
|
||||
|
||||
# ❌ Not OK: replacing with a subclass of the mid ancestor (B.Feat1) — still must target latest (C.Feat1)
|
||||
class BFeat1Alt(B.Feat1):
|
||||
pass
|
||||
|
||||
with self.assertRaises(dm.InvalidFieldValue):
|
||||
C(Feat1=BFeat1Alt)
|
||||
|
||||
def test_inheritance_tree_and_no_leakage(self):
|
||||
class A(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
a: int = 1
|
||||
|
||||
class F2(dm.Feature):
|
||||
b: int = 2
|
||||
|
||||
# ✅ Child inherits both features automatically
|
||||
class B(A):
|
||||
c: str = "extra"
|
||||
|
||||
b1 = B()
|
||||
self.assertIsInstance(b1.F1, A.F1)
|
||||
self.assertIsInstance(b1.F2, A.F2)
|
||||
self.assertEqual((b1.F1.a, b1.F2.b, b1.c), (1, 2, "extra"))
|
||||
|
||||
# ✅ Override only F2, F1 should still come from A
|
||||
class C(B):
|
||||
class F2(B.F2):
|
||||
bb: int = 22
|
||||
|
||||
c1 = C()
|
||||
self.assertIsInstance(c1.F1, A.F1) # unchanged
|
||||
self.assertIsInstance(c1.F2, C.F2) # overridden
|
||||
self.assertEqual((c1.F1.a, c1.F2.b, c1.F2.bb), (1, 2, 22))
|
||||
|
||||
# ✅ No leakage: instances of B are not affected by C's override
|
||||
b2 = B()
|
||||
self.assertIsInstance(b2.F2, A.F2)
|
||||
self.assertFalse(hasattr(b2.F2, "bb"))
|
||||
|
||||
# ✅ Adding a new feature in D is independent of previous appliances
|
||||
class D(C):
|
||||
class F3(dm.Feature):
|
||||
d: int = 3
|
||||
|
||||
d1 = D()
|
||||
self.assertIsInstance(d1.F1, A.F1)
|
||||
self.assertIsInstance(d1.F2, C.F2)
|
||||
self.assertIsInstance(d1.F3, D.F3)
|
||||
|
||||
# ✅ No leakage: instances of A and B should not see F3
|
||||
a1 = A()
|
||||
self.assertFalse(hasattr(a1, "F3"))
|
||||
b3 = B()
|
||||
self.assertFalse(hasattr(b3, "F3"))
|
||||
|
||||
def test_appliance_inheritance_tree_isolation(self):
|
||||
class A(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
a: int = 1
|
||||
|
||||
# Branch 1 overrides F1
|
||||
class B(A):
|
||||
class F1(A.F1):
|
||||
b: int = 2
|
||||
|
||||
# Branch 2 also overrides F1 differently
|
||||
class C(A):
|
||||
class F1(A.F1):
|
||||
c: int = 3
|
||||
|
||||
# ✅ Instances of B use B.F1
|
||||
b = B()
|
||||
self.assertIsInstance(b.F1, B.F1)
|
||||
self.assertEqual((b.F1.a, b.F1.b), (1, 2))
|
||||
self.assertFalse(hasattr(b.F1, "c"))
|
||||
|
||||
# ✅ Instances of C use C.F1
|
||||
c = C()
|
||||
self.assertIsInstance(c.F1, C.F1)
|
||||
self.assertEqual((c.F1.a, c.F1.c), (1, 3))
|
||||
self.assertFalse(hasattr(c.F1, "b"))
|
||||
|
||||
# ✅ Base appliance A still uses its original feature
|
||||
a = A()
|
||||
self.assertIsInstance(a.F1, A.F1)
|
||||
self.assertEqual(a.F1.a, 1)
|
||||
self.assertFalse(hasattr(a.F1, "b"))
|
||||
self.assertFalse(hasattr(a.F1, "c"))
|
||||
|
||||
# ✅ No leakage: B's override doesn't affect C and vice versa
|
||||
b2 = B()
|
||||
c2 = C()
|
||||
self.assertTrue(hasattr(b2.F1, "b"))
|
||||
self.assertFalse(hasattr(b2.F1, "c"))
|
||||
self.assertTrue(hasattr(c2.F1, "c"))
|
||||
self.assertFalse(hasattr(c2.F1, "b"))
|
||||
|
||||
def test_appliance_inheritance_tree_runtime_attach_isolation(self):
|
||||
class A(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
a: int = 1
|
||||
|
||||
class B(A):
|
||||
class F1(A.F1):
|
||||
b: int = 2
|
||||
|
||||
class C(A):
|
||||
class F1(A.F1):
|
||||
c: int = 3
|
||||
|
||||
# Define new runtime-attachable features
|
||||
class FextraB(B.F1):
|
||||
xb: int = 99
|
||||
|
||||
class FextraC(C.F1):
|
||||
xc: int = -99
|
||||
|
||||
# ✅ Attach to B at instantiation
|
||||
b = B(F1=FextraB)
|
||||
self.assertIsInstance(b.F1, FextraB)
|
||||
self.assertEqual((b.F1.a, b.F1.b, b.F1.xb), (1, 2, 99))
|
||||
self.assertFalse(hasattr(b.F1, "c"))
|
||||
self.assertFalse(hasattr(b.F1, "xc"))
|
||||
|
||||
# ✅ Attach to C at instantiation
|
||||
c = C(F1=FextraC)
|
||||
self.assertIsInstance(c.F1, FextraC)
|
||||
self.assertEqual((c.F1.a, c.F1.c, c.F1.xc), (1, 3, -99))
|
||||
self.assertFalse(hasattr(c.F1, "b"))
|
||||
self.assertFalse(hasattr(c.F1, "xb"))
|
||||
|
||||
# ✅ Base appliance still untouched
|
||||
a = A()
|
||||
self.assertIsInstance(a.F1, A.F1)
|
||||
self.assertEqual(a.F1.a, 1)
|
||||
self.assertFalse(hasattr(a.F1, "b"))
|
||||
self.assertFalse(hasattr(a.F1, "c"))
|
||||
self.assertFalse(hasattr(a.F1, "xb"))
|
||||
self.assertFalse(hasattr(a.F1, "xc"))
|
||||
|
||||
# ✅ Repeated instantiations stay isolated
|
||||
b2 = B()
|
||||
c2 = C()
|
||||
self.assertIsInstance(b2.F1, B.F1)
|
||||
self.assertIsInstance(c2.F1, C.F1)
|
||||
self.assertFalse(hasattr(b2.F1, "xb"))
|
||||
self.assertFalse(hasattr(c2.F1, "xc"))
|
||||
|
||||
def test_feature_dict_override_with_nested_containers(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
values: list[int] = [1, 2]
|
||||
|
||||
app = App(F1={"values": [5, 6]})
|
||||
self.assertEqual(app.F1.values, (5, 6)) # deepfreeze → tuple
|
||||
|
||||
# Invalid type in list should fail
|
||||
with self.assertRaises(dm.InvalidFieldValue):
|
||||
App(F1={"values": [1, "oops"]})
|
||||
|
||||
def test_dict_override_with_unknown_key(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
a: int = 1
|
||||
|
||||
# Dict override with unknown field 'zzz'
|
||||
with self.assertRaises(dm.InvalidFieldValue):
|
||||
App(F1={"zzz": 42})
|
||||
|
||||
def test_schema_isolation_across_multiple_overrides(self):
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
a: int = 1
|
||||
|
||||
class F1a(App.F1):
|
||||
a = 10
|
||||
|
||||
class F1b(App.F1):
|
||||
a = 20
|
||||
|
||||
app1 = App(F1=F1a)
|
||||
self.assertIsInstance(app1.F1, F1a)
|
||||
self.assertEqual(app1.F1.a, 10)
|
||||
|
||||
app2 = App(F1=F1b)
|
||||
self.assertIsInstance(app2.F1, F1b)
|
||||
self.assertEqual(app2.F1.a, 20)
|
||||
|
||||
# Original appliance schema must not be polluted
|
||||
app3 = App()
|
||||
self.assertIsInstance(app3.F1, App.F1)
|
||||
self.assertEqual(app3.F1.a, 1)
|
||||
|
||||
def test_inheritance_with_annotated_fields(self):
|
||||
|
||||
class App(dm.Appliance):
|
||||
class F1(dm.Feature):
|
||||
a: Annotated[int, dm.LAMFieldInfo(doc="field a")] = 1
|
||||
|
||||
# ✅ Subclass override must inherit from parent F1
|
||||
class F1Ex(App.F1):
|
||||
b: str = "ok"
|
||||
|
||||
app = App(F1=F1Ex)
|
||||
self.assertIsInstance(app.F1, F1Ex)
|
||||
self.assertEqual((app.F1.a, app.F1.b), (1, "ok"))
|
||||
|
||||
# ❌ Wrong: fresh Feature under same name
|
||||
with self.assertRaises(dm.InvalidFeatureInheritance):
|
||||
|
||||
class Bad(App):
|
||||
class F1(dm.Feature):
|
||||
fail: str = "oops"
|
||||
|
||||
|
||||
# ---------- main ----------
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,35 +0,0 @@
|
||||
# dabmodel (c) by chacha
|
||||
#
|
||||
# dabmodel is licensed under a
|
||||
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
|
||||
#
|
||||
# You should have received a copy of the license along with this
|
||||
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
|
||||
|
||||
import unittest
|
||||
from os import chdir
|
||||
from io import StringIO
|
||||
from contextlib import redirect_stdout,redirect_stderr
|
||||
from pathlib import Path
|
||||
|
||||
print(__name__)
|
||||
print(__package__)
|
||||
|
||||
from src import dabmodel
|
||||
|
||||
testdir_path = Path(__file__).parent.resolve()
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
class Testtest_module(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
def test_version(self):
|
||||
self.assertNotEqual(dabmodel.__version__,"?.?.?")
|
||||
|
||||
def test_test_module(self):
|
||||
|
||||
with redirect_stdout(StringIO()) as capted_stdout, redirect_stderr(StringIO()) as capted_stderr:
|
||||
self.assertEqual(dabmodel.test_function(41),42)
|
||||
self.assertEqual(len(capted_stderr.getvalue()),0)
|
||||
self.assertEqual(capted_stdout.getvalue().strip(),"Hello world !")
|
||||
Reference in New Issue
Block a user