generated from coulomb/repo-seed
tegwick
30598fd1ad
Add safe action interpolation and for_each binding for rule fan-out, update the weekly SBOM definition, cover the new evaluation path, and reconcile activity-core scope/workplans for the State Hub sync.
118 lines
3.3 KiB
Python
118 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from activity_core.rules.actions import expand_rule_actions
|
|
from activity_core.rules.evaluator import UnsafeExpression
|
|
|
|
|
|
class _Attrs:
|
|
def __init__(self, **kw):
|
|
for k, v in kw.items():
|
|
setattr(self, k, v)
|
|
|
|
|
|
class _Event:
|
|
def __init__(self, **attrs):
|
|
self.attributes = _Attrs(**attrs)
|
|
|
|
|
|
def test_action_field_path_interpolation_resolves_context_value() -> None:
|
|
rules = [
|
|
{
|
|
"id": "flag-stale-sbom",
|
|
"condition": "context.repos.sbom_age_days > 30",
|
|
"action": {
|
|
"task_template": "Run SBOM rescan for {context.repos.repo_slug}",
|
|
"target_repo": "context.repos.repo_slug",
|
|
"priority": "medium",
|
|
"labels": ["sbom", "{context.repos.repo_slug}"],
|
|
},
|
|
}
|
|
]
|
|
|
|
specs = expand_rule_actions(
|
|
rules,
|
|
_Event(),
|
|
{"repos": {"repo_slug": "activity-core", "sbom_age_days": 45}},
|
|
)
|
|
|
|
assert specs == [
|
|
{
|
|
"title": "Run SBOM rescan for activity-core",
|
|
"description": "",
|
|
"target_repo": "activity-core",
|
|
"priority": "medium",
|
|
"labels": ["sbom", "activity-core"],
|
|
"due_in_days": None,
|
|
"source_type": "rule",
|
|
"source_id": "flag-stale-sbom",
|
|
"triggering_event_id": "",
|
|
"activity_definition_id": "",
|
|
"condition": "context.repos.sbom_age_days > 30",
|
|
}
|
|
]
|
|
|
|
|
|
def test_for_each_binds_each_list_item_before_condition_and_action_rendering() -> None:
|
|
rules = [
|
|
{
|
|
"id": "flag-stale-sbom",
|
|
"for_each": "context.repos.repos",
|
|
"bind_as": "repo",
|
|
"condition": "context.repo.sbom_age_days > 30",
|
|
"action": {
|
|
"task_template": "Run SBOM rescan for {context.repo.repo_slug}",
|
|
"target_repo": "context.repo.repo_slug",
|
|
"priority": "medium",
|
|
"labels": ["sbom", "security", "automated"],
|
|
},
|
|
}
|
|
]
|
|
context = {
|
|
"repos": {
|
|
"repos": [
|
|
{"repo_slug": "repo-a", "sbom_age_days": 60},
|
|
{"repo_slug": "repo-b", "sbom_age_days": 10},
|
|
{"repo_slug": "repo-c", "sbom_age_days": 45},
|
|
]
|
|
}
|
|
}
|
|
|
|
specs = expand_rule_actions(rules, _Event(), context)
|
|
|
|
assert [spec["target_repo"] for spec in specs] == ["repo-a", "repo-c"]
|
|
assert [spec["title"] for spec in specs] == [
|
|
"Run SBOM rescan for repo-a",
|
|
"Run SBOM rescan for repo-c",
|
|
]
|
|
|
|
|
|
def test_for_each_rejects_non_path_expression() -> None:
|
|
rules = [
|
|
{
|
|
"id": "bad",
|
|
"for_each": "__import__('os')",
|
|
"condition": "",
|
|
"action": {"task_template": "bad"},
|
|
}
|
|
]
|
|
|
|
with pytest.raises(UnsafeExpression):
|
|
expand_rule_actions(rules, _Event(), {})
|
|
|
|
|
|
def test_template_placeholder_rejects_non_scalar_values() -> None:
|
|
rules = [
|
|
{
|
|
"id": "bad",
|
|
"condition": "",
|
|
"action": {
|
|
"task_template": "Run {context.repos}",
|
|
},
|
|
}
|
|
]
|
|
|
|
with pytest.raises(UnsafeExpression):
|
|
expand_rule_actions(rules, _Event(), {"repos": [{"repo_slug": "repo-a"}]})
|