clean up calendar display, added separated device runtime overrides

2026-07-13 02:43:52 -04:00
parent 6bf6fa4867
commit 42f75fecef
8 changed files with 317 additions and 110 deletions
@@ -44,7 +44,7 @@ def _smb_credentials_from_settings(settings: dict[str, Any]) -> dict[str, str]:
 	}


-def _runtime_overrides_from_settings(settings: dict[str, Any]) -> dict[str, int]:
+def _runtime_overrides_from_settings(settings: dict[str, Any]) -> dict[str, dict[str, int]]:
 	def _parse_positive_int(value: Any) -> int | None:
 		if value is None:
 			return None
@@ -58,15 +58,23 @@ def _runtime_overrides_from_settings(settings: dict[str, Any]) -> dict[str, int]
 			return None
 		return parsed if parsed > 0 else None

-	overrides: dict[str, int] = {}
-	for test_type, key in (
-		("P2P", "p2pRuntimeMinutes"),
-		("COE", "coeRuntimeMinutes"),
-		("P3P", "p3pRuntimeMinutes"),
-	):
-		minutes = _parse_positive_int(settings.get(key))
-		if minutes is not None:
-			overrides[test_type] = minutes
+	legacy_overrides = {
+		"P2P": _parse_positive_int(settings.get("p2pRuntimeMinutes")),
+		"COE": _parse_positive_int(settings.get("coeRuntimeMinutes")),
+		"P3P": _parse_positive_int(settings.get("p3pRuntimeMinutes")),
+	}
+
+	overrides: dict[str, dict[str, int]] = {}
+	for device_key, field_prefix in ((db.DEVICE_DUT, "dut"), (db.DEVICE_REF, "ref")):
+		device_overrides: dict[str, int] = {}
+		for test_type, suffix in (("P2P", "P2p"), ("COE", "Coe"), ("P3P", "P3p")):
+			minutes = _parse_positive_int(settings.get(f"{field_prefix}{suffix}RuntimeMinutes"))
+			if minutes is None:
+				minutes = legacy_overrides[test_type]
+			if minutes is not None:
+				device_overrides[test_type] = minutes
+		if device_overrides:
+			overrides[device_key] = device_overrides

 	return overrides

@@ -236,13 +236,17 @@ def parse_target_csv(
 	runtime_overrides: dict[str, Any] | None = None,
 ) -> ParseResult:
 	paths = _resolve_csv_paths(csv_path, smb_credentials=smb_credentials)
-	runtime_defaults = _resolve_runtime_defaults(runtime_overrides)
+	runtime_defaults_by_device = _resolve_runtime_defaults(runtime_overrides)
 	all_tests: list[TestRecord] = []
 	all_warnings: list[str] = []
 	seen_test_keys: set[tuple[str, str]] = set()

 	for path in paths:
-		parsed = _parse_single_csv(path, smb_credentials=smb_credentials, runtime_defaults=runtime_defaults)
+		parsed = _parse_single_csv(
+			path,
+			smb_credentials=smb_credentials,
+			runtime_defaults_by_device=runtime_defaults_by_device,
+		)
 		all_warnings.extend(parsed.warnings)
 		for record in parsed.tests:
 			key = (record.test_id, record.device)
@@ -257,13 +261,13 @@ def parse_target_csv(
 	return ParseResult(tests=all_tests, warnings=all_warnings)


-def _resolve_runtime_defaults(runtime_overrides: dict[str, Any] | None) -> dict[str, int]:
+def _coerce_runtime_defaults(raw_overrides: dict[str, Any] | None) -> dict[str, int]:
 	defaults = dict(RUNTIME_DEFAULTS)
-	if not runtime_overrides:
+	if not raw_overrides:
 		return defaults

 	for test_type in ("P2P", "COE", "P3P"):
-		raw_value = runtime_overrides.get(test_type)
+		raw_value = raw_overrides.get(test_type)
 		if raw_value is None:
 			continue
 		if isinstance(raw_value, str):
@@ -280,6 +284,25 @@ def _resolve_runtime_defaults(runtime_overrides: dict[str, Any] | None) -> dict[
 	return defaults


+def _resolve_runtime_defaults(runtime_overrides: dict[str, Any] | None) -> dict[str, dict[str, int]]:
+	legacy_defaults = _coerce_runtime_defaults(runtime_overrides)
+	if not runtime_overrides:
+		return {
+			DEVICE_DUT: dict(legacy_defaults),
+			DEVICE_REF: dict(legacy_defaults),
+		}
+
+	resolved: dict[str, dict[str, int]] = {}
+	for device in (DEVICE_DUT, DEVICE_REF):
+		device_defaults = dict(legacy_defaults)
+		raw_device_overrides = runtime_overrides.get(device)
+		if isinstance(raw_device_overrides, dict):
+			device_defaults.update(_coerce_runtime_defaults(raw_device_overrides))
+		resolved[device] = device_defaults
+
+	return resolved
+
+

 def _resolve_csv_paths(
 	csv_path: str | Path | list[str | Path] | tuple[str | Path, ...],
@@ -318,7 +341,7 @@ def _open_csv_handle(path: str, smb_credentials: dict[str, Any] | None):
 def _parse_single_csv(
 	path: str,
 	smb_credentials: dict[str, Any] | None,
-	runtime_defaults: dict[str, int],
+	runtime_defaults_by_device: dict[str, dict[str, int]],
 ) -> ParseResult:
 	with _open_csv_handle(path, smb_credentials=smb_credentials) as handle:
 		reader = csv.DictReader(handle)
@@ -352,7 +375,6 @@ def _parse_single_csv(
 			has_coe_pair = _normalize_yes_no(_row_get(row, "COE Pair")) if csv_format == "p2p_coe" else False
 			config = _build_config(row, csv_format)
 			signature = _victim_band_signature(row) if csv_format == "p2p_coe" else None
-			estimated_minutes = runtime_defaults.get(test_type, RUNTIME_DEFAULTS[test_type])
 			victim_band_source = "Victim Band" if csv_format == "p2p_coe" else "Band"
 			victim_band = _normalize_victim_band(_row_get(row, victim_band_source))

@@ -376,7 +398,7 @@ def _parse_single_csv(
 				victim_band=victim_band,
 				config=config,
 				throttled=throttled,
-				estimated_minutes=estimated_minutes,
+				estimated_minutes=RUNTIME_DEFAULTS[test_type],
 				status="pending",
 				excluded=False,
 				raw_payload=row,
@@ -401,6 +423,7 @@ def _parse_single_csv(
 				pairs = sorted(coe_by_signature_and_suffix.get(key, [])) if key else []

 			for device in (DEVICE_DUT, DEVICE_REF):
+				device_runtime_defaults = runtime_defaults_by_device.get(device, RUNTIME_DEFAULTS)
 				tests.append(
 					TestRecord(
 						test_id=record.test_id,
@@ -414,7 +437,7 @@ def _parse_single_csv(
 						victim_band=record.victim_band,
 						config=record.config,
 						throttled=record.throttled,
-						estimated_minutes=record.estimated_minutes,
+						estimated_minutes=device_runtime_defaults.get(record.test_type, RUNTIME_DEFAULTS[record.test_type]),
 						status=record.status,
 						excluded=record.excluded,
 						raw_payload=record.raw_payload,
@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+
+sys.path.append(str(Path(__file__).resolve().parent))
+
+from db import DEVICE_DUT, DEVICE_REF
+from parser import P2P_COE_REQUIRED_COLUMNS, parse_target_csv
+
+
+def _write_p2p_csv(directory: str, test_id: str = "P2PRXAX001") -> Path:
+    path = Path(directory) / "runtime-overrides.csv"
+    row = {column: "" for column in P2P_COE_REQUIRED_COLUMNS}
+    row["Priority"] = "1"
+    row["Index"] = "1"
+    row["TC ID"] = test_id
+
+    with path.open("w", encoding="utf-8", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=P2P_COE_REQUIRED_COLUMNS)
+        writer.writeheader()
+        writer.writerow(row)
+
+    return path
+
+
+class RuntimeOverridesParseTests(unittest.TestCase):
+    def test_parse_target_csv_applies_device_specific_runtime_overrides(self) -> None:
+        with TemporaryDirectory() as tmpdir:
+            csv_path = _write_p2p_csv(tmpdir)
+
+            parsed = parse_target_csv(
+                csv_path,
+                runtime_overrides={
+                    DEVICE_DUT: {"P2P": 91},
+                    DEVICE_REF: {"P2P": 123},
+                },
+            )
+
+        by_device = {record.device: record for record in parsed.tests}
+        self.assertEqual(by_device[DEVICE_DUT].estimated_minutes, 91)
+        self.assertEqual(by_device[DEVICE_REF].estimated_minutes, 123)
+
+    def test_parse_target_csv_keeps_legacy_runtime_override_for_both_devices(self) -> None:
+        with TemporaryDirectory() as tmpdir:
+            csv_path = _write_p2p_csv(tmpdir)
+
+            parsed = parse_target_csv(
+                csv_path,
+                runtime_overrides={"P2P": 88},
+            )
+
+        by_device = {record.device: record for record in parsed.tests}
+        self.assertEqual(by_device[DEVICE_DUT].estimated_minutes, 88)
+        self.assertEqual(by_device[DEVICE_REF].estimated_minutes, 88)
+
+
+if __name__ == "__main__":
+    unittest.main()