fix(tests): pytest nach Kernel Spine und Starter-Kit bereinigen
Some checks failed
Deploy Development / deploy (push) Successful in 45s
Test Suite / pytest-backend (push) Failing after 3m33s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 3s
Test Suite / compose-smoke (push) Has been skipped

list_compatible_methods nur Primary-Methoden; Agile-Graph auf Sprint-Scope; Tests ohne Starter-Action-Kollision; TenantContext-Factory fuer Kernel-Tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Lars 2026-07-26 16:15:32 +02:00
parent 75a6a63665
commit 0443b734c1
7 changed files with 99 additions and 15 deletions

View File

@ -56,7 +56,8 @@ def list_compatible_methods(
return tuple(
m
for m in _METHODS.values()
if method_compatible_with_archetype(
if m.method_role == "primary"
and method_compatible_with_archetype(
m, archetype_key, om_capabilities=om_capabilities
)
)

View File

@ -56,7 +56,6 @@ class AgileIterationStrategy:
cycle_id = str(active["id"])
from services import actions as action_service
from steering.graph.execution_engine import load_initiative_execution_graph_state
actions = action_service.list_actions_for_initiative(
tenant_id=ctx.tenant_id, initiative_id=initiative_id
@ -73,9 +72,22 @@ class AgileIterationStrategy:
if not scoped:
return []
graph_state = load_initiative_execution_graph_state(
tenant_id=ctx.tenant_id,
initiative_id=initiative_id,
scoped_ids = {str(action["id"]) for action in scoped}
from services import execution_plan as execution_plan_service
from steering.graph.execution_engine import compute_execution_graph_state
dependencies = execution_plan_service.list_dependencies_for_initiative(
tenant_id=ctx.tenant_id, initiative_id=initiative_id
)
internal_dependencies = [
dep
for dep in dependencies
if str(dep["predecessor_action_id"]) in scoped_ids
and str(dep["successor_action_id"]) in scoped_ids
]
graph_state = compute_execution_graph_state(
actions=scoped,
dependencies=internal_dependencies,
scope_roadmap_item_id=gate_scope,
)
ready = build_execution_ready_candidates(

View File

@ -74,6 +74,54 @@ def add_membership(*, tenant_id: str, user_id: str, tenant_role: str = "member")
conn.close()
def tenant_context_from_user(user: dict):
"""Build TenantContext for direct service/kernel calls in tests."""
from psycopg2.extras import RealDictCursor
from rights_registry import load_grants_for_roles
from tenant_context import TenantContext
conn = get_connection()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT t.id, t.slug, t.name, tm.tenant_role
FROM tenant_memberships tm
JOIN tenants t ON t.id = tm.tenant_id
WHERE tm.user_id = %s AND tm.tenant_id = %s
""",
(user["id"], user["tenant_id"]),
)
membership = cur.fetchone()
if not membership:
raise ValueError("Test user has no tenant membership")
finally:
conn.close()
tenant_role = membership["tenant_role"]
caps = frozenset(
load_grants_for_roles(
portal_role=user["portal_role"],
tenant_role=tenant_role,
)
)
return TenantContext(
user_id=user["id"],
email=user["email"],
display_name=user["display_name"],
portal_role=user["portal_role"],
tenant_id=str(membership["id"]),
tenant_slug=membership["slug"],
tenant_name=membership["name"],
tenant_role=tenant_role,
actor_id=user.get("actor_id"),
actor_type="human" if user.get("actor_id") else None,
session_token="test-session",
capabilities=caps,
)
def provision_user_in_tenant(
*,
tenant_role: str = "member",

View File

@ -15,6 +15,7 @@ def test_linear_critical_path_and_next_action(client):
token,
title="Neue Küche",
archetype_key="initiative.linear_project",
apply_starter_kit=False,
)
initiative_id = created.json()["id"]

View File

@ -6,6 +6,25 @@ from tests.factories import provision_user_in_tenant
from tests.test_initiatives_actions import _auth, _create_initiative, _login
_STARTER_ACTION_TITLE = "Erster Schritt — Planung konkretisieren"
def _remove_starter_action(client, token, initiative_id):
actions = client.get(
f"/api/initiatives/{initiative_id}/actions",
headers=_auth(token),
)
assert actions.status_code == 200
for action in actions.json():
if action.get("title") == _STARTER_ACTION_TITLE:
deleted = client.delete(
f"/api/actions/{action['id']}",
headers=_auth(token),
)
assert deleted.status_code == 204
return
def _gate_ids(client, token, initiative_id):
roadmap = client.get(
f"/api/initiatives/{initiative_id}/roadmap/items",
@ -30,6 +49,7 @@ def test_sequential_next_action_only_in_active_gate_horizon(client):
)
initiative_id = created.json()["id"]
g1_id, g2_id = _gate_ids(client, token, initiative_id)
_remove_starter_action(client, token, initiative_id)
off_horizon = client.post(
f"/api/initiatives/{initiative_id}/actions",
@ -77,6 +97,7 @@ def test_agile_on_linear_delegates_primary_without_active_sprint(client):
)
initiative_id = created.json()["id"]
g1_id, _ = _gate_ids(client, token, initiative_id)
_remove_starter_action(client, token, initiative_id)
action = client.post(
f"/api/initiatives/{initiative_id}/actions",
@ -115,6 +136,7 @@ def test_agile_on_linear_sprint_intersects_gate_horizon(client):
)
initiative_id = created.json()["id"]
g1_id, _ = _gate_ids(client, token, initiative_id)
_remove_starter_action(client, token, initiative_id)
cycle = client.post(
f"/api/initiatives/{initiative_id}/work-cycles",

View File

@ -124,6 +124,9 @@ def test_steering_methods_filtered_by_archetype(client):
def test_queue_pull_rejected_on_linear(client):
from tests.factories import provision_user_in_tenant
from tests.test_initiatives_actions import _auth, _create_initiative, _login
user = provision_user_in_tenant(tenant_role="member")
token = _login(client, user)
@ -145,6 +148,9 @@ def test_queue_pull_rejected_on_linear(client):
def test_operating_context_composable_modifiers_linear(client):
from tests.factories import provision_user_in_tenant
from tests.test_initiatives_actions import _auth, _create_initiative, _login
user = provision_user_in_tenant(tenant_role="member")
token = _login(client, user)

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from steering.kernel import apply_steering_event, evaluate_steering
from steering.kernel.binding import resolve_steering_binding
from tests.factories import provision_user_in_tenant
from tests.factories import provision_user_in_tenant, tenant_context_from_user
from tests.test_initiatives_actions import _auth, _create_initiative, _login
@ -20,9 +20,7 @@ def test_evaluate_steering_linear_binding(client):
)
initiative_id = created.json()["id"]
from tenant_context import TenantContext
ctx = TenantContext(tenant_id=user["tenant_id"], user_id=user["id"])
ctx = tenant_context_from_user(user)
binding = resolve_steering_binding(ctx, initiative_id)
assert binding.primary_method_key == "sequential_dependency"
assert binding.composition_modifier is None
@ -41,9 +39,7 @@ def test_evaluate_steering_returns_horizon_and_next(client):
)
initiative_id = created.json()["id"]
from tenant_context import TenantContext
ctx = TenantContext(tenant_id=user["tenant_id"], user_id=user["id"])
ctx = tenant_context_from_user(user)
evaluation = evaluate_steering(ctx, initiative_id=initiative_id)
assert evaluation.data_source == "steering_kernel_v0.1.1"
assert evaluation.lifecycle.slot_map.get("action_selection") == "active"
@ -97,9 +93,7 @@ def test_apply_steering_event_stub(client):
)
initiative_id = created.json()["id"]
from tenant_context import TenantContext
ctx = TenantContext(tenant_id=user["tenant_id"], user_id=user["id"])
ctx = tenant_context_from_user(user)
result = apply_steering_event(
ctx,
initiative_id=initiative_id,