18854854802

Committed 27 Oct 2025 08:23PM UTC coverage: 80.311% (+0.2%) from 80.141%

Build # 18854854802

Build Type

Pull #112

github

Committed by

web-flow

Commit Message

Merge dd00b2d3e into 3ef0e3272

Pull Request Pull Request #112: fix: multipass entries

Run Details

213 of 293 branches covered (72.7%)

Branch coverage included in aggregate %.

42 of 55 new or added lines in 15 files covered. (76.36%)

2 existing lines in 2 files now uncovered.

2418 of 2983 relevant lines covered (81.06%)

0.81 hits per line

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

96.45

/src/socm/core/models.py

from collections.abc import Iterable
from numbers import Number
from typing import Dict, List, Optional, Union, get_args, get_origin

from pydantic import BaseModel, Field
from radical.pilot import TaskDescription


class QosPolicy(BaseModel):
    name: str
    max_walltime: Optional[int] = None  # in minutes
    max_jobs: Optional[int] = None
    max_cores: Optional[int] = None


class Resource(BaseModel):
    name: str
    nodes: int
    cores_per_node: int
    memory_per_node: int
    qos: List[QosPolicy] = Field(default_factory=list)


class Workflow(BaseModel):
    name: str
    executable: str
    context: str
    subcommand: str = ""
    id: Optional[int] = None
    environment: Optional[Dict[str, str]] = None
    resources: Optional[Dict[str, int | float]] = None

    model_config = {
        "extra": "allow",
    }

    def get_command(self, **kargs) -> str:
        raise NotImplementedError("This method should be implemented in subclasses")

    def get_arguments(self, **kargs) -> str:
        raise NotImplementedError("This method should be implemented in subclasses")

    def get_numeric_fields(self, avoid_attributes: List[str] | None = None) -> List[str]:
        """
        Returns a list of field names that are either numeric types
        or iterable collections of numeric types.

        Uses Pydantic v2 model_fields for type introspection.

        Returns:
            List[str]: Field names with numeric values
        """
        if avoid_attributes is None:
            avoid_attributes = []

        numeric_fields = []

        # Get field information from Pydantic v2 model_fields
        for field_name, field_info in self.__class__.model_fields.items():
            # Get the annotation type
            if field_name in avoid_attributes or getattr(self, field_name) is None:
                continue
            field_type = field_info.annotation

            # Check for direct numeric types
            if isinstance(field_type, type) and issubclass(field_type, Number):
                numeric_fields.append(field_name)
                continue

            # Check for complex types (Optional, List, etc)
            origin = get_origin(field_type)
            if origin is not None:
                args = get_args(field_type)

                # Check for Optional numeric types
                if origin is Union:
                    for arg in args:
                        if isinstance(arg, type) and issubclass(arg, Number):
                            numeric_fields.append(field_name)
                            break
                # Check for iterables of numbers
                elif issubclass(origin, Iterable):
                    # Check if it's a parameterized generic like List[int]
                    if args and len(args) > 0:
                        element_type = args[0]
                        if isinstance(element_type, type) and issubclass(element_type, Number):
                            numeric_fields.append(field_name)

        # Also check actual instance values for numeric fields not captured by annotations
        for field_name, value in self.__dict__.items():
            if field_name not in numeric_fields and field_name not in avoid_attributes:
                if isinstance(value, Number):
                    numeric_fields.append(field_name)
                elif isinstance(value, Iterable) and not isinstance(value, (str, bytes, dict)):
                    # Check if all elements are numbers
                    try:
                        if all(isinstance(item, Number) for item in value):
                            numeric_fields.append(field_name)
                    except (TypeError, ValueError):
                        pass

        return numeric_fields

    def get_categorical_fields(self, avoid_attributes: List[str] | None = None) -> List[str]:
        """
        Returns a list of field names that are either string types
        or iterable collections of string types.

        Uses Pydantic v2 model_fields for type introspection.

        Returns:
            List[str]: Field names with categorical (string) values
        """
        if avoid_attributes is None:
            avoid_attributes = []
        categorical_fields = []

        # Get field information from Pydantic v2 model_fields
        for field_name, field_info in self.__class__.model_fields.items():
            # Get the annotation type
            if field_name in avoid_attributes or getattr(self, field_name) is None:
                continue
            field_type = field_info.annotation

            # Check for direct numeric types
            if isinstance(field_type, type) and issubclass(field_type, str):
                categorical_fields.append(field_name)
                continue

            # Check for complex types (Optional, List, etc)
            origin = get_origin(field_type)
            if origin is not None:
                args = get_args(field_type)

                # Check for Optional numeric types
                if origin is Union:
                    for arg in args:
                        if isinstance(arg, type) and issubclass(arg, str):
                            categorical_fields.append(field_name)
                            break
                # Check for iterables of numbers
                elif issubclass(origin, Iterable):
                    # Check if it's a parameterized generic like List[int]
                    if args and len(args) > 0:
                        element_type = args[0]
                        if isinstance(element_type, type) and issubclass(element_type, str):
                            categorical_fields.append(field_name)

        # Also check actual instance values for numeric fields not captured by annotations
        for field_name, value in self.__dict__.items():
            if field_name not in categorical_fields and field_name not in avoid_attributes:
                if isinstance(value, str):
                    categorical_fields.append(field_name)
                elif isinstance(value, Iterable) and not isinstance(value, (Number, bytes, dict)):
                    # Check if all elements are numbers
                    try:
                        if all(isinstance(item, str) for item in value):
                            categorical_fields.append(field_name)
                    except (TypeError, ValueError):
                        pass

        return categorical_fields

    def get_tasks(self) -> List[TaskDescription]:
        """
        Returns a list of TaskDescription objects for the workflow.
        This is a placeholder method and should be implemented in subclasses.
        """
        raise NotImplementedError("This method should be implemented in subclasses")


class Campaign(BaseModel):
    id: int
    workflows: List[Workflow]
    deadline: str
    target_resource: str = "tiger3"
    campaign_policy: str = "time"
    execution_schema: str = "batch"
    requested_resources: int = 0

1	from collections.abc import Iterable	1✔
2	from numbers import Number	1✔
3	from typing import Dict, List, Optional, Union, get_args, get_origin	1✔
4
5	from pydantic import BaseModel, Field	1✔
6	from radical.pilot import TaskDescription	1✔
7
8
9	class QosPolicy(BaseModel):	1✔
10	name: str	1✔
11	max_walltime: Optional[int] = None # in minutes	1✔
12	max_jobs: Optional[int] = None	1✔
13	max_cores: Optional[int] = None	1✔
14
15
16	class Resource(BaseModel):	1✔
17	name: str	1✔
18	nodes: int	1✔
19	cores_per_node: int	1✔
20	memory_per_node: int	1✔
21	qos: List[QosPolicy] = Field(default_factory=list)	1✔
22
23
24	class Workflow(BaseModel):	1✔
25	name: str	1✔
26	executable: str	1✔
27	context: str	1✔
28	subcommand: str = ""	1✔
29	id: Optional[int] = None	1✔
30	environment: Optional[Dict[str, str]] = None	1✔
31	resources: Optional[Dict[str, int \| float]] = None	1✔
32
33	model_config = {	1✔
34	"extra": "allow",
35	}
36
37	def get_command(self, **kargs) -> str:	1✔
38	raise NotImplementedError("This method should be implemented in subclasses")	1✔
39
40	def get_arguments(self, **kargs) -> str:	1✔
41	raise NotImplementedError("This method should be implemented in subclasses")	1✔
42
43	def get_numeric_fields(self, avoid_attributes: List[str] \| None = None) -> List[str]:	1✔
44	"""
45	Returns a list of field names that are either numeric types
46	or iterable collections of numeric types.
47
48	Uses Pydantic v2 model_fields for type introspection.
49
50	Returns:
51	List[str]: Field names with numeric values
52	"""
53	if avoid_attributes is None:	1✔
54	avoid_attributes = []	1✔
55
56	numeric_fields = []	1✔
57
58	# Get field information from Pydantic v2 model_fields
59	for field_name, field_info in self.__class__.model_fields.items():	1✔
60	# Get the annotation type
61	if field_name in avoid_attributes or getattr(self, field_name) is None:	1✔
62	continue	1✔
63	field_type = field_info.annotation	1✔
64
65	# Check for direct numeric types
66	if isinstance(field_type, type) and issubclass(field_type, Number):	1✔
67	numeric_fields.append(field_name)	1✔
68	continue	1✔
69
70	# Check for complex types (Optional, List, etc)
71	origin = get_origin(field_type)	1✔
72	if origin is not None:	1✔
73	args = get_args(field_type)	1✔
74
75	# Check for Optional numeric types
76	if origin is Union:	1✔
77	for arg in args:	1✔
78	if isinstance(arg, type) and issubclass(arg, Number):	1✔
79	numeric_fields.append(field_name)	1✔
80	break	1✔
81	# Check for iterables of numbers
82	elif issubclass(origin, Iterable):	1✔
83	# Check if it's a parameterized generic like List[int]
84	if args and len(args) > 0:	1✔
85	element_type = args[0]	1✔
86	if isinstance(element_type, type) and issubclass(element_type, Number):	1✔
87	numeric_fields.append(field_name)	1✔
88
89	# Also check actual instance values for numeric fields not captured by annotations
90	for field_name, value in self.__dict__.items():	1✔
91	if field_name not in numeric_fields and field_name not in avoid_attributes:	1✔
92	if isinstance(value, Number):	1✔
93	numeric_fields.append(field_name)	1✔
94	elif isinstance(value, Iterable) and not isinstance(value, (str, bytes, dict)):	1✔
95	# Check if all elements are numbers
96	try:	1✔
97	if all(isinstance(item, Number) for item in value):	1✔
98	numeric_fields.append(field_name)	1✔
99	except (TypeError, ValueError):	×
100	pass	×
101
102	return numeric_fields	1✔
103
104	def get_categorical_fields(self, avoid_attributes: List[str] \| None = None) -> List[str]:	1✔
105	"""
106	Returns a list of field names that are either string types
107	or iterable collections of string types.
108
109	Uses Pydantic v2 model_fields for type introspection.
110
111	Returns:
112	List[str]: Field names with categorical (string) values
113	"""
114	if avoid_attributes is None:	1✔
115	avoid_attributes = []	1✔
116	categorical_fields = []	1✔
117
118	# Get field information from Pydantic v2 model_fields
119	for field_name, field_info in self.__class__.model_fields.items():	1✔
120	# Get the annotation type
121	if field_name in avoid_attributes or getattr(self, field_name) is None:	1✔
122	continue	1✔
123	field_type = field_info.annotation	1✔
124
125	# Check for direct numeric types
126	if isinstance(field_type, type) and issubclass(field_type, str):	1✔
127	categorical_fields.append(field_name)	1✔
128	continue	1✔
129
130	# Check for complex types (Optional, List, etc)
131	origin = get_origin(field_type)	1✔
132	if origin is not None:	1✔
133	args = get_args(field_type)	1✔
134
135	# Check for Optional numeric types
136	if origin is Union:	1✔
137	for arg in args:	1✔
138	if isinstance(arg, type) and issubclass(arg, str):	1✔
139	categorical_fields.append(field_name)	1✔
140	break	1✔
141	# Check for iterables of numbers
142	elif issubclass(origin, Iterable):	1✔
143	# Check if it's a parameterized generic like List[int]
144	if args and len(args) > 0:	1✔
145	element_type = args[0]	1✔
146	if isinstance(element_type, type) and issubclass(element_type, str):	1✔
147	categorical_fields.append(field_name)	1✔
148
149	# Also check actual instance values for numeric fields not captured by annotations
150	for field_name, value in self.__dict__.items():	1✔
151	if field_name not in categorical_fields and field_name not in avoid_attributes:	1✔
152	if isinstance(value, str):	1✔
153	categorical_fields.append(field_name)	1✔
154	elif isinstance(value, Iterable) and not isinstance(value, (Number, bytes, dict)):	1✔
155	# Check if all elements are numbers
156	try:	1✔
157	if all(isinstance(item, str) for item in value):	1✔
UNCOV 158	categorical_fields.append(field_name)	×
159	except (TypeError, ValueError):	×
160	pass	×
161
162	return categorical_fields	1✔
163
164	def get_tasks(self) -> List[TaskDescription]:	1✔
165	"""
166	Returns a list of TaskDescription objects for the workflow.
167	This is a placeholder method and should be implemented in subclasses.
168	"""
169	raise NotImplementedError("This method should be implemented in subclasses")	1✔
170
171
172	class Campaign(BaseModel):	1✔
173	id: int	1✔
174	workflows: List[Workflow]	1✔
175	deadline: str	1✔
176	target_resource: str = "tiger3"	1✔
177	campaign_policy: str = "time"	1✔
178	execution_schema: str = "batch"	1✔
179	requested_resources: int = 0	1✔

simonsobs / so_campaign_manager / 18854854802

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