PICurv 0.1.0
A Parallel Particle-In-Cell Solver for Curvilinear LES
 
Loading...
Searching...
No Matches
Functions | Variables
audit_freshness Namespace Reference

Functions

str digest_of (list paths)
 Deterministic digest over an ordered list of files.
 
list surface_paths (dict surface)
 The files a surface fingerprints.
 
str enforcement_of (dict surface)
 Whether a suspicion on this surface blocks.
 
list validate_manifest (list surfaces, set published)
 Structural checks on the manifest itself.
 
dict evaluate (list surfaces, set published)
 Classify every surface as current, suspect, or an integrity failure.
 
list blocking_entries (dict report)
 The suspicions that must fail the build.
 
list attest (list surfaces, list wanted)
 Record the current digests as reviewed.
 
int main ()
 Report freshness, or re-attest reviewed surfaces.
 

Variables

 REPO_ROOT = Path(__file__).resolve().parents[2]
 
str MANIFEST = REPO_ROOT / "tests" / "tooling" / "freshness_manifest.json"
 
str PAGE_TYPES = REPO_ROOT / "tests" / "tooling" / "page_types.json"
 
tuple VALID_TIERS = ("hard", "soft")
 
tuple VALID_ENFORCEMENT = ("blocking", "report")
 
dict SURFACE_KEYS
 
str UNATTESTED = "unattested"
 

Detailed Description

Tiered documentation freshness: hard suspicion, soft suspicion, integrity failure.

A fingerprint answers one question: has the thing a page describes changed since a
human or agent last compared the page against it? It never claims the page is wrong,
and matching digests never claim the page is right. It routes review.

Function Documentation

◆ digest_of()

str audit_freshness.digest_of ( list  paths)

Deterministic digest over an ordered list of files.

The path is hashed alongside the bytes so that moving content between watched files is itself a change.

Parameters
[in]pathsRepository-relative paths.
Returns
Hex digest prefixed with its algorithm.

Definition at line 31 of file audit_freshness.py.

31def digest_of(paths: list) -> str:
32 """!
33 @brief Deterministic digest over an ordered list of files.
34
35 @details The path is hashed alongside the bytes so that moving content between
36 watched files is itself a change.
37 @param[in] paths Repository-relative paths.
38 @return Hex digest prefixed with its algorithm.
39 """
40 accumulator = hashlib.sha256()
41 for relative in sorted(paths):
42 accumulator.update(relative.encode("utf-8"))
43 accumulator.update(b"\0")
44 accumulator.update((REPO_ROOT / relative).read_bytes())
45 accumulator.update(b"\0")
46 return f"sha256:{accumulator.hexdigest()}"
47
48
Here is the caller graph for this function:

◆ surface_paths()

list audit_freshness.surface_paths ( dict  surface)

The files a surface fingerprints.

Parameters
[in]surfaceOne manifest entry.
Returns
Repository-relative paths.

Definition at line 49 of file audit_freshness.py.

49def surface_paths(surface: dict) -> list:
50 """!
51 @brief The files a surface fingerprints.
52 @param[in] surface One manifest entry.
53 @return Repository-relative paths.
54 """
55 if surface["tier"] == "hard":
56 return [surface["artifact"]]
57 return list(surface.get("watched_paths", []))
58
59
Head of a generic C-style linked list.
Definition variables.h:475
Here is the caller graph for this function:

◆ enforcement_of()

str audit_freshness.enforcement_of ( dict  surface)

Whether a suspicion on this surface blocks.

Parameters
[in]surfaceOne manifest entry.
Returns
"blocking" or "report".

Definition at line 60 of file audit_freshness.py.

60def enforcement_of(surface: dict) -> str:
61 """!
62 @brief Whether a suspicion on this surface blocks.
63 @param[in] surface One manifest entry.
64 @return "blocking" or "report".
65 """
66 return surface.get("enforcement") or ("blocking" if surface["tier"] == "hard" else "report")
67
68
Here is the caller graph for this function:

◆ validate_manifest()

list audit_freshness.validate_manifest ( list  surfaces,
set  published 
)

Structural checks on the manifest itself.

A malformed manifest is an integrity failure, not ordinary staleness: it means coverage is broken and the absence of suspicion proves nothing.

Parameters
[in]surfacesManifest entries.
[in]publishedIds of published pages.
Returns
List of integrity failures.

Definition at line 69 of file audit_freshness.py.

69def validate_manifest(surfaces: list, published: set) -> list:
70 """!
71 @brief Structural checks on the manifest itself.
72
73 @details A malformed manifest is an integrity failure, not ordinary staleness: it
74 means coverage is broken and the absence of suspicion proves nothing.
75 @param[in] surfaces Manifest entries.
76 @param[in] published Ids of published pages.
77 @return List of integrity failures.
78 """
79 failures: list = []
80 seen: set = set()
81 for surface in surfaces:
82 identifier = surface.get("id")
83 if not identifier:
84 failures.append("a surface declares no id")
85 continue
86 if identifier in seen:
87 failures.append(f"{identifier}: declared more than once")
88 seen.add(identifier)
89 unknown = set(surface) - SURFACE_KEYS
90 if unknown:
91 failures.append(f"{identifier}: unknown field(s) {sorted(unknown)}")
92 if surface.get("tier") not in VALID_TIERS:
93 failures.append(f"{identifier}: tier {surface.get('tier')!r} is not hard or soft")
94 continue
95 if enforcement_of(surface) not in VALID_ENFORCEMENT:
96 failures.append(f"{identifier}: enforcement must be blocking or report")
97 if surface["tier"] == "hard":
98 if not surface.get("artifact"):
99 failures.append(
100 f"{identifier}: a hard surface must name the normalized artifact it "
101 f"fingerprints"
102 )
103 if not surface.get("regenerate"):
104 failures.append(
105 f"{identifier}: a hard surface must name the command that regenerates it"
106 )
107 else:
108 if not surface.get("watched_paths"):
109 failures.append(f"{identifier}: a soft surface must watch at least one path")
110 if enforcement_of(surface) == "blocking" and not surface.get("promotion_reason"):
111 failures.append(
112 f"{identifier}: a soft surface promoted to blocking must state why"
113 )
114 if not surface.get("owning_pages"):
115 failures.append(
116 f"{identifier}: no owning page; a fingerprint with nowhere to route its "
117 f"suspicion cannot ask anyone to review anything"
118 )
119 for field in ("owning_pages", "dependent_pages"):
120 for page in surface.get(field, []):
121 if page not in published:
122 failures.append(f"{identifier}: {field} names '{page}', which is not published")
123 for relative in surface_paths(surface):
124 if not (REPO_ROOT / relative).is_file():
125 failures.append(
126 f"{identifier}: watched path '{relative}' does not exist; coverage is "
127 f"broken, not merely stale"
128 )
129 return failures
130
131
Here is the call graph for this function:
Here is the caller graph for this function:

◆ evaluate()

dict audit_freshness.evaluate ( list  surfaces,
set  published 
)

Classify every surface as current, suspect, or an integrity failure.

Parameters
[in]surfacesManifest entries.
[in]publishedIds of published pages.
Returns
Report with the four classifications and the pages needing review.

Definition at line 132 of file audit_freshness.py.

132def evaluate(surfaces: list, published: set) -> dict:
133 """!
134 @brief Classify every surface as current, suspect, or an integrity failure.
135 @param[in] surfaces Manifest entries.
136 @param[in] published Ids of published pages.
137 @return Report with the four classifications and the pages needing review.
138 """
139 integrity = validate_manifest(surfaces, published)
140 report = {"integrity_failures": integrity, "hard_current": [], "hard_suspect": [],
141 "soft_current": [], "soft_suspect": [], "unattested": []}
142 if integrity:
143 return report
144 for surface in surfaces:
145 current = digest_of(surface_paths(surface))
146 attested = surface.get("attested_digest")
147 entry = {
148 "id": surface["id"],
149 "title": surface.get("title", surface["id"]),
150 "tier": surface["tier"],
151 "enforcement": enforcement_of(surface),
152 "current_digest": current,
153 "attested_digest": attested,
154 "pages": list(surface.get("owning_pages", [])) +
155 list(surface.get("dependent_pages", [])),
156 "regenerate": surface.get("regenerate"),
157 }
158 if attested in (None, UNATTESTED):
159 report["unattested"].append(entry)
160 elif attested == current:
161 report[f"{surface['tier']}_current"].append(entry)
162 else:
163 report[f"{surface['tier']}_suspect"].append(entry)
164 return report
165
166
Here is the call graph for this function:
Here is the caller graph for this function:

◆ blocking_entries()

list audit_freshness.blocking_entries ( dict  report)

The suspicions that must fail the build.

Parameters
[in]reportOutput of evaluate().
Returns
Entries whose enforcement is blocking.

Definition at line 167 of file audit_freshness.py.

167def blocking_entries(report: dict) -> list:
168 """!
169 @brief The suspicions that must fail the build.
170 @param[in] report Output of evaluate().
171 @return Entries whose enforcement is blocking.
172 """
173 suspects = report["hard_suspect"] + report["soft_suspect"] + report["unattested"]
174 return [entry for entry in suspects if entry["enforcement"] == "blocking"]
175
176
Here is the caller graph for this function:

◆ attest()

list audit_freshness.attest ( list  surfaces,
list  wanted 
)

Record the current digests as reviewed.

Parameters
[in]surfacesManifest entries, mutated in place.
[in]wantedSurface ids, or ["all"].
Returns
Ids that were re-attested.

Definition at line 177 of file audit_freshness.py.

177def attest(surfaces: list, wanted: list) -> list:
178 """!
179 @brief Record the current digests as reviewed.
180 @param[in] surfaces Manifest entries, mutated in place.
181 @param[in] wanted Surface ids, or ["all"].
182 @return Ids that were re-attested.
183 """
184 changed = []
185 for surface in surfaces:
186 if wanted != ["all"] and surface["id"] not in wanted:
187 continue
188 current = digest_of(surface_paths(surface))
189 if surface.get("attested_digest") != current:
190 surface["attested_digest"] = current
191 changed.append(surface["id"])
192 return changed
193
194
Here is the call graph for this function:
Here is the caller graph for this function:

◆ main()

int audit_freshness.main ( )

Report freshness, or re-attest reviewed surfaces.

Returns
Process status code.

Definition at line 195 of file audit_freshness.py.

195def main() -> int:
196 """!
197 @brief Report freshness, or re-attest reviewed surfaces.
198 @return Process status code.
199 """
200 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
201 parser.add_argument("--attest", nargs="+", metavar="ID",
202 help="Record current digests as reviewed. Pass 'all' for every surface.")
203 parser.add_argument("--json", action="store_true", help="Emit the report as JSON.")
204 parser.add_argument("--page", metavar="PAGE_ID",
205 help="Report only the surfaces that route review to this page.")
206 arguments = parser.parse_args()
207
208 document = json.loads(MANIFEST.read_text(encoding="utf-8"))
209 surfaces = document["surfaces"]
210 published = set(json.loads(PAGE_TYPES.read_text(encoding="utf-8"))["assignments"])
211
212 if arguments.attest:
213 failures = validate_manifest(surfaces, published)
214 if failures:
215 print("Cannot attest against a manifest with integrity failures:", file=sys.stderr)
216 for failure in failures:
217 print(f" {failure}", file=sys.stderr)
218 return 2
219 changed = attest(surfaces, arguments.attest)
220 MANIFEST.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
221 if changed:
222 print("Re-attested: " + ", ".join(changed))
223 print("This records that the owning pages were compared against these sources.\n"
224 "It is not a claim that the comparison found nothing - only that it happened.")
225 else:
226 print("Nothing to re-attest; every requested surface already matches.")
227 return 0
228
229 report = evaluate(surfaces, published)
230 if arguments.page:
231 for key in ("hard_current", "hard_suspect", "soft_current", "soft_suspect", "unattested"):
232 report[key] = [e for e in report[key] if arguments.page in e["pages"]]
233 if arguments.json:
234 print(json.dumps(report, indent=2))
235 return 1 if report["integrity_failures"] or blocking_entries(report) else 0
236
237 if report["integrity_failures"]:
238 print("Freshness integrity failures (coverage is broken):", file=sys.stderr)
239 for failure in report["integrity_failures"]:
240 print(f" {failure}", file=sys.stderr)
241 return 2
242
243 for entry in report["hard_suspect"]:
244 print(f"HARD SUSPECT {entry['id']} - {entry['title']}", file=sys.stderr)
245 print(f" the normalized artifact changed since it was last reviewed", file=sys.stderr)
246 print(f" review: {', '.join(entry['pages'])}", file=sys.stderr)
247 print(f" then: python3 tests/tooling/audit_freshness.py --attest {entry['id']}",
248 file=sys.stderr)
249 for entry in report["soft_suspect"]:
250 stream = sys.stderr if entry["enforcement"] == "blocking" else sys.stdout
251 label = "SOFT SUSPECT (blocking)" if entry["enforcement"] == "blocking" else "soft suspect"
252 print(f"{label} {entry['id']} - {entry['title']}", file=stream)
253 print(f" watched sources changed; no semantic extractor exists for this surface",
254 file=stream)
255 print(f" review: {', '.join(entry['pages'])}", file=stream)
256 for entry in report["unattested"]:
257 blocking_entry = entry["enforcement"] == "blocking"
258 stream = sys.stderr if blocking_entry else sys.stdout
259 label = "UNATTESTED (blocking)" if blocking_entry else "unattested"
260 print(f"{label} {entry['id']} - {entry['title']}: never reviewed against its "
261 f"sources", file=stream)
262 print(f" review: {', '.join(entry['pages'])}", file=stream)
263
264 counts = (f"{len(report['hard_current'])} hard-current, {len(report['hard_suspect'])} "
265 f"hard-suspect, {len(report['soft_current'])} soft-current, "
266 f"{len(report['soft_suspect'])} soft-suspect, "
267 f"{len(report['unattested'])} never attested")
268 blocking = blocking_entries(report)
269 if blocking:
270 print(f"\nFreshness: {counts}. Blocking on {len(blocking)} surface(s).", file=sys.stderr)
271 print("A suspicion is a request to review, not a claim the page is wrong. Compare the\n"
272 "page with its sources, correct what drifted, then re-attest.", file=sys.stderr)
273 return 1
274 print(f"Freshness: {counts}.")
275 if report["soft_suspect"] or report["unattested"]:
276 print("Advisory entries above do not block. A never-attested surface means no page has\n"
277 "yet been compared against it - an honest gap, not a failure.")
278 return 0
279
280
int main(int argc, char **argv)
Entry point for the postprocessor executable.
Here is the call graph for this function:
Here is the caller graph for this function:

Variable Documentation

◆ REPO_ROOT

audit_freshness.REPO_ROOT = Path(__file__).resolve().parents[2]

Definition at line 18 of file audit_freshness.py.

◆ MANIFEST

str audit_freshness.MANIFEST = REPO_ROOT / "tests" / "tooling" / "freshness_manifest.json"

Definition at line 19 of file audit_freshness.py.

◆ PAGE_TYPES

str audit_freshness.PAGE_TYPES = REPO_ROOT / "tests" / "tooling" / "page_types.json"

Definition at line 20 of file audit_freshness.py.

◆ VALID_TIERS

tuple audit_freshness.VALID_TIERS = ("hard", "soft")

Definition at line 22 of file audit_freshness.py.

◆ VALID_ENFORCEMENT

tuple audit_freshness.VALID_ENFORCEMENT = ("blocking", "report")

Definition at line 23 of file audit_freshness.py.

◆ SURFACE_KEYS

dict audit_freshness.SURFACE_KEYS
Initial value:
1= {"id", "title", "tier", "enforcement", "promotion_reason", "artifact",
2 "regenerate", "watched_paths", "owning_pages", "dependent_pages",
3 "attested_digest", "attested_scope", "note"}

Definition at line 24 of file audit_freshness.py.

◆ UNATTESTED

str audit_freshness.UNATTESTED = "unattested"

Definition at line 28 of file audit_freshness.py.