Validate cross-references within a FeatureDefinition.
Returns a list of error strings; empty list means the feat is valid.
Source code in dizzy/src/dizzy/feat_loader.py
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 | def validate_feat(feat: FeatureDefinition) -> list[str]:
"""Validate cross-references within a FeatureDefinition.
Returns a list of error strings; empty list means the feat is valid.
"""
errors: list[str] = []
command_names = {c.name for c in feat.commands or []}
event_names = {e.name for e in feat.events or []}
query_names = {q.name for q in feat.queries or []}
model_names = {m.name for m in feat.models or []}
models_by_name = {m.name: m for m in feat.models or []}
environment_names = {e.name for e in feat.environment or []}
telemetry_names = {t.name for t in feat.telemetry or []}
def _check_extras(kind: str, item) -> None:
"""Validate environment/telemetry references on a function definition."""
for name in getattr(item, "environment", None) or []:
if name not in environment_names:
errors.append(
f"{kind} '{item.name}': environment '{name}' not declared in environment"
)
for name in getattr(item, "telemetry", None) or []:
if name not in telemetry_names:
errors.append(f"{kind} '{item.name}': telemetry '{name}' not declared in telemetry")
for proc in feat.procedures or []:
if proc.command not in command_names:
errors.append(
f"procedure '{proc.name}': command '{proc.command}' not declared in commands"
)
for q in proc.queries or []:
if q not in query_names:
errors.append(f"procedure '{proc.name}': query '{q}' not declared in queries")
for e in proc.emits or []:
if e not in event_names:
errors.append(f"procedure '{proc.name}': emits '{e}' not declared in events")
_check_extras("procedure", proc)
for policy in feat.policies or []:
if policy.event not in event_names:
errors.append(f"policy '{policy.name}': event '{policy.event}' not declared in events")
for q in policy.queries or []:
if q not in query_names:
errors.append(f"policy '{policy.name}': query '{q}' not declared in queries")
for e in policy.emits or []:
if e not in command_names:
errors.append(f"policy '{policy.name}': emits '{e}' not declared in commands")
_check_extras("policy", policy)
for proj in feat.projections or []:
if proj.event not in event_names:
errors.append(f"projection '{proj.name}': event '{proj.event}' not declared in events")
if proj.model is not None and proj.model not in model_names:
errors.append(f"projection '{proj.name}': model '{proj.model}' not declared in models")
_check_extras("projection", proj)
for query in feat.queries or []:
_check_extras("querier", query)
if query.model is not None and query.model not in model_names:
errors.append(f"query '{query.name}': model '{query.model}' not declared in models")
if query.model is not None and query.adapter is None:
errors.append(f"query '{query.name}': model declared without adapter")
if query.adapter is not None and query.model is None:
errors.append(f"query '{query.name}': adapter declared without model")
if (
query.model is not None
and query.adapter is not None
and query.model in models_by_name
and query.adapter not in (models_by_name[query.model].adapters or [])
):
errors.append(
f"query '{query.name}': adapter '{query.adapter}' "
f"not declared on model '{query.model}'"
)
for proj in feat.projections or []:
if proj.model is not None and proj.adapter is None:
errors.append(f"projection '{proj.name}': model declared without adapter")
if proj.adapter is not None and proj.model is None:
errors.append(f"projection '{proj.name}': adapter declared without model")
if (
proj.model is not None
and proj.adapter is not None
and proj.model in models_by_name
and proj.adapter not in (models_by_name[proj.model].adapters or [])
):
errors.append(
f"projection '{proj.name}': adapter '{proj.adapter}' "
f"not declared on model '{proj.model}'"
)
logger.debug("validated feat", extra={"errors": len(errors)})
return errors
|