forked from ChelseaKR/tods-validate
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathfindings.py
More file actions
157 lines (135 loc) · 5.69 KB
/
Copy pathfindings.py
File metadata and controls
157 lines (135 loc) · 5.69 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
"""Finding and severity types shared by rules and reporting."""
from __future__ import annotations
import hashlib
import json
from collections.abc import Mapping
from dataclasses import dataclass
from dataclasses import field as dataclass_field
from enum import IntEnum
class Severity(IntEnum):
"""Ordered so that max() picks the worst severity."""
INFO = 0
WARNING = 1
ERROR = 2
def __str__(self) -> str:
return self.name
def _fingerprint_payload(
rule_id: str,
file: str | None,
field: str | None,
data: Mapping[str, str] | None,
) -> str:
"""Canonical JSON payload for a content fingerprint.
Built only from stable content: rule ID, file, field, and the rule's own
structured machine context (``data`` — offending value, referenced ID, and
other FIX-05 parameters). Row and message are deliberately excluded so
inserting or removing an unrelated row elsewhere in the file does not
change every subsequent finding's identity.
"""
items = sorted((data or {}).items())
value = (data or {}).get("value") if data else None
canonical = {
"rule_id": rule_id,
"file": file,
"field": field,
"data": items,
"value": value,
}
return json.dumps(canonical, sort_keys=True, default=str)
def fingerprint_from_parts(
rule_id: str,
file: str | None,
field: str | None,
data: Mapping[str, str] | None,
) -> str:
"""Content fingerprint computed from raw parts.
Shared by ``Finding.fingerprint()`` and by ``baseline.py``, which
recomputes a fingerprint from a stored baseline dict that predates the
``fingerprint`` field but already carries ``data``.
"""
payload = _fingerprint_payload(rule_id, file, field, data)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
@dataclass(frozen=True)
class Finding:
"""One validation result.
Messages are written for transit schedulers, not programmers: say what is
wrong, where, and what good looks like.
"""
rule_id: str
severity: Severity
message: str
file: str | None = None
# 1-based line number in the CSV file, counting the header as line 1.
row: int | None = None
field: str | None = None
suggestion: str | None = None
# Structured machine context (FIX-05): offending value, expected/allowed
# values, referenced ID, or other rule-specific key/value pairs. Additive
# and optional — None for rules that have not been migrated to emit it
# yet. This is what makes ``fingerprint()`` below distinguish findings
# without relying on row number. Excluded from equality and hashing (a
# Mapping is not hashable), so Finding stays hashable and comparable.
data: Mapping[str, str] | None = dataclass_field(default=None, compare=False)
# pointer() of the "root" finding that structurally caused this one (e.g. a
# TODS-E201 fired only because a TODS-E104 ragged row left the field blank).
# None for findings that are not a known downstream echo of another one.
# Never used to drop a finding from machine-readable formats -- it only lets
# renderers collapse an echo under its cause for humans.
caused_by: str | None = None
# The rule's spec-declared severity, set only when local policy (a
# `[severity]` table, see config.py) remapped this finding to a
# different severity. None means ``severity`` is the spec's own value.
# Every report renderer must disclose remapped findings; see report.py.
severity_original: Severity | None = None
def location(self) -> str:
parts = []
if self.file:
parts.append(self.file)
if self.row is not None:
parts.append(f"row {self.row}")
if self.field:
parts.append(f"field {self.field!r}")
return ", ".join(parts)
def pointer(self) -> str | None:
"""A stable, machine-parseable location identifier.
Of the form ``file.txt#L4`` or ``file.txt#L4/field``, so consumers can
deep-link a finding without parsing the human ``location()`` string.
Returns None for findings not tied to a file.
"""
if not self.file:
return None
ref = self.file
if self.row is not None:
ref += f"#L{self.row}"
if self.field:
ref += f"/{self.field}"
return ref
def fingerprint(self) -> str:
"""Content-anchored identity, stable across row renumbering.
Excludes ``row`` and ``message`` on purpose: inserting a row earlier
in the file, or a message wording tweak, must not change identity. It
is a heuristic, not a guarantee — two distinct findings that share
rule, file, field, and (if present) identical ``data`` will
fingerprint identically, and a row whose *content* changes (not just
its position) will still churn even though its row number may not.
See ``baseline.py`` for the honesty note this implies for
``--baseline``.
"""
return fingerprint_from_parts(self.rule_id, self.file, self.field, self.data)
def to_dict(self) -> dict[str, object]:
return {
"rule_id": self.rule_id,
"severity": str(self.severity),
"file": self.file,
"row": self.row,
"field": self.field,
"location": self.pointer(),
"data": dict(self.data) if self.data is not None else None,
"message": self.message,
"suggestion": self.suggestion,
"caused_by": self.caused_by,
"fingerprint": self.fingerprint(),
"severity_original": (
str(self.severity_original) if self.severity_original is not None else None
),
}