4608042044

Build # 4608042044

Build Type

push

github

Committed by HSN

Commit Message

chore: package update

Run Details

992 of 1167 relevant lines covered (85.0%)

0.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.74

/chk/modules/testcase/support/execute.py

import abc
import re
from collections.abc import Callable
from pathlib import Path
from typing import Any


from chk.infrastructure.contexts import app
from chk.infrastructure.exception import err_message
from chk.infrastructure.file_loader import FileContext, PathFrom
from chk.infrastructure.helper import dict_get
from chk.modules.testcase.constants import (
    TestcaseConfigNode as TstConf,
    ExecuteConfigNode as ExConf,
)
from chk.modules.version.support import DocumentMixin


class ExecuteMixin(DocumentMixin):
    """
    Mixin for Execute sub-spec
    """

    @abc.abstractmethod
    def get_file_context(self) -> FileContext:
        """Abstract method to get file context"""

    def execute_validated(self) -> dict:
        """Validate the schema against config"""

        try:
            execute_doc = self.execute_as_dict()
            if not isinstance(execute_doc, dict):
                raise TypeError("execute_validated: invalid execute spec")

            if (
                result_val := dict_get(execute_doc, f"{ExConf.ROOT}.{ExConf.RESULT}")
            ) is None:
                return {}

            if isinstance(result_val, list):
                for each_var in result_val:
                    if not isinstance(each_var, str):
                        raise TypeError(
                            "{'execute': {'result': 'list variable must be string'}}"
                        )
                    if not re.search(r"^({@[a-zA-Z]\w*}|_)$", each_var):
                        raise TypeError(
                            "{'execute': {'result': 'list variable name start with $'}}"
                        )
            else:
                raise TypeError("{'execute': {'result': 'must be list type'}}")

        except Exception as ex:
            raise RuntimeError(err_message("fatal.V0009", extra=ex)) from ex

        return execute_doc if isinstance(execute_doc, dict) else {}

    def execute_as_dict(
        self, with_key: bool = True, compiled: bool = False
    ) -> dict | None:
        """Get execute as dictionary"""

        execute_doc = self.as_dict(f"{TstConf.ROOT}.{ExConf.ROOT}", False, compiled)
        return {ExConf.ROOT: execute_doc} if with_key else execute_doc

    def execute_out_file(self, cb: Callable) -> Any:
        """Handle out file execution"""

        base_file = self.get_file_context().filepath

        execute_doc = self.execute_as_dict(with_key=False, compiled=True)
        if not isinstance(execute_doc, dict):
            raise TypeError("execute_out_file: invalid execute spec")

        file_name = dict_get(execute_doc, f"{ExConf.FILE}")
        file_name = PathFrom(Path(base_file)).absolute(file_name)

        execute_with_doc = (
            execute_doc[ExConf.WITH] if ExConf.WITH in execute_doc else {}
        )

        file_ctx: FileContext = FileContext.from_file(
            file_name,
            options=dict(self.get_file_context().options) | {"dump": False},
            arguments=dict(variables=execute_with_doc),
        )

        return cb(file_ctx)

    def execute_prepare_results(self, value_l: list[object]) -> None:
        """Make results out of values"""

        execute_doc = self.execute_as_dict(with_key=False, compiled=True)

        if not isinstance(execute_doc, dict):
            raise TypeError("execute_prepare_results: invalid execute spec")

        result_l = dict_get(execute_doc, f"{ExConf.RESULT}")

        if result_l is not None:
            if len(result_l) != len(value_l):
                raise ValueError("{'execute': {'result': 'value length do not match'}}")

            result_dict = {
                variable_name: value_l[index]
                for index, variable_name in enumerate(result_l)
                if variable_name != "_"
            }

        else:
            result_dict = {
                "_response": value_l,
            }

        app.set_local(self.get_file_context().filepath_hash, result_dict, ExConf.LOCAL)

1	import abc
2	import re	1✔
3	from collections.abc import Callable	1✔
4	from pathlib import Path	1✔
5	from typing import Any	1✔
6
7
8	from chk.infrastructure.contexts import app	1✔
9	from chk.infrastructure.exception import err_message	1✔
10	from chk.infrastructure.file_loader import FileContext, PathFrom	1✔
11	from chk.infrastructure.helper import dict_get	1✔
12	from chk.modules.testcase.constants import (	1✔
13	TestcaseConfigNode as TstConf,
14	ExecuteConfigNode as ExConf,
15	)
16	from chk.modules.version.support import DocumentMixin	1✔
17
18
19	class ExecuteMixin(DocumentMixin):	1✔
20	"""
21	Mixin for Execute sub-spec
22	"""
23
24	@abc.abstractmethod	1✔
25	def get_file_context(self) -> FileContext:	1✔
26	"""Abstract method to get file context"""
27
28	def execute_validated(self) -> dict:	1✔
29	"""Validate the schema against config"""
30
31	try:	1✔
32	execute_doc = self.execute_as_dict()	1✔
33	if not isinstance(execute_doc, dict):	1✔
34	raise TypeError("execute_validated: invalid execute spec")	×
35
36	if (	1✔
37	result_val := dict_get(execute_doc, f"{ExConf.ROOT}.{ExConf.RESULT}")
38	) is None:
39	return {}	1✔
40
41	if isinstance(result_val, list):	1✔
42	for each_var in result_val:	1✔
43	if not isinstance(each_var, str):	1✔
44	raise TypeError(	×
45	"{'execute': {'result': 'list variable must be string'}}"
46	)
47	if not re.search(r"^({@[a-zA-Z]\w*}\|_)$", each_var):	1✔
48	raise TypeError(	1✔
49	"{'execute': {'result': 'list variable name start with $'}}"
50	)
51	else:
52	raise TypeError("{'execute': {'result': 'must be list type'}}")	1✔
53
54	except Exception as ex:	1✔
55	raise RuntimeError(err_message("fatal.V0009", extra=ex)) from ex	1✔
56
57	return execute_doc if isinstance(execute_doc, dict) else {}	1✔
58
59	def execute_as_dict(	1✔
60	self, with_key: bool = True, compiled: bool = False
61	) -> dict \| None:
62	"""Get execute as dictionary"""
63
64	execute_doc = self.as_dict(f"{TstConf.ROOT}.{ExConf.ROOT}", False, compiled)	1✔
65	return {ExConf.ROOT: execute_doc} if with_key else execute_doc	1✔
66
67	def execute_out_file(self, cb: Callable) -> Any:	1✔
68	"""Handle out file execution"""
69
70	base_file = self.get_file_context().filepath	1✔
71
72	execute_doc = self.execute_as_dict(with_key=False, compiled=True)	1✔
73	if not isinstance(execute_doc, dict):	1✔
74	raise TypeError("execute_out_file: invalid execute spec")	×
75
76	file_name = dict_get(execute_doc, f"{ExConf.FILE}")	1✔
77	file_name = PathFrom(Path(base_file)).absolute(file_name)	1✔
78
79	execute_with_doc = (	1✔
80	execute_doc[ExConf.WITH] if ExConf.WITH in execute_doc else {}
81	)
82
83	file_ctx: FileContext = FileContext.from_file(	1✔
84	file_name,
85	options=dict(self.get_file_context().options) \| {"dump": False},
86	arguments=dict(variables=execute_with_doc),
87	)
88
89	return cb(file_ctx)	1✔
90
91	def execute_prepare_results(self, value_l: list[object]) -> None:	1✔
92	"""Make results out of values"""
93
94	execute_doc = self.execute_as_dict(with_key=False, compiled=True)	1✔
95
96	if not isinstance(execute_doc, dict):	1✔
97	raise TypeError("execute_prepare_results: invalid execute spec")	×
98
99	result_l = dict_get(execute_doc, f"{ExConf.RESULT}")	1✔
100
101	if result_l is not None:	1✔
102	if len(result_l) != len(value_l):	1✔
103	raise ValueError("{'execute': {'result': 'value length do not match'}}")	×
104
105	result_dict = {	1✔
106	variable_name: value_l[index]
107	for index, variable_name in enumerate(result_l)
108	if variable_name != "_"
109	}
110
111	else:
112	result_dict = {	1✔
113	"_response": value_l,
114	}
115
116	app.set_local(self.get_file_context().filepath_hash, result_dict, ExConf.LOCAL)	1✔

chkware / cli / 4608042044

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous