generated from coulomb/repo-seed
Expand rule actions for per-repo tasks
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.
This commit is contained in:
117
tests/rules/test_actions.py
Normal file
117
tests/rules/test_actions.py
Normal file
@@ -0,0 +1,117 @@
|
||||
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"}]})
|
||||
@@ -20,7 +20,7 @@ import pytest
|
||||
from activity_core.definition_parser import parse_file
|
||||
from activity_core.issue_sink import NullSink
|
||||
from activity_core.models import EventEnvelope
|
||||
from activity_core.rules.evaluator import evaluate_condition
|
||||
from activity_core.rules.actions import expand_rule_actions
|
||||
from activity_core.rules.models import TaskRef, TaskSpec
|
||||
|
||||
_DEFINITIONS_DIR = Path(__file__).parent.parent / "activity-definitions"
|
||||
@@ -59,27 +59,24 @@ def _run_rule_pipeline(
|
||||
spawn_log: list[dict] = []
|
||||
triggering_event_id = str(uuid.uuid4())
|
||||
|
||||
for repo in repos:
|
||||
context = {"repos": repo}
|
||||
if not evaluate_condition(rule["condition"], event, context):
|
||||
continue
|
||||
|
||||
action = rule.get("action", {})
|
||||
context = {"repos": {"repos": repos}}
|
||||
for spec_dict in expand_rule_actions([rule], event, context):
|
||||
spec = TaskSpec(
|
||||
title=f"Run SBOM rescan — {repo['repo_slug']}",
|
||||
description="SBOM rescan needed — age threshold exceeded.",
|
||||
target_repo=repo["repo_slug"],
|
||||
priority=action.get("priority", "medium"),
|
||||
labels=action.get("labels", []),
|
||||
title=spec_dict["title"],
|
||||
description=spec_dict["description"],
|
||||
target_repo=spec_dict["target_repo"],
|
||||
priority=spec_dict["priority"],
|
||||
labels=spec_dict["labels"],
|
||||
due_in_days=spec_dict["due_in_days"],
|
||||
source_type="rule",
|
||||
source_id=rule["id"],
|
||||
source_id=spec_dict["source_id"],
|
||||
triggering_event_id=triggering_event_id,
|
||||
)
|
||||
ref = sink.emit(spec)
|
||||
task_refs.append(ref)
|
||||
spawn_log.append({
|
||||
"source_id": rule["id"],
|
||||
"condition_matched": rule["condition"],
|
||||
"source_id": spec_dict["source_id"],
|
||||
"condition_matched": spec_dict["condition"],
|
||||
"triggering_event_id": triggering_event_id,
|
||||
"task_ref": ref.external_id,
|
||||
})
|
||||
@@ -121,7 +118,7 @@ def test_pipeline_emits_one_task_for_stale_repo_only():
|
||||
assert len(spawn_log) == 1
|
||||
entry = spawn_log[0]
|
||||
assert entry["source_id"] == "flag-stale-sbom"
|
||||
assert entry["condition_matched"] == "context.repos.sbom_age_days > 30"
|
||||
assert entry["condition_matched"] == "context.repo.sbom_age_days > 30"
|
||||
assert entry["triggering_event_id"] == spawn_log[0]["triggering_event_id"]
|
||||
|
||||
|
||||
|
||||
40
tests/test_rule_evaluation_activity.py
Normal file
40
tests/test_rule_evaluation_activity.py
Normal file
@@ -0,0 +1,40 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from activity_core import activities
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_evaluate_rules_returns_interpolated_task_specs() -> None:
|
||||
result = await activities.evaluate_rules({
|
||||
"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", "{context.repo.repo_slug}"],
|
||||
},
|
||||
}
|
||||
],
|
||||
"event": {},
|
||||
"context": {
|
||||
"repos": {
|
||||
"repos": [
|
||||
{"repo_slug": "fresh-repo", "sbom_age_days": 5},
|
||||
{"repo_slug": "stale-repo", "sbom_age_days": 40},
|
||||
]
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["title"] == "Run SBOM rescan for stale-repo"
|
||||
assert result[0]["target_repo"] == "stale-repo"
|
||||
assert result[0]["labels"] == ["sbom", "stale-repo"]
|
||||
assert result[0]["condition"] == "context.repo.sbom_age_days > 30"
|
||||
Reference in New Issue
Block a user