Return all workloads from disk that match scope and platform.
(
container_tag: str,
n_repeat: int = 1,
time_limit: int = 1800,
tag: Optional[str] = None,
environment: Optional[str] = None,
test_cases: str = "all",
scope: Optional[str] = None,
model: Optional[str] = None,
test_case: Optional[str] = None,
container_image: Optional[str] = None,
)
| 185 | |
| 186 | |
| 187 | def load_workloads( |
| 188 | container_tag: str, |
| 189 | n_repeat: int = 1, |
| 190 | time_limit: int = 1800, |
| 191 | tag: Optional[str] = None, |
| 192 | environment: Optional[str] = None, |
| 193 | test_cases: str = "all", |
| 194 | scope: Optional[str] = None, |
| 195 | model: Optional[str] = None, |
| 196 | test_case: Optional[str] = None, |
| 197 | container_image: Optional[str] = None, |
| 198 | ) -> List[jetclient.JETWorkloadManifest]: |
| 199 | """Return all workloads from disk that match scope and platform.""" |
| 200 | recipes_dir = BASE_PATH / ".." / "recipes" |
| 201 | local_dir = BASE_PATH / ".." / "local_recipes" |
| 202 | |
| 203 | workloads: List[jetclient.JETWorkloadManifest] = [] |
| 204 | build_workloads: List[jetclient.JETClient] = [] |
| 205 | for file in list(recipes_dir.glob("*.yaml")) + list(local_dir.glob("*.yaml")): |
| 206 | workloads += load_and_flatten(config_path=str(file)) |
| 207 | if file.stem.startswith("_build"): |
| 208 | build_workloads.append(load_config(config_path=str(file))) |
| 209 | |
| 210 | if scope: |
| 211 | workloads = filter_by_scope(workload_manifests=workloads, scope=scope) |
| 212 | |
| 213 | if workloads and environment: |
| 214 | workloads = filter_by_environment(workload_manifests=workloads, environment=environment) |
| 215 | |
| 216 | if workloads and model: |
| 217 | workloads = filter_by_model(workload_manifests=workloads, model=model) |
| 218 | |
| 219 | if workloads and tag: |
| 220 | workloads = filter_by_tag(workload_manifests=workloads, tag=tag) |
| 221 | |
| 222 | if workloads and test_cases != "all": |
| 223 | workloads = filter_by_test_cases(workload_manifests=workloads, test_cases=test_cases) |
| 224 | |
| 225 | if workloads and test_case: |
| 226 | workloads = [filter_by_test_case(workload_manifests=workloads, test_case=test_case)] |
| 227 | |
| 228 | if not workloads: |
| 229 | return [] |
| 230 | |
| 231 | for workload in list(workloads): |
| 232 | for build_workload in build_workloads: |
| 233 | if ( |
| 234 | workload.spec.build == build_workload.spec.name |
| 235 | ) and build_workload not in workloads: |
| 236 | container_image = container_image or build_workload.spec.source.image |
| 237 | build_workload.spec.source.image = f"{container_image}:{container_tag}" |
| 238 | workloads.append(build_workload) |
| 239 | workload.spec.n_repeat = n_repeat |
| 240 | workload.spec.time_limit = time_limit |
| 241 | return workloads |
nothing calls this directly
no test coverage detected