From 27f1a763f0f06b6e4ddc826f54b1da56db6f59c1 Mon Sep 17 00:00:00 2001 From: Adam Souzis Date: Tue, 17 Dec 2024 06:47:12 -0800 Subject: [PATCH] dsl: Add "runtime_func" function decorator to enable a function to be invoked as an eval expression function. Add tests for last few commits of dsl functionality. --- tests/examples/constraints.py | 5 +- tests/examples/dsl_relationships.py | 12 +- tests/examples/type_errors.py | 25 ++++ tests/test_constraints.py | 38 +++-- tests/test_dsl.py | 1 + tests/test_dsl_integration.py | 220 +++++++++++++++++++++++----- tests/test_eval.py | 8 + unfurl/job.py | 3 +- unfurl/testing.py | 11 +- unfurl/tosca_plugins/expr.py | 59 ++++++-- 10 files changed, 313 insertions(+), 69 deletions(-) create mode 100644 tests/examples/type_errors.py diff --git a/tests/examples/constraints.py b/tests/examples/constraints.py index 3ba4855c..f3e80e84 100644 --- a/tests/examples/constraints.py +++ b/tests/examples/constraints.py @@ -1,6 +1,7 @@ import unfurl import tosca from tosca import gb +from typing_extensions import reveal_type class ContainerService(tosca.nodes.Root): image: str url: str @@ -40,8 +41,8 @@ def _class_init(cls) -> None: # the proxy's backend is set to the container's url cls.proxy.backend_url = cls.container.url cls.proxy.hosting.name = "app" - tosca.in_range(2*gb, 20*gb).apply_constraint(cls.proxy.hosting.mem_size) - # same as: + tosca.in_range(1*gb*2, 20*gb).apply_constraint(cls.proxy.hosting.mem_size) + # the following is equivalent, but has a static type error: # cls.proxy.hosting.mem_size = tosca.in_range(2*gb, 20*gb) topology = App("myapp", proxy=ProxyContainerHost()) diff --git a/tests/examples/dsl_relationships.py b/tests/examples/dsl_relationships.py index e75a5060..38cc9335 100644 --- a/tests/examples/dsl_relationships.py +++ b/tests/examples/dsl_relationships.py @@ -1,12 +1,22 @@ # TESTS: # _target # f-strings: __str__ => "{{ 'expr' | eval }}" +# unfurl.tosca_plugins imports in safe mode +# runtime_func in safe mode +import unfurl import tosca from typing import Optional +from unfurl.tosca_plugins.expr import runtime_func +from unfurl.tosca_plugins.functions import to_dns_label + +@runtime_func +def disk_label(label: str) -> str: + return to_dns_label(label + "_disk") class Volume(tosca.nodes.Root): disk_label: str + disk_size: tosca.Size = 100 * tosca.GB class VolumeAttachment(tosca.relationships.AttachesTo): _target: Volume @@ -26,5 +36,5 @@ def _class_init(cls): # XXX intent = when_expr(cls.volume_attachment, 'mount', 'skip') intent = 'mount' if cls.volume_attachment else 'skip', target = "HOST", - mountpoint = f"/mnt/{cls.volume_attachment._target.disk_label}", + mountpoint = f"/mnt/{disk_label(cls.volume_attachment._target.disk_label)}", ) diff --git a/tests/examples/type_errors.py b/tests/examples/type_errors.py new file mode 100644 index 00000000..792efddd --- /dev/null +++ b/tests/examples/type_errors.py @@ -0,0 +1,25 @@ +import unfurl +import tosca +from tosca import gb, Size +from typing_extensions import reveal_type +class ContainerService(tosca.nodes.Root): + image: str + url: str + mem_size: tosca.Size + name: str = tosca.CONSTRAINED + + @classmethod + def _class_init(cls) -> None: + # the proxy's backend is set to the container's url + g: Size = 2 * gb + bar: Size = g * 2.0 + baz: Size = g * 2 + wrong = g * "ddd" + change_unit: Size = g * tosca.MB + mismatch = g * 2.0 * tosca.kHz + mismatch = g * tosca.HZ + tosca.in_range(2*gb*2, 2*gb*2).apply_constraint(cls.mem_size) + tosca.in_range(2*gb*2, 2*gb*2).apply_constraint(cls.url) + baz + 50*tosca.GHz + tosca.MB.as_int(baz) + tosca.GHz.as_int(baz) diff --git a/tests/test_constraints.py b/tests/test_constraints.py index 3739c439..7f16f814 100644 --- a/tests/test_constraints.py +++ b/tests/test_constraints.py @@ -16,14 +16,7 @@ from tosca.python2yaml import PythonToYaml from click.testing import CliRunner from unfurl.util import change_cwd - - -def _verify_mypy(path): - stdout, stderr, return_code = api.run(["--disable-error-code=override", path]) - if stdout: - print(stdout) - assert "no issues found in 1 source file" in stdout - assert return_code == 0, (stderr, stdout) +from unfurl.testing import assert_no_mypy_errors def test_constraints(): @@ -190,7 +183,22 @@ def test_constraints(): def test_mypy(path): # assert mypy ok basepath = os.path.join(os.path.dirname(__file__), "examples", path) - _verify_mypy(basepath) + assert_no_mypy_errors(basepath, "--disable-error-code=override") + + +@unittest.skipIf("slow" in os.getenv("UNFURL_TEST_SKIP", ""), "UNFURL_TEST_SKIP set") +def test_mypy_errors(): + # assert mypy expected errors + expected = [ + 'error: Unsupported operand types for * ("Size" and "str") [operator]', + 'Unsupported operand types for * ("Size" and "_Unit[Frequency]") [operator]', + 'Argument 1 to "apply_constraint" of "DataConstraint" has incompatible type "str"; expected "Size" [arg-type]', + 'Unsupported operand types for + ("Size" and "Frequency")', + 'Argument 1 to "as_int" of "_Unit" has incompatible type "Size"; expected "Frequency"', + "Found 6 errors in 1 file", + ] + basepath = os.path.join(os.path.dirname(__file__), "examples", "type_errors.py") + assert_no_mypy_errors(basepath, expected=expected) constraints_yaml = """ @@ -301,7 +309,6 @@ def test_computed_properties(): "source": 80, "source_range": None, }, - "a_list": [1], "data_list": [ { @@ -312,7 +319,7 @@ def test_computed_properties(): "source": 80, "source_range": None, }, - "additional": 1 + "additional": 1, } ], "extra": "extra", @@ -353,7 +360,10 @@ def test_computed_properties(): "node_types": { "Volume": { "derived_from": "tosca.nodes.Root", - "properties": {"disk_label": {"type": "string"}}, + "properties": { + "disk_label": {"type": "string"}, + "disk_size": {"type": "scalar-unit.size", "default": "100 GB"}, + }, }, "TestTarget": { "derived_from": "tosca.nodes.Root", @@ -361,7 +371,7 @@ def test_computed_properties(): "volume_mount": { "type": "VolumeMountArtifact", "properties": { - "mountpoint": "/mnt/{{ '.targets::volume_attachment::.target::disk_label' | eval }}" + "mountpoint": "/mnt/{{ {'eval': {'computed': ['service_template.dsl_relationships:disk_label', {'eval': '.targets::volume_attachment::.target::disk_label'}]}} | map_value }}", }, "file": "", "intent": "mount", @@ -396,7 +406,7 @@ def test_computed_properties(): def test_relationships(): basepath = os.path.join(os.path.dirname(__file__), "examples/") # loads yaml with with a json include - local = LocalEnv(basepath + "dsl-ensemble.yaml") + local = LocalEnv(basepath + "dsl-ensemble.yaml") # loads dsl_relationships.py manifest = local.get_manifest(skip_validation=True, safe_mode=True) service_template = manifest.manifest.expanded["spec"]["service_template"] # pprint.pprint(service_template, indent=2) diff --git a/tests/test_dsl.py b/tests/test_dsl.py index 64b25d56..fae1a767 100644 --- a/tests/test_dsl.py +++ b/tests/test_dsl.py @@ -425,6 +425,7 @@ def db_server_configure(**kw: Any) -> Any: ''' + def test_example_helloworld(): src, src_tpl = _to_python(example_helloworld_yaml) tosca_tpl = _to_yaml(src, True) diff --git a/tests/test_dsl_integration.py b/tests/test_dsl_integration.py index 0665fdf4..1a34138a 100644 --- a/tests/test_dsl_integration.py +++ b/tests/test_dsl_integration.py @@ -2,11 +2,14 @@ import pytest import unfurl import tosca +from tosca import Size, MB +from unfurl.eval import Ref from unfurl.logs import is_sensitive -from unfurl.testing import runtime_test +from unfurl.testing import runtime_test, create_runner from unfurl.tosca_plugins import expr, functions from typing import Optional, Type from unfurl.util import UnfurlError +import tosca._tosca class Service(tosca.nodes.Root): @@ -22,7 +25,9 @@ class Service(tosca.nodes.Root): parent: "Service" = tosca.find_required_by("connects_to") - connects_to: Optional["Service"] = tosca.Requirement(default=None, relationship=unfurl.relationships.Configures) + connects_to: Optional["Service"] = tosca.Requirement( + default=None, relationship=unfurl.relationships.Configures + ) def test_runtime_test(): @@ -93,17 +98,20 @@ class test(tosca.Namespace): service = Service(connects_to=connection) assert test.connection._name == "test.connection" - assert test.connection.find_required_by(requirement, expected_type) == tosca.Eval( - {"eval": "::test.connection::.sources::connects_to"} - ) - assert test.connection.find_all_required_by(requirement, expected_type) == tosca.Eval( - {"eval": "::test.connection::.sources::connects_to", "foreach": "$true"} - ) + assert test.connection.find_required_by(requirement, expected_type) == tosca.Eval({ + "eval": "::test.connection::.sources::connects_to" + }) + assert test.connection.find_all_required_by( + requirement, expected_type + ) == tosca.Eval({ + "eval": "::test.connection::.sources::connects_to", + "foreach": "$true", + }) if expected_type: - assert test.connection.find_required_by(requirement, expected_type).url == tosca.Eval( - {"eval": "::test.connection::.sources::connects_to::url"} - ) + assert test.connection.find_required_by( + requirement, expected_type + ).url == tosca.Eval({"eval": "::test.connection::.sources::connects_to::url"}) with pytest.raises(TypeError): test.connection.find_required_by(Service.connects_to, tosca.nodes.Compute) @@ -111,9 +119,9 @@ class test(tosca.Namespace): with pytest.raises(TypeError): test.connection.find_all_required_by(Service.connects_to, tosca.nodes.Compute) - assert test.connection.find_configured_by(Service.url) == tosca.Eval( - {"eval": "::test.connection::.configured_by::url"} - ) + assert test.connection.find_configured_by(Service.url) == tosca.Eval({ + "eval": "::test.connection::.configured_by::url" + }) tosca.global_state.mode = "yaml" assert test.service.connects_to == test.connection @@ -123,15 +131,14 @@ class test(tosca.Namespace): assert topology.service.connects_to == topology.connection assert topology.connection.connects_to == None assert ( - topology.connection.find_required_by(requirement, expected_type) == topology.service - ) - assert ( - topology.connection.find_all_required_by(requirement, expected_type) == [topology.service] + topology.connection.find_required_by(requirement, expected_type) + == topology.service ) + assert topology.connection.find_all_required_by(requirement, expected_type) == [ + topology.service + ] - assert ( - topology.service.find_all_required_by(requirement, expected_type) == [] - ) + assert topology.service.find_all_required_by(requirement, expected_type) == [] assert topology.connection.parent == topology.service # print ( topology.connection._instance.template.sources ) @@ -140,13 +147,17 @@ class test(tosca.Namespace): topology.connection.find_required_by(Service.connects_to, tosca.nodes.Compute) with pytest.raises(TypeError): - topology.connection.find_all_required_by(Service.connects_to, tosca.nodes.Compute) - - topology.service.host = 'example.com' + topology.connection.find_all_required_by( + Service.connects_to, tosca.nodes.Compute + ) + + topology.service.host = "example.com" assert topology.connection.find_configured_by(Service.url) == "https://example.com" + def test_hosted_on(): tosca.global_state.mode = "spec" + class test(tosca.Namespace): server = tosca.nodes.Compute( os=tosca.capabilities.OperatingSystem( @@ -156,23 +167,36 @@ class test(tosca.Namespace): ) software = tosca.nodes.SoftwareComponent(host=[server]) - setattr(software, "architecture", tosca.find_hosted_on(tosca.nodes.Compute.os).architecture) + setattr( + software, + "architecture", + tosca.find_hosted_on(tosca.nodes.Compute.os).architecture, + ) - assert test.software.find_hosted_on(tosca.nodes.Compute.os).architecture == tosca.Eval( - {"eval": "::test.software::.hosted_on::.capabilities::[.name=os]::architecture"} - ) - assert test.software.architecture == {'eval': '.hosted_on::.capabilities::[.name=os]::architecture' } + assert test.software.find_hosted_on( + tosca.nodes.Compute.os + ).architecture == tosca.Eval({ + "eval": "::test.software::.hosted_on::.capabilities::[.name=os]::architecture" + }) + assert test.software.architecture == { + "eval": ".hosted_on::.capabilities::[.name=os]::architecture" + } topology = runtime_test(test) - assert topology.software.find_hosted_on(tosca.nodes.Compute.os).architecture == "x86_64" + assert ( + topology.software.find_hosted_on(tosca.nodes.Compute.os).architecture + == "x86_64" + ) assert topology.software.architecture == "x86_64" def validate_pw(password: str) -> bool: return len(password) > 4 + def test_expressions(): tosca.global_state.mode = "spec" + class Inputs(tosca.TopologyInputs): domain: str @@ -182,13 +206,15 @@ class Test(tosca.nodes.Root): default_expr: str = expr.fallback(None, "foo") or_expr: str = expr.or_expr(default_expr, "ignored") label: str = functions.to_dns_label(expr.get_input("missing", "fo!o")) - password: str = tosca.Property(options=expr.sensitive | expr.validate(validate_pw), default="default") + password: str = tosca.Property( + options=expr.sensitive | expr.validate(validate_pw), default="default" + ) class test(tosca.Namespace): service = Service() test_node = Test() - assert Inputs.domain == tosca.EvalData({'get_input': 'domain'}) + assert Inputs.domain == tosca.EvalData({"get_input": "domain"}) topology = runtime_test(test) assert expr.get_env("MISSING", "default") == "default" @@ -200,22 +226,142 @@ class test(tosca.Namespace): input: str = expr.get_input("MISSING") assert expr.get_dir(topology.service, "src").get() == os.path.dirname(__file__) # XXX assert topology.test_node.path1 == os.path.dirname(__file__) - assert expr.abspath(topology.service, "test_dsl_integration.py", "src").get() == __file__ + assert ( + expr.abspath(topology.service, "test_dsl_integration.py", "src").get() + == __file__ + ) assert expr.uri(None) != topology.test_node.url assert expr.uri(topology.test_node) == topology.test_node.url - assert functions.to_label("fo!oo", replace='_') == 'fo_oo' - assert expr.template(topology.test_node, contents="{%if 1 %}{{SELF.url}}{%endif%}") == "#::test.test_node" + assert functions.to_label("fo!oo", replace="_") == "fo_oo" + assert ( + expr.template(topology.test_node, contents="{%if 1 %}{{SELF.url}}{%endif%}") + == "#::test.test_node" + ) assert topology.test_node.default_expr == "foo" assert topology.test_node.or_expr == "foo" assert topology.test_node.label == "fo--o" assert is_sensitive(topology.test_node.password) - with pytest.raises(UnfurlError, match=r'validation failed for'): + with pytest.raises(UnfurlError, match=r"validation failed for"): topology.test_node.password = "" # XXX test: # "if_expr", and_expr # "lookup", # "to_env", # "get_ensemble_metadata", - # "negate", + # "not_", # "as_bool", # tempfile + + +from tosca import MB, GB, mb, gb, Size +import math + + +@expr.runtime_func +def calc_size(size1: Size, size2: Size) -> Size: + if size1 is None or size2 is None: + return None + # print("calc_size", size1, size2, max(size1, size2).ceil(GB)) + return GB.as_int(max(size1, size2)) * GB + + +@pytest.mark.parametrize( + "safe_mode", [True, False] +) +def test_units(safe_mode): + tosca.global_state.safe_mode = safe_mode + tosca.global_state.mode = "spec" + + g = 2 * gb + foo: float = float(20 * g) + bar: Size = g * 2.0 + bar: Size = g * 2 + baz: Size = 20.0 * g + one_mb = 1 * mb + assert abs(-one_mb) == one_mb + assert abs(one_mb) == +one_mb + assert hash(one_mb) + assert bool(0 * MB) == bool(0.0) + assert one_mb.value == 1000000.0 + assert str(one_mb) == "1 MB" + assert repr(one_mb) == "1.0*MB" + assert one_mb.as_unit == 1.0 + assert one_mb.to_yaml() == "1 MB" + assert str(one_mb * GB) == "0.001 GB" + with pytest.raises(TypeError, match="Hz"): + str(one_mb * tosca.HZ) + + mem_size: Size = 4000 * MB + 10 * GB + assert mem_size == 14 * GB + assert mem_size == 14000000000 + + + class Topology(tosca.Namespace): + host = tosca.capabilities.Compute( + num_cpus=1, + disk_size=10 * GB, + mem_size=mem_size, + ) + + class Test(tosca.nodes.Root): + mem_size: Size = tosca.Attribute() + host: tosca.capabilities.Compute + + test = Test(host=host) + + assert host.mem_size + assert host.mem_size == 14 * GB + compute = tosca.capabilities.Compute( + num_cpus=1, + disk_size=host.mem_size * 2, + mem_size=test.mem_size + test.mem_size, + ) + + assert compute.disk_size + assert compute.mem_size + assert isinstance(compute.disk_size * 2, Size) + # compute.mem_size depends on Test.mem_size which is a tosca attribute so it needs to be EvalData + assert isinstance(compute.mem_size, tosca.EvalData) + assert isinstance(compute.mem_size * 2, tosca.EvalData) + with pytest.raises(AttributeError): + assert Test.mem_size.as_unit + with pytest.raises(AttributeError): + assert compute.mem_size.as_unit + type_pun: Size = calc_size(compute.disk_size, compute.mem_size) + 4 * GB + assert isinstance(type_pun, tosca.EvalData) + if not tosca.global_state.safe_mode: + assert calc_size(compute.disk_size, compute.disk_size) == 28 * GB + test2 = Test( + host=tosca.capabilities.Compute( + mem_size=calc_size(compute.disk_size, compute.mem_size) + ) + ) + test3 = Test(host=compute) + + topology, runner = create_runner(Topology) + # make sure we can serialize this + str_io = runner.manifest.manifest.save() + # print(str_io.getvalue()) + assert ( + """ + eval: + computed: + - tests.test_dsl_integration:calc_size + - eval: + scalar: 28000 MB + - eval: ::Topology.test3::.capabilities[.name=host]::mem_size + """.replace(" ", "") + in str_io.getvalue().replace(" ", "") + ) + calcd = runner.manifest.rootResource.query( + "::Topology.test::.capabilities::[.name=host]::mem_size", trace=0 + ) + assert calcd == "14000 MB" + topology.test.mem_size = 2 * MB # set attribute value so expression resolves + assert topology.test.mem_size == 2 * MB + assert topology.test3.host.mem_size == 4 * MB + tosca.global_state.safe_mode = False + result = Ref(topology.type_pun.expr).resolve_one( + tosca.global_state.context.copy(trace=0) + ) + assert result == 32 * GB diff --git a/tests/test_eval.py b/tests/test_eval.py index ffedd17c..85372ef4 100644 --- a/tests/test_eval.py +++ b/tests/test_eval.py @@ -183,6 +183,8 @@ def test_funcs(self): assert 3 == Ref(test6).resolve_one(RefContext(resource)) test7 = {"eval": {"sub": [1, 1]}} assert 0 == Ref(test7).resolve_one(RefContext(resource)) + test8 = {"eval": dict(scalar_value="6 mb", unit="gb", round=2)} + assert 0.01 == Ref(test8).resolve_one(RefContext(resource)) def test_circular_refs(self): more = {} @@ -376,6 +378,12 @@ def test_jinjaTemplate(self): ) assert val == "a.b" + val = map_value( + "{{ __unfurl.scalar_value('500 kb' | scalar + '1 mb' | scalar, 'gb') }}", + ctx + ) + assert val == 0.0015 + def test_templateFunc(self): query = { "eval": {"template": "{%if testVar %}{{success}}{%else%}failed{%endif%}"}, diff --git a/unfurl/job.py b/unfurl/job.py index c18ea5fd..b40db21d 100644 --- a/unfurl/job.py +++ b/unfurl/job.py @@ -1666,7 +1666,7 @@ def run_job( class Runner: "this class is only used by unit tests, application uses start_job() above" - def __init__(self, manifest): + def __init__(self, manifest: "YamlManifest"): self.manifest = manifest assert self.manifest.tosca self.job = None @@ -1682,6 +1682,7 @@ def run(self, jobOptions=None): jobOptions = JobOptions() if not self.job: self.job = _plan(self.manifest, jobOptions) + assert self.job rendered, count = _render(self.job) if not jobOptions.planOnly: self.job.run(rendered) diff --git a/unfurl/testing.py b/unfurl/testing.py index 4374f920..6849582a 100644 --- a/unfurl/testing.py +++ b/unfurl/testing.py @@ -26,16 +26,19 @@ import tosca from tosca.python2yaml import PythonToYaml from .dsl import proxy_instance +import pprint try: from mypy import api - def assert_no_mypy_errors(path, *args): + def assert_no_mypy_errors(path, *args, expected=["no issues found in 1 source file"]): stdout, stderr, return_code = api.run([path, *args] ) if stdout: print(stdout) - assert "no issues found in 1 source file" in stdout - assert return_code == 0, (stderr, stdout) + for msg in expected: + assert msg in stdout, f"not found in stdout: {msg}" + # if errors, return_code == 1 + assert return_code != 2, (stderr, stdout) except ImportError: assert_no_mypy_errors = None # type: ignore @@ -111,7 +114,7 @@ def _check_job(job, i, step): def lifecycle( - manifest: Manifest, steps: Iterable[Step] = DEFAULT_STEPS + manifest: YamlManifest, steps: Iterable[Step] = DEFAULT_STEPS ) -> Iterable[Job]: runner = Runner(manifest) for i, step in enumerate(steps, start=1): diff --git a/unfurl/tosca_plugins/expr.py b/unfurl/tosca_plugins/expr.py index b3d035f9..05330b40 100644 --- a/unfurl/tosca_plugins/expr.py +++ b/unfurl/tosca_plugins/expr.py @@ -1,7 +1,7 @@ """ Type-safe equivalents to Unfurl's Eval `Expression Functions`. -When called in "spec" mode (e.g. as part of a class definition or in ``_class_init_``) they will return eval expression +When called in `"spec" mode ` (e.g. as part of a class definition or in ``_class_init_``) they will return eval expression that will get executed. But note that the type signature will match the result of the expression, not the eval expression itself. (This type punning enables effective static type checking). @@ -15,6 +15,8 @@ The former variant can only be used in runtime mode as live objects are not available outside that mode. In "spec" mode, the None variant must be used and at runtime the eval expression returned by that function will be evaluated using the current context's instance. + +User-defined functions can be made available as an expression functions by the `runtime_func` decorator. """ # put this module is in unfurl/tosca_plugins because modules in this package are whitelisted as safe @@ -108,7 +110,7 @@ def get_context(obj: ToscaType, kw: Optional[Dict[str, Any]] = None) -> RefConte "get_dir", "template", "get_nodes_of_type", - "negate", + "not_", "as_bool", "uri", "concat", @@ -119,8 +121,49 @@ def get_context(obj: ToscaType, kw: Optional[Dict[str, Any]] = None) -> RefConte # XXX kubernetes_current_namespace # XXX kubectl, # XXX get_artifact + # XXX python + "runtime_func", ] +F = TypeVar("F", bound=Callable[..., Any]) + + +@overload +def runtime_func(_func: None = None) -> Callable[[F], F]: ... + + +@overload +def runtime_func(_func: F) -> F: ... + + +def runtime_func(_func: Optional[F] = None) -> Union[F, Callable[[F], F]]: + """ + A decorator for making a function invocable as a runtime expression. + When the decorated function invoked in `"spec" mode `, if any of its arguments contain :py:class:`tosca.EvalData`, + then the function will return :py:class:`tosca.EvalData` containing a JSON representation of the invocation as an `expression function ` that will be evaluated at runtime. + Otherwise, the function will eagerly execute as a normal Python function. + """ + + def _make_computed(func: F) -> F: + def wrapped(*args, **kwargs): + if global_state_mode() == "runtime": + ctx = global_state_context() + return func(*map_value(args, ctx), **map_value(kwargs, ctx)) + elif ( + not safe_mode() and not has_function(args) and not has_function(kwargs) + ): + return func(*args, **kwargs) + else: + kwargs["computed"] = [f"{func.__module__}:{func.__qualname__}", *args] + return EvalData({"eval": kwargs}) + + return cast(F, wrapped) + + if _func is None: + return _make_computed + else: + return _make_computed(_func) + def get_nodes_of_type(cls: Type[ToscaType]) -> list: if global_state_mode() == "runtime": @@ -134,12 +177,10 @@ def get_nodes_of_type(cls: Type[ToscaType]) -> list: return EvalData({"get_nodes_of_type": cls.tosca_type_name()}) # type: ignore -def negate(val) -> bool: +def not_(val) -> bool: if global_state_mode() == "runtime": return not bool(val) else: - if isinstance(val, EvalData): - val = val.expr return cast(bool, EvalData(dict(eval={"not": val, "map_value": 1}))) @@ -147,8 +188,6 @@ def as_bool(val) -> bool: if global_state_mode() == "runtime": return bool(val) else: - if isinstance(val, EvalData): - val = val.expr return cast(bool, EvalData(dict(eval={"not": {"not": val, "map_value": 1}}))) @@ -240,9 +279,9 @@ def get_env(name=None, default=None, *, ctx=None): def if_expr(if_cond, then: T, otherwise: U = None) -> Union[T, U]: """Returns an eval expression like: - {"eval": {"if": if_cond, "then": then, "else": otherwise} + ``{"eval": {"if": if_cond, "then": then, "else": otherwise}`` - This will not evaluate at runtime mode because all arguments will evaluated + This will not evaluate at `runtime mode ` because all arguments will evaluated before calling this function, defeating eval expressions' (and Python's) short-circuit semantics. To avoid unexpected behavior, an error will be raised if invoked during runtime mode. Instead just use a Python 'if' statement or expression. @@ -269,7 +308,7 @@ def or_expr(left: T, right: U) -> Union[T, U]: def fallback(left: Optional[T], right: T) -> T: if global_state_mode() == "runtime": raise UnfurlError( - "'fallback()' can not be valuate in runtime mode, instead just use Python's 'or' operator." + "'fallback()' can not be valuate in `runtime mode `, instead just use Python's 'or' operator." ) else: return EvalData(dict(eval={"or": [left, right], "map_value": 1})) # type: ignore