test: harden blocked-task regression tests per review

This commit is contained in:
Abhimanyu Saharan
2026-03-04 23:01:24 +05:30
parent 130947dcea
commit 636c130a3b
2 changed files with 30 additions and 8 deletions

View File

@@ -18,12 +18,17 @@ def test_openapi_blocked_task_error_includes_code_field() -> None:
code_schema = props["code"] code_schema = props["code"]
any_of = code_schema.get("anyOf") any_of = code_schema.get("anyOf")
assert isinstance(any_of, list) and any_of if any_of:
assert isinstance(any_of, list)
has_string_branch = any(branch.get("type") == "string" for branch in any_of) has_string_branch = any(branch.get("type") == "string" for branch in any_of)
assert has_string_branch assert has_string_branch
has_null_branch = any( has_null_branch = any(
branch.get("type") == "null" or branch.get("nullable") is True for branch in any_of branch.get("type") == "null" or branch.get("nullable") is True for branch in any_of
) )
assert has_null_branch assert has_null_branch
else:
# Alternative encoding used by some schema versions for Optional[str].
assert code_schema.get("type") == "string"
assert code_schema.get("nullable") is True

View File

@@ -58,7 +58,15 @@ async def test_lead_dependency_only_update_allowed_when_task_blocked() -> None:
openclaw_session_id="agent:lead:session", openclaw_session_id="agent:lead:session",
), ),
) )
session.add(Task(id=dep_id, board_id=board_id, title="dep", description=None)) session.add(
Task(
id=dep_id,
board_id=board_id,
title="dep",
description=None,
status="inbox",
),
)
session.add( session.add(
Task( Task(
id=task_id, id=task_id,
@@ -107,6 +115,15 @@ async def test_lead_dependency_only_update_allowed_when_task_blocked() -> None:
assert reloaded is not None assert reloaded is not None
assert reloaded.status == "review" assert reloaded.status == "review"
assert reloaded.assigned_agent_id is None assert reloaded.assigned_agent_id is None
dependency_rows = (
await session.exec(
select(TaskDependency).where(
col(TaskDependency.task_id) == task_id,
col(TaskDependency.depends_on_task_id) == dep_id,
),
)
).all()
assert len(dependency_rows) == 1
finally: finally:
await engine.dispose() await engine.dispose()