{{ message }}
Repository navigation
Expand file tree
/
Copy pathtest_assets_event_log_static.py
More file actions
337 lines (266 loc) · 12.9 KB
/
Copy pathtest_assets_event_log_static.py
File metadata and controls
337 lines (266 loc) · 12.9 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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
"""Static discipline check for the ``[assets-event]`` log lines.
Pure :mod:`ast` analysis: no module under ``app/assets`` is imported or
executed, so this check never needs a running ComfyUI. It deliberately lives at
the ``tests-unit`` root rather than under ``tests-unit/assets_test/``, whose
autouse fixture boots a ComfyUI subprocess for every test in that subtree.
Four rules are enforced over every emit call site:
a. keyword fields come from the closed vocabulary, the event is a string literal
b. no other log line anywhere carries the tag, so the tap only ever sees emits
c. the call sites present in the tree match an explicit manifest
d. ``error_type=`` values come from ``event_log.error_type()``, never a string
"""
from __future__ import annotations
import ast
from collections import Counter
from collections.abc import Iterator
from pathlib import Path
from typing import NamedTuple
from app.assets.event_log import ALLOWED_EVENTS, ALLOWED_FIELDS, TAG
REPO_ROOT = Path(__file__).resolve().parents[1]
MODULE_SCOPE = "<module>"
EVENT_LOG_NAME = "event_log"
LOG_METHODS = frozenset({"debug", "info", "warning", "error", "exception", "critical", "log"})
FUNCTION_NODES = (ast.FunctionDef, ast.AsyncFunctionDef)
class CallSite(NamedTuple):
"""The identity of one emit call: file, enclosing function, event."""
path: str
function: str
event: str
# The manifest of every tagged event this branch emits: (file, enclosing
# function, event) triples that must be present in the tree exactly as written.
EXPECTED_CALL_SITES: frozenset[CallSite] = frozenset(
{
# todo 10 - seeder lifecycle + the single assets.enabled site
CallSite("server.py", "__init__", "assets.enabled"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_started"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_completed"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_failed"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_cancelled"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.marked_missing"),
CallSite("app/assets/seeder.py", "mark_missing_outside_prefixes", "seeder.marked_missing"),
CallSite("app/assets/seeder.py", "_run_fast_phase", "seeder.batch_insert_failed"),
CallSite("app/assets/seeder.py", "_emit_marked_missing", "seeder.marked_missing"),
# todo 11 - scanner failure paths
CallSite("app/assets/scanner.py", "sync_root_safely", "scanner.fast_scan_failed"),
CallSite("app/assets/scanner.py", "live_references_safely", "scanner.fast_scan_failed"),
CallSite(
"app/assets/scanner.py", "mark_unlisted_references_missing_safely", "scanner.fast_scan_failed"
),
CallSite("app/assets/scanner.py", "sync_temp_references_safely", "scanner.temp_sync_failed"),
CallSite(
"app/assets/scanner.py", "mark_missing_outside_prefixes_safely", "scanner.mark_missing_failed"
),
CallSite("app/assets/scanner.py", "enrich_asset", "scanner.hash_failed"),
CallSite("app/assets/scanner.py", "enrich_asset", "scanner.hash_discarded_modified"),
CallSite("app/assets/scanner.py", "enrich_assets_batch", "scanner.enrich_failed"),
# todo 16 - discovery/enrich stat failures, emit-once per scan per site
CallSite("app/assets/scanner.py", "build_asset_specs", "scanner.stat_failed"),
CallSite("app/assets/scanner.py", "enrich_asset", "scanner.stat_failed"),
CallSite("app/assets/scanner.py", "seed_asset_specs", "scanner.invalid_mtime"),
CallSite(
"app/assets/scanner_admission.py", "tick_watch_list", "scanner.watch_stat_failed"
),
CallSite(
"app/assets/scanner_admission.py", "tick_watch_list", "scanner.watch_spec_failed"
),
CallSite(
"app/assets/scanner_admission.py", "tick_watch_list", "scanner.watch_seed_failed"
),
}
)
class Aliases(NamedTuple):
"""The names one module binds to the event_log module and its functions."""
module: frozenset[str]
emit: frozenset[str]
error_type: frozenset[str]
class Scan(NamedTuple):
"""Everything the AST walk learned about the tree."""
files: tuple[str, ...]
call_sites: Counter[CallSite]
vocabulary: tuple[str, ...]
event_names: tuple[str, ...]
error_types: tuple[str, ...]
tagged_logs: tuple[str, ...]
def _scanned_files(root: Path) -> tuple[str, ...]:
"""Every assets module, plus server.py for its single assets.enabled emit."""
assets = sorted(p.relative_to(root).as_posix() for p in root.glob("app/assets/**/*.py"))
return (*assets, "server.py")
def _scoped_nodes(tree: ast.Module) -> Iterator[tuple[ast.AST, str]]:
"""Yield every node paired with the name of its innermost enclosing function."""
def walk(node: ast.AST, scope: str) -> Iterator[tuple[ast.AST, str]]:
for child in ast.iter_child_nodes(node):
child_scope = child.name if isinstance(child, FUNCTION_NODES) else scope
yield child, child_scope
yield from walk(child, child_scope)
yield from walk(tree, MODULE_SCOPE)
def _resolve_aliases(tree: ast.Module) -> Aliases:
module: set[str] = set()
emit: set[str] = set()
error_type: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name.split(".")[-1] == EVENT_LOG_NAME:
module.add(alias.asname or alias.name)
elif isinstance(node, ast.ImportFrom):
from_event_log = (node.module or "").rsplit(".", 1)[-1] == EVENT_LOG_NAME
for alias in node.names:
if from_event_log and alias.name == "emit":
emit.add(alias.asname or alias.name)
elif from_event_log and alias.name == "error_type":
error_type.add(alias.asname or alias.name)
elif not from_event_log and alias.name == EVENT_LOG_NAME:
module.add(alias.asname or alias.name)
return Aliases(frozenset(module), frozenset(emit), frozenset(error_type))
def _dotted_name(node: ast.expr) -> str | None:
parts: list[str] = []
while isinstance(node, ast.Attribute):
parts.append(node.attr)
node = node.value
if not isinstance(node, ast.Name):
return None
parts.append(node.id)
return ".".join(reversed(parts))
def _is_emit_call(func: ast.expr, aliases: Aliases) -> bool:
if isinstance(func, ast.Attribute) and func.attr == "emit":
return _dotted_name(func.value) in aliases.module
return isinstance(func, ast.Name) and func.id in aliases.emit
def _is_unresolvable_emit_call(func: ast.expr, aliases: Aliases) -> bool:
return (
bool(aliases.module)
and isinstance(func, ast.Attribute)
and func.attr == "emit"
and _dotted_name(func.value) is None
)
def _is_error_type_call(value: ast.expr, aliases: Aliases) -> bool:
"""True only for a call to the sanctioned event_log.error_type()."""
if not isinstance(value, ast.Call):
return False
func = value.func
if isinstance(func, ast.Attribute) and func.attr == "error_type":
return _dotted_name(func.value) in aliases.module
return isinstance(func, ast.Name) and func.id in aliases.error_type
def _carries_tag(node: ast.Call) -> bool:
return any(
isinstance(child, ast.Constant) and isinstance(child.value, str) and TAG in child.value
for child in ast.walk(node)
)
def _is_log_call(func: ast.expr) -> bool:
return isinstance(func, ast.Attribute) and func.attr in LOG_METHODS
def _event_of(call: ast.Call) -> str | None:
"""The literal event name, or None when it is not a plain string literal."""
if len(call.args) != 1:
return None
first = call.args[0]
if not isinstance(first, ast.Constant) or not isinstance(first.value, str):
return None
return first.value
def _field_faults(call: ast.Call, aliases: Aliases) -> Iterator[tuple[str, str]]:
"""(category, reason) for every keyword that breaks rule (a) or rule (d)."""
for keyword in call.keywords:
if keyword.arg is None:
yield "vocabulary", "**splat fields cannot be checked statically"
elif keyword.arg not in ALLOWED_FIELDS:
yield "vocabulary", f"field {keyword.arg!r} is not in ALLOWED_FIELDS"
elif keyword.arg == "error_type" and not _is_error_type_call(keyword.value, aliases):
yield (
"error_types",
"error_type= must be a call to event_log.error_type(), got "
f"{ast.unparse(keyword.value)!r}",
)
def _file_faults(call: ast.Call, aliases: Aliases) -> Iterator[tuple[str, str]]:
if _is_unresolvable_emit_call(call.func, aliases):
yield "event_names", "the emit receiver cannot be resolved statically"
elif _is_emit_call(call.func, aliases):
event = _event_of(call)
if event not in ALLOWED_EVENTS:
yield "event_names", "the event must be one string literal in the allowed vocabulary"
yield from _field_faults(call, aliases)
elif _is_log_call(call.func) and _carries_tag(call):
yield "tagged_logs", f"log line carries {TAG} outside event_log.emit()"
def _scan_file(root: Path, relative: str) -> tuple[Counter[CallSite], list[tuple[str, str]]]:
tree = ast.parse((root / relative).read_text(encoding="utf-8"), filename=relative)
aliases = _resolve_aliases(tree)
sites: Counter[CallSite] = Counter()
faults: list[tuple[str, str]] = []
for node, scope in _scoped_nodes(tree):
if not isinstance(node, ast.Call):
continue
if _is_emit_call(node.func, aliases):
event = _event_of(node)
if event is not None and event in ALLOWED_EVENTS:
sites[CallSite(relative, scope, event)] += 1
for category, reason in _file_faults(node, aliases):
faults.append((category, f"{relative}:{node.lineno}: {reason}"))
return sites, faults
def _write_scan_fixture(root: Path, relative: str, source: str) -> None:
path = root / relative
path.parent.mkdir(parents=True)
path.write_text(source, encoding="utf-8")
def scan_repository(root: Path = REPO_ROOT) -> Scan:
files = _scanned_files(root)
sites: Counter[CallSite] = Counter()
found: dict[str, list[str]] = {"vocabulary": [], "event_names": [], "error_types": [], "tagged_logs": []}
for relative in files:
file_sites, faults = _scan_file(root, relative)
sites += file_sites
for category, message in faults:
found[category].append(message)
return Scan(
files=files,
call_sites=sites,
vocabulary=tuple(found["vocabulary"]),
event_names=tuple(found["event_names"]),
error_types=tuple(found["error_types"]),
tagged_logs=tuple(found["tagged_logs"]),
)
SCAN = scan_repository()
def test_the_walk_actually_covers_the_assets_tree() -> None:
"""Guards every other check: a broken glob would make them all vacuous."""
assert "app/assets/event_log.py" in SCAN.files
assert "app/assets/seeder.py" in SCAN.files
assert "server.py" in SCAN.files
assert len(SCAN.files) > 20
def test_emit_fields_stay_inside_the_closed_vocabulary() -> None:
assert SCAN.vocabulary == ()
def test_emit_events_are_literals_in_the_allowed_vocabulary() -> None:
assert SCAN.event_names == ()
def test_error_type_values_come_from_event_log_error_type() -> None:
assert SCAN.error_types == ()
def test_no_other_log_line_carries_the_event_tag() -> None:
assert SCAN.tagged_logs == ()
def test_call_sites_match_the_manifest() -> None:
manifest = Counter(EXPECTED_CALL_SITES)
unexpected = SCAN.call_sites - manifest
missing = manifest - SCAN.call_sites
assert not unexpected, (
f"emit call sites not in the manifest: {sorted(unexpected)} — add them to "
"EXPECTED_CALL_SITES"
)
assert not missing, f"manifest call sites absent from the tree: {sorted(missing)}"
def test_qualified_event_log_import_is_scanned(tmp_path: Path) -> None:
relative = "app/assets/qualified.py"
_write_scan_fixture(
tmp_path,
relative,
"import app.assets.event_log\n\n"
"def probe():\n"
' app.assets.event_log.emit("seeder.scan_started", phase="fast")\n',
)
sites, faults = _scan_file(tmp_path, relative)
assert faults == []
assert sites == Counter(
{CallSite(relative, "probe", "seeder.scan_started"): 1}
)
def test_unresolvable_emit_receiver_is_a_scan_failure(tmp_path: Path) -> None:
relative = "app/assets/dynamic.py"
_write_scan_fixture(
tmp_path,
relative,
"from app.assets import event_log\n\n"
"def probe(provider):\n"
' provider().emit("seeder.scan_started", phase="fast")\n',
)
sites, faults = _scan_file(tmp_path, relative)
assert sites == Counter()
assert [category for category, _reason in faults] == ["event_names"]
You can’t perform that action at this time.