(
path: pathlib.Path, candidates: Sequence[dict[str, str]]
)
| 367 | |
| 368 | |
| 369 | def load_results( |
| 370 | path: pathlib.Path, candidates: Sequence[dict[str, str]] |
| 371 | ) -> dict[str, dict[str, str]]: |
| 372 | metadata, rows = parse_tsv( |
| 373 | path, |
| 374 | marker="cbm-virustotal-results-v2", |
| 375 | fields=RESULT_FIELDS, |
| 376 | ) |
| 377 | expected_count = len(candidates) |
| 378 | required_metadata = { |
| 379 | "scan_objects": expected_count, |
| 380 | "associations": expected_count, |
| 381 | "min_engines_policy": MIN_ENGINES, |
| 382 | } |
| 383 | for key, expected in required_metadata.items(): |
| 384 | if metadata.get(key) != str(expected): |
| 385 | raise ContractError(f"VT results metadata has invalid {key}") |
| 386 | if set(metadata) != { |
| 387 | "scan_objects", |
| 388 | "associations", |
| 389 | "min_engines_policy", |
| 390 | "min_completed_engines", |
| 391 | "max_completed_engines", |
| 392 | }: |
| 393 | raise ContractError("VT results metadata fields are incomplete or unexpected") |
| 394 | if len(rows) != expected_count: |
| 395 | raise ContractError("VT results do not contain exactly one row per candidate") |
| 396 | |
| 397 | by_path: dict[str, dict[str, str]] = {} |
| 398 | analysis_ids: set[str] = set() |
| 399 | candidate_by_path = {row["scan_path"]: row for row in candidates} |
| 400 | for row in rows: |
| 401 | candidate = candidate_by_path.get(row["scan_path"]) |
| 402 | if candidate is None or row["scan_path"] in by_path: |
| 403 | raise ContractError(f"VT results contain a surplus or duplicate row: {row['scan_path']}") |
| 404 | validate_result_row(row, candidate) |
| 405 | if row["analysis_id"] in analysis_ids: |
| 406 | raise ContractError("VT results reuse one analysis id for multiple objects") |
| 407 | analysis_ids.add(row["analysis_id"]) |
| 408 | by_path[row["scan_path"]] = row |
| 409 | if set(by_path) != set(candidate_by_path): |
| 410 | raise ContractError("VT results are missing candidate objects") |
| 411 | |
| 412 | completed = [int(row["completed_engines"]) for row in rows] |
| 413 | if ( |
| 414 | not metadata["min_completed_engines"].isdigit() |
| 415 | or not metadata["max_completed_engines"].isdigit() |
| 416 | or int(metadata["min_completed_engines"]) != min(completed) |
| 417 | or int(metadata["max_completed_engines"]) != max(completed) |
| 418 | ): |
| 419 | raise ContractError("VT results engine-range metadata does not match its rows") |
| 420 | return by_path |
| 421 | |
| 422 | |
| 423 | def write_tsv( |
no test coverage detected