Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / check / drift.py: 0%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# Licensed under the Apache License, Version 2.0 

3# ============================================================================ 

4"""Drift check: is the on-disk artifact still the bytes the frozen plan emits? 

5 

6``check --all`` re-runs the emit offline and compares the 

7regenerated bytes against the bundle currently on disk. A mismatch means the 

8bundle was hand-edited after generation (or the source/plan changed without a 

9signature bump) — the artifact is stale and must not be trusted. 

10 

11Distinct from ``verify_signature`` (which compares the *projected spec* against 

12the stored meta) and ``verify_output_hashes`` (which compares file bytes 

13against the *hashes recorded in meta*): this check re-derives the expected 

14bytes from the meta itself, so it catches an edit that was also re-touched in 

15meta, or an emit-layer regression. 

16""" 

17from __future__ import annotations 

18 

19import logging 

20import os 

21from typing import Any, Optional 

22 

23from hyper_parallel.codegen.artifact import resolve_artifact_layout 

24 

25logger = logging.getLogger(__name__) 

26 

27 

28def verify_drift_intact(meta: Any, yaml_path: str) -> list[str]: 

29 """Regenerate the bundle from ``meta`` and compare with the on-disk files. 

30 

31 Returns a list of drift descriptions (empty = intact). Raises 

32 ``FileNotFoundError`` when meta or the bundle directory is missing. When 

33 the original modeling source file can no longer be read (e.g. a remote-code 

34 cache was evicted), drift is *reported*, not silently skipped: the artifact 

35 cannot be re-validated against its origin. 

36 """ 

37 layout = resolve_artifact_layout(yaml_path, model_name=_model_name(meta)) 

38 if not os.path.isdir(layout.artifact_dir): 

39 raise FileNotFoundError(f"codegen drift: bundle missing at {layout.artifact_dir}") 

40 

41 from hyper_parallel.codegen.emit import emit_bundle 

42 

43 files = emit_bundle(layout, meta) 

44 

45 drifts: list[str] = [] 

46 for name, expected in files.items(): 

47 path = os.path.join(layout.artifact_dir, name) 

48 if not os.path.isfile(path): 

49 drifts.append(f"{name}: missing on disk") 

50 continue 

51 with open(path, "r", encoding="utf-8", errors="replace") as handle: 

52 actual = handle.read() 

53 if actual != expected: 

54 drifts.append(f"{name}: bytes differ from re-emitted bundle") 

55 return drifts 

56 

57 

58def _model_name(meta: Any) -> Optional[str]: 

59 """The model name recorded in meta (the generated filename stems from it).""" 

60 return getattr(meta, "model_name", None) 

61 

62 

63__all__ = [ 

64 "verify_drift_intact", 

65]