|
12 | 12 |
|
13 | 13 | from types import SimpleNamespace |
14 | 14 |
|
15 | | -from rust_ok import Ok |
| 15 | +from rust_ok import Err, Ok |
16 | 16 |
|
17 | 17 | import r2x_sienna.plugins as plugins |
18 | 18 | import r2x_sienna.upgrader.data_upgrader as data_upgrader |
@@ -383,3 +383,107 @@ def test_version_detector_and_run_sienna_upgrades_paths(tmp_path: Path, monkeypa |
383 | 383 | monkeypatch.setattr(data_upgrader, "_resolve_json_path", lambda path: None) |
384 | 384 | result_ok = run_sienna_upgrades(json_path=tmp_path, ctx=SimpleNamespace()) |
385 | 385 | assert result_ok.is_ok() |
| 386 | + |
| 387 | + |
| 388 | +def test_upgrader_upgrade_error_paths(tmp_path: Path, monkeypatch) -> None: |
| 389 | + upgrader = SiennaUpgrader(tmp_path) |
| 390 | + |
| 391 | + # current_version autodetection fails |
| 392 | + monkeypatch.setattr(upgrader.version_reader, "read_version", lambda _path: None) |
| 393 | + result = upgrader.upgrade(current_version=None) |
| 394 | + assert result.is_err() |
| 395 | + assert "Could not determine Sienna version" in str(result.err()) |
| 396 | + |
| 397 | + # SYSTEM upgrade with no JSON files in directory |
| 398 | + empty_dir = tmp_path / "empty_system_dir" |
| 399 | + empty_dir.mkdir() |
| 400 | + empty_upgrader = SiennaUpgrader(empty_dir) |
| 401 | + result = empty_upgrader.upgrade(current_version="1.0.0", upgrade_type=UpgradeType.SYSTEM) |
| 402 | + assert result.is_err() |
| 403 | + assert "No JSON file found" in str(result.err()) |
| 404 | + |
| 405 | + # SYSTEM upgrade with invalid JSON file |
| 406 | + bad_json = tmp_path / "bad_system.json" |
| 407 | + bad_json.write_text("{invalid", encoding="utf-8") |
| 408 | + bad_upgrader = SiennaUpgrader(bad_json) |
| 409 | + result = bad_upgrader.upgrade(current_version="1.0.0", upgrade_type=UpgradeType.SYSTEM) |
| 410 | + assert result.is_err() |
| 411 | + assert "Failed to load JSON" in str(result.err()) |
| 412 | + |
| 413 | + |
| 414 | +def test_upgrader_upgrade_step_failure_and_file_mode_error(tmp_path: Path, monkeypatch) -> None: |
| 415 | + system_json = tmp_path / "system.json" |
| 416 | + system_json.write_text(json.dumps({"data": {"components": []}}), encoding="utf-8") |
| 417 | + |
| 418 | + def broken_upgrade(_data: dict[str, Any]) -> dict[str, Any]: |
| 419 | + raise RuntimeError("step failed") |
| 420 | + |
| 421 | + failing_system_step = UpgradeStep( |
| 422 | + name="broken_upgrade", |
| 423 | + func=broken_upgrade, |
| 424 | + target_version="2.0.0", |
| 425 | + upgrade_type=UpgradeType.SYSTEM, |
| 426 | + ) |
| 427 | + upgrader = SiennaUpgrader(system_json, steps=[failing_system_step]) |
| 428 | + result = upgrader.upgrade(current_version="1.0.0", upgrade_type=UpgradeType.SYSTEM) |
| 429 | + assert result.is_err() |
| 430 | + assert "Failed broken_upgrade" in str(result.err()) |
| 431 | + |
| 432 | + file_step = UpgradeStep( |
| 433 | + name="file_step", |
| 434 | + func=lambda path: path, |
| 435 | + target_version="2.0.0", |
| 436 | + upgrade_type=UpgradeType.FILE, |
| 437 | + ) |
| 438 | + file_upgrader = SiennaUpgrader(system_json, steps=[file_step]) |
| 439 | + monkeypatch.setattr(data_upgrader, "run_upgrade_step", lambda path, step: Err("file step failed")) |
| 440 | + result = file_upgrader.upgrade(current_version="1.0.0", upgrade_type=UpgradeType.FILE) |
| 441 | + assert result.is_err() |
| 442 | + assert "file step failed" in str(result.err()) |
| 443 | + |
| 444 | + |
| 445 | +def test_run_sienna_upgrades_store_and_h5_error_paths(tmp_path: Path, monkeypatch) -> None: |
| 446 | + system_json = tmp_path / "sys.json" |
| 447 | + system_json.write_text( |
| 448 | + json.dumps({"data_format_version": "5.0.0", "data": {"time_series_storage_file": "ts.h5"}}), |
| 449 | + encoding="utf-8", |
| 450 | + ) |
| 451 | + h5_path = tmp_path / "ts.h5" |
| 452 | + h5_path.write_bytes(b"dummy") |
| 453 | + |
| 454 | + class Store: |
| 455 | + folder = tmp_path |
| 456 | + |
| 457 | + # Cover store.folder branch and "could not detect version" skip. |
| 458 | + monkeypatch.setattr(data_upgrader.SiennaUpgrader.version_reader, "read_version", lambda path: None) |
| 459 | + monkeypatch.setattr(data_upgrader, "_resolve_json_path", lambda path: None) |
| 460 | + result_skip = run_sienna_upgrades(store=Store(), ctx=SimpleNamespace()) |
| 461 | + assert result_skip.is_ok() |
| 462 | + |
| 463 | + # Cover upgrade result error branch. |
| 464 | + monkeypatch.setattr(data_upgrader.SiennaUpgrader.version_reader, "read_version", lambda path: "5.0.0") |
| 465 | + monkeypatch.setattr(data_upgrader.SiennaUpgrader, "upgrade", lambda self, **kwargs: Err("upgrade failed")) |
| 466 | + result_upgrade_err = run_sienna_upgrades(store=Store(), ctx=SimpleNamespace()) |
| 467 | + assert result_upgrade_err.is_err() |
| 468 | + assert "upgrade failed" in str(result_upgrade_err.err()) |
| 469 | + |
| 470 | + # Cover JSON/H5 inspection error branch. |
| 471 | + monkeypatch.setattr(data_upgrader.SiennaUpgrader, "upgrade", lambda self, **kwargs: Ok(self.path)) |
| 472 | + monkeypatch.setattr(data_upgrader, "_resolve_json_path", lambda path: system_json) |
| 473 | + monkeypatch.setattr( |
| 474 | + data_upgrader, |
| 475 | + "_resolve_time_series_h5_path", |
| 476 | + lambda path: (_ for _ in ()).throw(RuntimeError("inspect failed")), |
| 477 | + ) |
| 478 | + result_inspect_err = run_sienna_upgrades(store=Store(), ctx=SimpleNamespace()) |
| 479 | + assert result_inspect_err.is_err() |
| 480 | + assert "Failed to inspect JSON" in str(result_inspect_err.err()) |
| 481 | + |
| 482 | + # Cover H5 upgrade error branch. |
| 483 | + monkeypatch.setattr(data_upgrader, "_resolve_time_series_h5_path", lambda path: h5_path) |
| 484 | + monkeypatch.setattr( |
| 485 | + data_upgrader, "_upgrade_h5_time_series_metadata", lambda path: Err("h5 upgrade failed") |
| 486 | + ) |
| 487 | + result_h5_err = run_sienna_upgrades(store=Store(), ctx=SimpleNamespace()) |
| 488 | + assert result_h5_err.is_err() |
| 489 | + assert "h5 upgrade failed" in str(result_h5_err.err()) |
0 commit comments