zhou/pyflowx
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases 6 Wiki Activity

style: 统一调整代码格式，将单行列表展开为多行缩进格式

Browse Source
This commit is contained in:
gooker_young
2026-07-04 19:49:10 +08:00
parent 3f966a230e
commit 52e20e3f93
15 changed files with 1097 additions and 829 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pyflowx/examples/async_aggregation.py
+8 -6
View File
@@ -31,12 +31,14 @@ def aggregate(ctx: px.Context) -> dict[str, Any]:
def main() -> None:
graph = px.Graph.from_specs([
# Static positional args parameterise the same function twice.
px.TaskSpec("fetch_user", fetch_user, args=(1,)),
px.TaskSpec("fetch_posts", fetch_posts, args=(1,)),
px.TaskSpec("aggregate", aggregate, depends_on=("fetch_user", "fetch_posts")),
])
graph = px.Graph.from_specs(
[
# Static positional args parameterise the same function twice.
px.TaskSpec("fetch_user", fetch_user, args=(1,)),
px.TaskSpec("fetch_posts", fetch_posts, args=(1,)),
px.TaskSpec("aggregate", aggregate, depends_on=("fetch_user", "fetch_posts")),
]
)
print("=== Dry run ===")
_ = px.run(graph, strategy="async", dry_run=True)
src/pyflowx/examples/etl_pipeline.py
+15 -13
View File
@@ -46,19 +46,21 @@ def load(transform: list[dict[str, Any]]) -> int:
def main() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("extract_customers", extract_customers, tags=("extract",)),
px.TaskSpec("extract_orders", extract_orders, tags=("extract",)),
px.TaskSpec(
"transform",
transform,
depends_on=("extract_customers", "extract_orders"),
tags=("transform",),
),
px.TaskSpec(
"load", load, depends_on=("transform",), retry=px.RetryPolicy(max_attempts=1, delay=1.0), tags=("load",)
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("extract_customers", extract_customers, tags=("extract",)),
px.TaskSpec("extract_orders", extract_orders, tags=("extract",)),
px.TaskSpec(
"transform",
transform,
depends_on=("extract_customers", "extract_orders"),
tags=("transform",),
),
px.TaskSpec(
"load", load, depends_on=("transform",), retry=px.RetryPolicy(max_attempts=1, delay=1.0), tags=("load",)
),
]
)
print("=== Execution plan ===")
print(graph.describe())
src/pyflowx/examples/parallel_run.py
+7 -5
View File
@@ -29,11 +29,13 @@ def merge(fetch_a: str, fetch_b: str) -> str:
def main() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("fetch_a", fetch_a),
px.TaskSpec("fetch_b", fetch_b),
px.TaskSpec("merge", merge, depends_on=("fetch_a", "fetch_b")),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fetch_a", fetch_a),
px.TaskSpec("fetch_b", fetch_b),
px.TaskSpec("merge", merge, depends_on=("fetch_a", "fetch_b")),
]
)
print("=== Mermaid diagram ===")
print(graph.to_mermaid("LR"))
tests/cli/test_emlmanager.py
+210 -182
View File
@@ -125,19 +125,21 @@ class TestEmailDatabase:
# Insert test emails
for i in range(5):
db.insert_email({
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
results = db.search_emails(limit=3)
assert len(results) == 3
@@ -148,33 +150,37 @@ class TestEmailDatabase:
db_path = tmp_path / "test.db"
db = emlmanager.EmailDatabase(db_path)
db.insert_email({
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Important Meeting",
"sender": "sender1@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Meeting body",
"body_html": "<p>Meeting body</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Important Meeting",
"sender": "sender1@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Meeting body",
"body_html": "<p>Meeting body</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
db.insert_email({
"file_path": "/test/path2.eml",
"file_hash": "hash2",
"subject": "Casual Chat",
"sender": "sender2@example.com",
"recipients": "recipient@example.com",
"date": "Tue, 2 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-02T12:00:00",
"body_text": "Chat body",
"body_html": "<p>Chat body</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path2.eml",
"file_hash": "hash2",
"subject": "Casual Chat",
"sender": "sender2@example.com",
"recipients": "recipient@example.com",
"date": "Tue, 2 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-02T12:00:00",
"body_text": "Chat body",
"body_html": "<p>Chat body</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
results = db.search_emails(keyword="Meeting", field="subject")
assert len(results) == 1
@@ -186,33 +192,37 @@ class TestEmailDatabase:
db_path = tmp_path / "test.db"
db = emlmanager.EmailDatabase(db_path)
db.insert_email({
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Test",
"sender": "alice@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Body",
"body_html": "<p>Body</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Test",
"sender": "alice@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Body",
"body_html": "<p>Body</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
db.insert_email({
"file_path": "/test/path2.eml",
"file_hash": "hash2",
"subject": "Test",
"sender": "bob@example.com",
"recipients": "recipient@example.com",
"date": "Tue, 2 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-02T12:00:00",
"body_text": "Body",
"body_html": "<p>Body</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path2.eml",
"file_hash": "hash2",
"subject": "Test",
"sender": "bob@example.com",
"recipients": "recipient@example.com",
"date": "Tue, 2 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-02T12:00:00",
"body_text": "Body",
"body_html": "<p>Body</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
results = db.search_emails(keyword="alice", field="sender")
assert len(results) == 1
@@ -224,19 +234,21 @@ class TestEmailDatabase:
db_path = tmp_path / "test.db"
db = emlmanager.EmailDatabase(db_path)
db.insert_email({
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Project Update",
"sender": "manager@example.com",
"recipients": "team@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Please review the quarterly report",
"body_html": "<p>Please review the quarterly report</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Project Update",
"sender": "manager@example.com",
"recipients": "team@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Please review the quarterly report",
"body_html": "<p>Please review the quarterly report</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
# Search for keyword in subject
results = db.search_emails(keyword="Project", field="all")
@@ -253,47 +265,53 @@ class TestEmailDatabase:
db = emlmanager.EmailDatabase(db_path)
# Insert emails with same subject (different prefixes)
db.insert_email({
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Meeting Tomorrow",
"sender": "sender1@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Body 1",
"body_html": "<p>Body 1</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path1.eml",
"file_hash": "hash1",
"subject": "Meeting Tomorrow",
"sender": "sender1@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Body 1",
"body_html": "<p>Body 1</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
db.insert_email({
"file_path": "/test/path2.eml",
"file_hash": "hash2",
"subject": "Re: Meeting Tomorrow",
"sender": "sender2@example.com",
"recipients": "recipient@example.com",
"date": "Tue, 2 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-02T12:00:00",
"body_text": "Body 2",
"body_html": "<p>Body 2</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path2.eml",
"file_hash": "hash2",
"subject": "Re: Meeting Tomorrow",
"sender": "sender2@example.com",
"recipients": "recipient@example.com",
"date": "Tue, 2 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-02T12:00:00",
"body_text": "Body 2",
"body_html": "<p>Body 2</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
db.insert_email({
"file_path": "/test/path3.eml",
"file_hash": "hash3",
"subject": "Different Topic",
"sender": "sender3@example.com",
"recipients": "recipient@example.com",
"date": "Wed, 3 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-03T12:00:00",
"body_text": "Body 3",
"body_html": "<p>Body 3</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path3.eml",
"file_hash": "hash3",
"subject": "Different Topic",
"sender": "sender3@example.com",
"recipients": "recipient@example.com",
"date": "Wed, 3 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-03T12:00:00",
"body_text": "Body 3",
"body_html": "<p>Body 3</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
grouped = db.get_grouped_emails()
# Should have 2 groups: "Meeting Tomorrow" and "Different Topic"
@@ -322,19 +340,21 @@ class TestEmailDatabase:
assert db.get_email_count() == 0
for i in range(3):
db.insert_email({
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
assert db.get_email_count() == 3
db.close()
@@ -346,19 +366,21 @@ class TestEmailDatabase:
# Insert some emails
for i in range(3):
db.insert_email({
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
assert db.get_email_count() == 3
@@ -687,19 +709,21 @@ class TestEmlManagerHandler:
# Insert some emails
for i in range(3):
db.insert_email({
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": f"/test/path{i}.eml",
"file_hash": f"hash{i}",
"subject": f"Subject {i}",
"sender": f"sender{i}@example.com",
"recipients": "recipient@example.com",
"date": f"Mon, {i + 1} Jan 2024 12:00:00 +0000",
"date_parsed": f"2024-01-0{i + 1}T12:00:00",
"body_text": f"Body {i}",
"body_html": f"<p>Body {i}</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
# Create a mock handler instance without calling __init__
handler = Mock(spec=emlmanager.EmlManagerHandler)
@@ -721,19 +745,21 @@ class TestEmlManagerHandler:
db = emlmanager.EmailDatabase(db_path)
# Insert test email
db.insert_email({
"file_path": "/test/path.eml",
"file_hash": "hash",
"subject": "Test Subject",
"sender": "sender@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Test body",
"body_html": "<p>Test body</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path.eml",
"file_hash": "hash",
"subject": "Test Subject",
"sender": "sender@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Test body",
"body_html": "<p>Test body</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
# Create a mock handler instance without calling __init__
handler = Mock(spec=emlmanager.EmlManagerHandler)
@@ -756,19 +782,21 @@ class TestEmlManagerHandler:
db = emlmanager.EmailDatabase(db_path)
# Insert test email
db.insert_email({
"file_path": "/test/path.eml",
"file_hash": "hash",
"subject": "Test Subject",
"sender": "sender@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Test body",
"body_html": "<p>Test body</p>",
"has_attachments": 0,
"file_size": 1024,
})
db.insert_email(
{
"file_path": "/test/path.eml",
"file_hash": "hash",
"subject": "Test Subject",
"sender": "sender@example.com",
"recipients": "recipient@example.com",
"date": "Mon, 1 Jan 2024 12:00:00 +0000",
"date_parsed": "2024-01-01T12:00:00",
"body_text": "Test body",
"body_html": "<p>Test body</p>",
"has_attachments": 0,
"file_size": 1024,
}
)
assert db.get_email_count() == 1
tests/cli/test_profiler.py
+6 -4
View File
@@ -56,10 +56,12 @@ def _build_simple_profile() -> ProfileReport:
report = px.RunReport()
report.results["a"] = _result("a", start, 1.0)
report.results["b"] = _result("b", start + timedelta(seconds=1), 2.0)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b", deps=("a",)),
]
)
return ProfileReport.from_report(report, graph)
tests/test_advanced_features.py
+273 -197
View File
@@ -82,9 +82,11 @@ class TestRetryPolicy:
raise RuntimeError("not yet")
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("flaky", flaky, retry=RetryPolicy(max_attempts=3)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("flaky", flaky, retry=RetryPolicy(max_attempts=3)),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report["flaky"] == "ok"
@@ -95,9 +97,11 @@ class TestRetryPolicy:
def always_fail() -> None:
raise RuntimeError("nope")
graph = px.Graph.from_specs([
px.TaskSpec("f", always_fail, retry=RetryPolicy(max_attempts=3)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("f", always_fail, retry=RetryPolicy(max_attempts=3)),
]
)
with pytest.raises(px.TaskFailedError) as exc_info:
px.run(graph, strategy="sequential")
assert exc_info.value.attempts == 3
@@ -111,13 +115,15 @@ class TestRetryPolicy:
raise KeyError("not retried")
# retry_on=(ValueError,) -> KeyError 不应被重试
graph = px.Graph.from_specs([
px.TaskSpec(
"f",
fail_with_keyerror,
retry=RetryPolicy(max_attempts=3, retry_on=(ValueError,)),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"f",
fail_with_keyerror,
retry=RetryPolicy(max_attempts=3, retry_on=(ValueError,)),
),
]
)
with pytest.raises(px.TaskFailedError) as exc_info:
px.run(graph, strategy="sequential")
# KeyError 不在 retry_on 中，应只尝试 1 次
@@ -136,13 +142,15 @@ class TestRetryPolicy:
raise RuntimeError("not yet")
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec(
"flaky",
flaky,
retry=RetryPolicy(max_attempts=3, delay=0.05, backoff=2.0),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"flaky",
flaky,
retry=RetryPolicy(max_attempts=3, delay=0.05, backoff=2.0),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
# 第 2 次重试应在 delay=0.05 后，第 3 次应在 0.05*2=0.10 后
@@ -161,9 +169,11 @@ class TestRetryPolicy:
raise RuntimeError("not yet")
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("flaky", flaky, retry=RetryPolicy(max_attempts=3)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("flaky", flaky, retry=RetryPolicy(max_attempts=3)),
]
)
report = px.run(graph, strategy="async")
assert report.success
assert report["flaky"] == "ok"
@@ -187,9 +197,11 @@ class TestTaskHooks:
return "ok"
hooks = TaskHooks(pre_run=pre_run)
graph = px.Graph.from_specs([
px.TaskSpec("t", fn, hooks=hooks),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("t", fn, hooks=hooks),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert events == ["pre:t", "run"]
@@ -205,9 +217,11 @@ class TestTaskHooks:
return 42
hooks = TaskHooks(post_run=post_run)
graph = px.Graph.from_specs([
px.TaskSpec("t", fn, hooks=hooks),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("t", fn, hooks=hooks),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert captured == {"name": "t", "result": 42}
@@ -223,9 +237,11 @@ class TestTaskHooks:
raise ValueError("boom")
hooks = TaskHooks(on_failure=on_failure)
graph = px.Graph.from_specs([
px.TaskSpec("t", fn, hooks=hooks, continue_on_error=True),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("t", fn, hooks=hooks, continue_on_error=True),
]
)
report = px.run(graph, strategy="sequential")
# continue_on_error=True -> 报告成功但任务失败
assert report.success
@@ -242,14 +258,16 @@ class TestTaskHooks:
events.append("post")
hooks = TaskHooks(pre_run=pre_run, post_run=post_run)
graph = px.Graph.from_specs([
px.TaskSpec(
"t",
fn=lambda: "ok",
hooks=hooks,
conditions=(lambda _ctx: False,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"t",
fn=lambda: "ok",
hooks=hooks,
conditions=(lambda _ctx: False,),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report.result_of("t").status == TaskStatus.SKIPPED
@@ -266,9 +284,11 @@ class TestTaskHooks:
return "ok"
hooks = TaskHooks(pre_run=pre_run)
graph = px.Graph.from_specs([
px.TaskSpec("t", fn, hooks=hooks),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("t", fn, hooks=hooks),
]
)
report = px.run(graph, strategy="async")
assert report.success
assert events == ["pre:t", "run"]
@@ -414,10 +434,12 @@ class TestSoftDependencies:
order.append("fast")
return f"after-{slow}"
graph = px.Graph.from_specs([
px.TaskSpec("slow", slow),
px.TaskSpec("fast", fast, soft_depends_on=("slow",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("slow", slow),
px.TaskSpec("fast", fast, soft_depends_on=("slow",)),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
# soft 依赖应等待 slow 完成后再执行 fast
@@ -433,15 +455,17 @@ class TestSoftDependencies:
def downstream(fail: str = "default") -> str:
return f"got:{fail}"
graph = px.Graph.from_specs([
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec(
"downstream",
downstream,
soft_depends_on=("fail",),
continue_on_error=True,
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec(
"downstream",
downstream,
soft_depends_on=("fail",),
continue_on_error=True,
),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
# fail 失败但下游仍执行（使用默认值）
@@ -450,9 +474,11 @@ class TestSoftDependencies:
def test_soft_dependency_validation_unknown_dep(self) -> None:
with pytest.raises(px.MissingDependencyError):
px.Graph.from_specs([
px.TaskSpec("a", lambda: "ok", soft_depends_on=("missing",)),
])
px.Graph.from_specs(
[
px.TaskSpec("a", lambda: "ok", soft_depends_on=("missing",)),
]
)
def test_soft_and_hard_dependency_combined(self) -> None:
order: list[str] = []
@@ -469,11 +495,13 @@ class TestSoftDependencies:
order.append("c")
return f"c-{b}"
graph = px.Graph.from_specs([
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
px.TaskSpec("c", c, depends_on=("b",), soft_depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
px.TaskSpec("c", c, depends_on=("b",), soft_depends_on=("a",)),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert order == ["a", "b", "c"]
@@ -495,11 +523,13 @@ class TestDependencyDrivenScheduling:
def c(b: int) -> int:
return b + 1
graph = px.Graph.from_specs([
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
px.TaskSpec("c", c, depends_on=("b",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
px.TaskSpec("c", c, depends_on=("b",)),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report["a"] == 1
@@ -514,10 +544,12 @@ class TestDependencyDrivenScheduling:
async def b(a: str) -> str:
return f"b-{a}"
graph = px.Graph.from_specs([
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report["b"] == "b-a"
@@ -537,12 +569,14 @@ class TestDependencyDrivenScheduling:
def d(b: int, c: int) -> int:
return b + c
graph = px.Graph.from_specs([
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
px.TaskSpec("c", c, depends_on=("a",)),
px.TaskSpec("d", d, depends_on=("b", "c")),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
px.TaskSpec("c", c, depends_on=("a",)),
px.TaskSpec("d", d, depends_on=("b", "c")),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report["a"] == 10
@@ -574,11 +608,13 @@ class TestConcurrencyLimits:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make_fn(1), concurrency_key="db"),
px.TaskSpec("b", make_fn(2), concurrency_key="db"),
px.TaskSpec("c", make_fn(3), concurrency_key="db"),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make_fn(1), concurrency_key="db"),
px.TaskSpec("b", make_fn(2), concurrency_key="db"),
px.TaskSpec("c", make_fn(3), concurrency_key="db"),
]
)
report = px.run(
graph,
strategy="dependency",
@@ -604,10 +640,12 @@ class TestConcurrencyLimits:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make_fn("a"), concurrency_key="db1"),
px.TaskSpec("b", make_fn("b"), concurrency_key="db2"),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make_fn("a"), concurrency_key="db1"),
px.TaskSpec("b", make_fn("b"), concurrency_key="db2"),
]
)
report = px.run(
graph,
strategy="dependency",
@@ -633,12 +671,14 @@ class TestConcurrencyLimits:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make_fn(1), concurrency_key="pool"),
px.TaskSpec("b", make_fn(2), concurrency_key="pool"),
px.TaskSpec("c", make_fn(3), concurrency_key="pool"),
px.TaskSpec("d", make_fn(4), concurrency_key="pool"),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make_fn(1), concurrency_key="pool"),
px.TaskSpec("b", make_fn(2), concurrency_key="pool"),
px.TaskSpec("c", make_fn(3), concurrency_key="pool"),
px.TaskSpec("d", make_fn(4), concurrency_key="pool"),
]
)
report = px.run(
graph,
strategy="dependency",
@@ -666,11 +706,13 @@ class TestPriority:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("low", make_fn("low"), priority=1),
px.TaskSpec("high", make_fn("high"), priority=10),
px.TaskSpec("mid", make_fn("mid"), priority=5),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("low", make_fn("low"), priority=1),
px.TaskSpec("high", make_fn("high"), priority=10),
px.TaskSpec("mid", make_fn("mid"), priority=5),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
# 高优先级先执行
@@ -696,10 +738,12 @@ class TestContinueOnError:
def downstream() -> str:
return "ran"
graph = px.Graph.from_specs([
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec("downstream", downstream, depends_on=("fail",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec("downstream", downstream, depends_on=("fail",)),
]
)
report = px.run(graph, strategy="sequential")
# continue_on_error 使整体报告成功（不抛异常）
assert report.success
@@ -716,10 +760,12 @@ class TestContinueOnError:
def downstream() -> str:
return "ran"
graph = px.Graph.from_specs([
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec("downstream", downstream, soft_depends_on=("fail",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec("downstream", downstream, soft_depends_on=("fail",)),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report.result_of("fail").status == TaskStatus.FAILED
@@ -734,10 +780,12 @@ class TestContinueOnError:
def other() -> str:
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec("other", other),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", fail, continue_on_error=True),
px.TaskSpec("other", other),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report.result_of("fail").status == TaskStatus.FAILED
@@ -750,10 +798,12 @@ class TestContinueOnError:
def other() -> str:
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("fail", fail),
px.TaskSpec("other", other),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", fail),
px.TaskSpec("other", other),
]
)
with pytest.raises(px.TaskFailedError):
px.run(graph, strategy="sequential")
@@ -854,9 +904,11 @@ class TestCompose:
g_extract = px.Graph.from_specs([px.TaskSpec("extract", extract)])
# transform 图：通过 _pending_refs 引用 "extract" 命令
# transform 自身不声明 depends_on，由 compose 展开时自动连接
g_transform = px.Graph.from_specs([
px.TaskSpec("transform", transform),
])
g_transform = px.Graph.from_specs(
[
px.TaskSpec("transform", transform),
]
)
g_transform._pending_refs = ["extract"]
resolved = px.compose({"extract": g_extract, "transform": g_transform})
@@ -943,18 +995,22 @@ class TestCacheKey:
return key
graph1 = px.Graph.from_specs([
px.TaskSpec("t", expensive, args=(5,), cache_key=make_cache_key(5)),
])
graph1 = px.Graph.from_specs(
[
px.TaskSpec("t", expensive, args=(5,), cache_key=make_cache_key(5)),
]
)
report1 = px.run(graph1, strategy="sequential", state=backend)
assert report1.success
assert report1["t"] == 10
assert calls["n"] == 1
# 第二次运行相同输入应命中缓存
graph2 = px.Graph.from_specs([
px.TaskSpec("t", expensive, args=(5,), cache_key=make_cache_key(5)),
])
graph2 = px.Graph.from_specs(
[
px.TaskSpec("t", expensive, args=(5,), cache_key=make_cache_key(5)),
]
)
report2 = px.run(graph2, strategy="sequential", state=backend)
assert report2.success
assert report2["t"] == 10
@@ -976,16 +1032,20 @@ class TestCacheKey:
return key
graph1 = px.Graph.from_specs([
px.TaskSpec("t", expensive, args=(5,), cache_key=make_cache_key(5)),
])
graph1 = px.Graph.from_specs(
[
px.TaskSpec("t", expensive, args=(5,), cache_key=make_cache_key(5)),
]
)
px.run(graph1, strategy="sequential", state=backend)
assert calls["n"] == 1
# 不同输入应 miss
graph2 = px.Graph.from_specs([
px.TaskSpec("t", expensive, args=(7,), cache_key=make_cache_key(7)),
])
graph2 = px.Graph.from_specs(
[
px.TaskSpec("t", expensive, args=(7,), cache_key=make_cache_key(7)),
]
)
px.run(graph2, strategy="sequential", state=backend)
assert calls["n"] == 2
@@ -997,13 +1057,15 @@ class TestEnvAndCwd:
"""测试环境变量与工作目录隔离。"""
def test_env_override_for_cmd(self) -> None:
graph = px.Graph.from_specs([
px.TaskSpec(
"print_var",
cmd=[sys.executable, "-c", "import os; print(os.environ.get('PYFLOWX_TEST_VAR', 'unset'))"],
env={"PYFLOWX_TEST_VAR": "isolated"},
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"print_var",
cmd=[sys.executable, "-c", "import os; print(os.environ.get('PYFLOWX_TEST_VAR', 'unset'))"],
env={"PYFLOWX_TEST_VAR": "isolated"},
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -1011,13 +1073,15 @@ class TestEnvAndCwd:
# 在 tmp_path 下创建标记文件
marker = tmp_path / "marker.txt"
marker.write_text("found")
graph = px.Graph.from_specs([
px.TaskSpec(
"check_cwd",
cmd=["ls", "marker.txt"],
cwd=tmp_path,
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"check_cwd",
cmd=["ls", "marker.txt"],
cwd=tmp_path,
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -1027,13 +1091,15 @@ class TestEnvAndCwd:
def check_env() -> str:
return os.environ.get("PYFLOWX_LEAK_TEST", "not-set")
graph = px.Graph.from_specs([
px.TaskSpec(
"t",
check_env,
env={"PYFLOWX_LEAK_TEST": "leaked"},
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"t",
check_env,
env={"PYFLOWX_LEAK_TEST": "leaked"},
),
]
)
# fn 任务的环境变量隔离仅在 cmd 任务生效，fn 共享进程环境
# 这里验证 fn 任务不修改外层环境
report = px.run(graph, strategy="sequential")
@@ -1059,21 +1125,23 @@ class TestContextAwareConditions:
def path_b(decide: str = "") -> str:
return f"ran-b:{decide}"
graph = px.Graph.from_specs([
px.TaskSpec("decide", decide),
px.TaskSpec(
"path_a",
path_a,
depends_on=("decide",),
conditions=(BuiltinConditions.DEP_EQUALS("decide", "path_a"),),
),
px.TaskSpec(
"path_b",
path_b,
depends_on=("decide",),
conditions=(BuiltinConditions.DEP_EQUALS("decide", "path_b"),),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("decide", decide),
px.TaskSpec(
"path_a",
path_a,
depends_on=("decide",),
conditions=(BuiltinConditions.DEP_EQUALS("decide", "path_a"),),
),
px.TaskSpec(
"path_b",
path_b,
depends_on=("decide",),
conditions=(BuiltinConditions.DEP_EQUALS("decide", "path_b"),),
),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report.result_of("path_a").status == TaskStatus.SKIPPED
@@ -1087,15 +1155,17 @@ class TestContextAwareConditions:
def only_if_nonempty(source: list[int]) -> str:
return f"has-{len(source)}"
graph = px.Graph.from_specs([
px.TaskSpec("source", source),
px.TaskSpec(
"only_if_nonempty",
only_if_nonempty,
depends_on=("source",),
conditions=(BuiltinConditions.DEP_TRUTHY("source"),),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("source", source),
px.TaskSpec(
"only_if_nonempty",
only_if_nonempty,
depends_on=("source",),
conditions=(BuiltinConditions.DEP_TRUTHY("source"),),
),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report["only_if_nonempty"] == "has-3"
@@ -1107,15 +1177,17 @@ class TestContextAwareConditions:
def only_if_nonempty(source: list[int]) -> str:
return "should-not-run"
graph = px.Graph.from_specs([
px.TaskSpec("source", source),
px.TaskSpec(
"only_if_nonempty",
only_if_nonempty,
depends_on=("source",),
conditions=(BuiltinConditions.DEP_TRUTHY("source"),),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("source", source),
px.TaskSpec(
"only_if_nonempty",
only_if_nonempty,
depends_on=("source",),
conditions=(BuiltinConditions.DEP_TRUTHY("source"),),
),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report.result_of("only_if_nonempty").status == TaskStatus.SKIPPED
@@ -1127,15 +1199,17 @@ class TestContextAwareConditions:
def downstream(source: int) -> str:
return f"got-{source}"
graph = px.Graph.from_specs([
px.TaskSpec("source", source),
px.TaskSpec(
"downstream",
downstream,
depends_on=("source",),
conditions=(BuiltinConditions.DEP_MATCHES("source", lambda v: v > 10),),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("source", source),
px.TaskSpec(
"downstream",
downstream,
depends_on=("source",),
conditions=(BuiltinConditions.DEP_MATCHES("source", lambda v: v > 10),),
),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report["downstream"] == "got-42"
@@ -1165,10 +1239,12 @@ class TestPerTaskStrategy:
await asyncio.sleep(0.01)
return f"async-{sync}"
graph = px.Graph.from_specs([
px.TaskSpec("sync", sync_fn),
px.TaskSpec("async", async_fn, depends_on=("sync",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("sync", sync_fn),
px.TaskSpec("async", async_fn, depends_on=("sync",)),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report["async"] == "async-sync"
tests/test_chain_dsl.py
+1
View File
@@ -68,6 +68,7 @@ def test_chain_execution_order() -> None:
def fn() -> str:
order.append(name)
return name
return fn
a = TaskSpec("a", make("a"))
tests/test_command_refs.py
+3 -3
View File
@@ -93,7 +93,7 @@ class TestCommandReferences:
with pytest.raises(ValueError, match="循环引用"):
px.CliRunner(
strategy="sequential",
aliases={
aliases={
"cmd1": px.Graph.from_specs(["cmd1", task1]),
},
)
@@ -105,7 +105,7 @@ class TestCommandReferences:
with pytest.raises(ValueError, match="引用的命令 'invalid' 不存在"):
px.CliRunner(
strategy="sequential",
aliases={
aliases={
"cmd1": px.Graph.from_specs(["invalid", task1]),
},
)
@@ -117,7 +117,7 @@ class TestCommandReferences:
with pytest.raises(ValueError, match="任务 'invalid' 不存在于命令 'cmd1' 中"):
px.CliRunner(
strategy="sequential",
aliases={
aliases={
"cmd1": px.Graph.from_specs([task1]),
"cmd2": px.Graph.from_specs(["cmd1.invalid"]),
},
tests/test_executors.py
+118 -78
View File
@@ -27,10 +27,12 @@ def test_sequential_basic() -> None:
def double(extract: list[int]) -> list[int]:
return [x * 2 for x in extract]
graph = px.Graph.from_specs([
px.TaskSpec("extract", extract),
px.TaskSpec("double", double, depends_on=("extract",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("extract", extract),
px.TaskSpec("double", double, depends_on=("extract",)),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report["extract"] == [1, 2, 3]
@@ -47,12 +49,14 @@ def test_sequential_diamond() -> None:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
px.TaskSpec("c", make("c"), depends_on=("a",)),
px.TaskSpec("d", make("d"), depends_on=("b", "c")),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
px.TaskSpec("c", make("c"), depends_on=("a",)),
px.TaskSpec("d", make("d"), depends_on=("b", "c")),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report["d"] == "d"
@@ -66,10 +70,12 @@ def test_failure_propagates() -> None:
def downstream(_boom: None) -> int:
return 1
graph = px.Graph.from_specs([
px.TaskSpec("boom", boom),
px.TaskSpec("downstream", downstream, depends_on=("boom",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("boom", boom),
px.TaskSpec("downstream", downstream, depends_on=("boom",)),
]
)
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
assert exc_info.value.task == "boom"
@@ -85,9 +91,11 @@ def test_retries_then_succeeds() -> None:
raise RuntimeError("not yet")
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("flaky", flaky, retry=px.RetryPolicy(max_attempts=3)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("flaky", flaky, retry=px.RetryPolicy(max_attempts=3)),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report["flaky"] == "ok"
@@ -105,9 +113,11 @@ def test_retries_with_delay() -> None:
raise RuntimeError("not yet")
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("flaky", flaky, retry=px.RetryPolicy(max_attempts=2, delay=0.1)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("flaky", flaky, retry=px.RetryPolicy(max_attempts=2, delay=0.1)),
]
)
report = px.run(graph, strategy="sequential")
elapsed = time.time() - start_time
assert report.success
@@ -122,9 +132,11 @@ def test_timeout_then_retry_async(caplog: pytest.LogCaptureFixture) -> None:
await asyncio.sleep(10) # 会触发超时
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("slow", slow_task, timeout=0.2, retry=px.RetryPolicy(max_attempts=2)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("slow", slow_task, timeout=0.2, retry=px.RetryPolicy(max_attempts=2)),
]
)
with caplog.at_level(logging.WARNING, logger="pyflowx"):
with pytest.raises(px.TaskFailedError) as exc_info:
_ = px.run(graph, strategy="async")
@@ -138,9 +150,11 @@ def test_retries_exhausted() -> None:
def always_fail() -> None:
raise RuntimeError("nope")
graph = px.Graph.from_specs([
px.TaskSpec("f", always_fail, retry=px.RetryPolicy(max_attempts=3)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("f", always_fail, retry=px.RetryPolicy(max_attempts=3)),
]
)
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
assert exc_info.value.attempts == 3
@@ -155,11 +169,13 @@ def test_threaded_parallelism() -> None:
time.sleep(0.3)
return "done"
graph = px.Graph.from_specs([
px.TaskSpec("a", slow),
px.TaskSpec("b", slow),
px.TaskSpec("c", slow),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", slow),
px.TaskSpec("b", slow),
px.TaskSpec("c", slow),
]
)
start = time.time()
report = px.run(graph, strategy="thread", max_workers=3)
elapsed = time.time() - start
@@ -182,11 +198,13 @@ def test_threaded_layer_barrier() -> None:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b")),
px.TaskSpec("c", make("c"), depends_on=("a", "b")),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b")),
px.TaskSpec("c", make("c"), depends_on=("a", "b")),
]
)
report = px.run(graph, strategy="thread", max_workers=2)
assert report.success
# c must finish after both a and b.
@@ -205,10 +223,12 @@ def test_async_basic() -> None:
async def transform(fetch: int) -> int:
return fetch * 2
graph = px.Graph.from_specs([
px.TaskSpec("fetch", fetch),
px.TaskSpec("transform", transform, depends_on=("fetch",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fetch", fetch),
px.TaskSpec("transform", transform, depends_on=("fetch",)),
]
)
report = px.run(graph, strategy="async")
assert report.success
assert report["transform"] == 84
@@ -237,10 +257,12 @@ def test_async_mixed_sync_and_async() -> None:
await asyncio.sleep(0.01)
return sync_task + 5
graph = px.Graph.from_specs([
px.TaskSpec("sync_task", sync_task),
px.TaskSpec("async_task", async_task, depends_on=("sync_task",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("sync_task", sync_task),
px.TaskSpec("async_task", async_task, depends_on=("sync_task",)),
]
)
report = px.run(graph, strategy="async")
assert report.success
assert report["async_task"] == 15
@@ -288,10 +310,12 @@ def test_memory_backend_resume() -> None:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
]
)
backend = MemoryBackend()
_ = px.run(graph, strategy="sequential", state=backend)
assert runs == ["a", "b"]
@@ -377,9 +401,11 @@ def test_async_timeout_retry_then_succeed() -> None:
await asyncio.sleep(10) # 触发超时
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("a", flaky, retry=px.RetryPolicy(max_attempts=3), timeout=0.05),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", flaky, retry=px.RetryPolicy(max_attempts=3), timeout=0.05),
]
)
report = px.run(graph, strategy="async")
assert report.success
assert report["a"] == "ok"
@@ -396,9 +422,11 @@ def test_async_failure_retry_branch(caplog: pytest.LogCaptureFixture) -> None:
raise RuntimeError("not yet")
return "ok"
graph = px.Graph.from_specs([
px.TaskSpec("a", flaky, retry=px.RetryPolicy(max_attempts=3)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", flaky, retry=px.RetryPolicy(max_attempts=3)),
]
)
with caplog.at_level("WARNING", logger="pyflowx"):
report = px.run(graph, strategy="async")
assert report.success
@@ -421,10 +449,12 @@ def test_threaded_skips_cached_tasks() -> None:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
]
)
backend = px.MemoryBackend()
# 第一次运行填充缓存
_ = px.run(graph, strategy="thread", max_workers=2, state=backend)
@@ -464,10 +494,12 @@ def test_async_skips_cached_tasks() -> None:
runs.append("b")
return a + "b"
graph = px.Graph.from_specs([
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
]
)
backend = px.MemoryBackend()
_ = px.run(graph, strategy="async", state=backend)
assert runs == ["a", "b"]
@@ -543,10 +575,12 @@ def test_downstream_skipped_when_upstream_skipped_sequential() -> None:
def downstream(upstream: str) -> str:
return upstream + "_processed"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", cmd=["echo", "hello"], conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("upstream", cmd=["echo", "hello"], conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SKIPPED
@@ -560,10 +594,12 @@ def test_downstream_skipped_when_upstream_skipped_thread() -> None:
def downstream(upstream: str) -> str:
return upstream + "_processed"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", cmd=["echo", "hello"], conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("upstream", cmd=["echo", "hello"], conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
]
)
report = px.run(graph, strategy="thread", max_workers=2)
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SKIPPED
@@ -581,10 +617,12 @@ def test_downstream_skipped_when_upstream_skipped_async() -> None:
never_true = lambda _ctx: False # noqa: E731
graph = px.Graph.from_specs([
px.TaskSpec("upstream", upstream, conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("upstream", upstream, conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
]
)
report = px.run(graph, strategy="async")
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SKIPPED
@@ -601,10 +639,12 @@ def test_downstream_executes_when_upstream_succeeds() -> None:
def downstream(upstream: str) -> str:
return upstream + "_processed"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", upstream, conditions=(always_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("upstream", upstream, conditions=(always_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SUCCESS
tests/test_executors_edge_cases.py
+63 -43
View File
@@ -271,10 +271,12 @@ def test_allow_upstream_skip_allows_execution_after_skipped() -> None:
def downstream_task() -> str:
return "ran despite upstream skipped"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", fn=lambda: "up", conditions=(never_true,)),
px.TaskSpec("downstream", fn=downstream_task, depends_on=("upstream",), allow_upstream_skip=True),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("upstream", fn=lambda: "up", conditions=(never_true,)),
px.TaskSpec("downstream", fn=downstream_task, depends_on=("upstream",), allow_upstream_skip=True),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report.results["upstream"].status == TaskStatus.SKIPPED
@@ -291,10 +293,12 @@ def test_upstream_failed_skips_downstream() -> None:
def downstream():
return "should not run"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", fn=boom),
px.TaskSpec("downstream", fn=downstream, depends_on=("upstream",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("upstream", fn=boom),
px.TaskSpec("downstream", fn=downstream, depends_on=("upstream",)),
]
)
with pytest.raises(px.TaskFailedError):
px.run(graph, strategy="sequential")
@@ -342,11 +346,13 @@ def test_concurrency_key_thread() -> None:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", fn=make("a"), concurrency_key="group1"),
px.TaskSpec("b", fn=make("b"), concurrency_key="group1"),
px.TaskSpec("c", fn=make("c"), concurrency_key="group1"),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", fn=make("a"), concurrency_key="group1"),
px.TaskSpec("b", fn=make("b"), concurrency_key="group1"),
px.TaskSpec("c", fn=make("c"), concurrency_key="group1"),
]
)
report = px.run(graph, strategy="thread", max_workers=10, concurrency_limits={"group1": 1})
assert report.success
# 由于 concurrency_key 限制为 1，任务应串行执行
@@ -366,10 +372,12 @@ def test_concurrency_key_async() -> None:
await asyncio.sleep(0.01)
return "b"
graph = px.Graph.from_specs([
px.TaskSpec("a", fn=task_a, concurrency_key="group1"),
px.TaskSpec("b", fn=task_b, concurrency_key="group1"),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", fn=task_a, concurrency_key="group1"),
px.TaskSpec("b", fn=task_b, concurrency_key="group1"),
]
)
report = px.run(graph, strategy="async", concurrency_limits={"group1": 1})
assert report.success
@@ -388,12 +396,14 @@ def test_dependency_strategy_basic() -> None:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", fn=make("a")),
px.TaskSpec("b", fn=make("b"), depends_on=("a",)),
px.TaskSpec("c", fn=make("c"), depends_on=("a",)),
px.TaskSpec("d", fn=make("d"), depends_on=("b", "c")),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", fn=make("a")),
px.TaskSpec("b", fn=make("b"), depends_on=("a",)),
px.TaskSpec("c", fn=make("c"), depends_on=("a",)),
px.TaskSpec("d", fn=make("d"), depends_on=("b", "c")),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert "a" in order
@@ -409,10 +419,12 @@ def test_dependency_strategy_async() -> None:
async def b(a: str):
return a + "b"
graph = px.Graph.from_specs([
px.TaskSpec("a", fn=a),
px.TaskSpec("b", fn=b, depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", fn=a),
px.TaskSpec("b", fn=b, depends_on=("a",)),
]
)
report = px.run(graph, strategy="dependency")
assert report.success
assert report["b"] == "ab"
@@ -427,10 +439,12 @@ def test_continue_on_error_marks_failed_but_continues() -> None:
def boom():
raise ValueError("boom")
graph = px.Graph.from_specs([
px.TaskSpec("fail", fn=boom, continue_on_error=True),
px.TaskSpec("other", fn=lambda: "ok"), # 无依赖，应继续
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", fn=boom, continue_on_error=True),
px.TaskSpec("other", fn=lambda: "ok"), # 无依赖，应继续
]
)
# continue_on_error=True 时 run 不抛异常，report.success 为 True
report = px.run(graph, strategy="sequential")
# report.success 为 True 因为没有抛 TaskFailedError
@@ -448,10 +462,12 @@ def test_continue_on_error_downstream_skipped() -> None:
def downstream():
return "should not run"
graph = px.Graph.from_specs([
px.TaskSpec("fail", fn=boom, continue_on_error=True),
px.TaskSpec("dep", fn=downstream, depends_on=("fail",), allow_upstream_skip=False),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", fn=boom, continue_on_error=True),
px.TaskSpec("dep", fn=downstream, depends_on=("fail",), allow_upstream_skip=False),
]
)
report = px.run(graph, strategy="sequential")
# report.success 为 True 因为 continue_on_error 阻止了 TaskFailedError
assert report.success
@@ -468,10 +484,12 @@ def test_soft_depends_on_default_value_injection() -> None:
def task_with_soft_dep(a: str | None = None) -> str:
return f"a={a}"
graph = px.Graph.from_specs([
px.TaskSpec("a", fn=lambda: "value"),
px.TaskSpec("b", fn=task_with_soft_dep, soft_depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", fn=lambda: "value"),
px.TaskSpec("b", fn=task_with_soft_dep, soft_depends_on=("a",)),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
assert report["b"] == "a=value"
@@ -484,10 +502,12 @@ def test_soft_depends_on_skipped_injects_none() -> None:
def task_with_soft_dep(skipped: str | None = None) -> str:
return f"skipped={skipped}"
graph = px.Graph.from_specs([
px.TaskSpec("skipped", fn=lambda: "value", conditions=(never_true,)),
px.TaskSpec("b", fn=task_with_soft_dep, soft_depends_on=("skipped",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("skipped", fn=lambda: "value", conditions=(never_true,)),
px.TaskSpec("b", fn=task_with_soft_dep, soft_depends_on=("skipped",)),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
# 软依赖被 skipped 时注入 None（因为 global_context 中有 skipped，值为 None）
tests/test_graph.py
+98 -70
View File
@@ -14,11 +14,13 @@ def _fn() -> None:
def test_from_specs_builds_graph() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("c", _fn, depends_on=("a", "b")),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("c", _fn, depends_on=("a", "b")),
]
)
assert set(graph.names) == {"a", "b", "c"}
assert graph.dependencies("c") == ("a", "b")
assert len(graph) == 3
@@ -27,19 +29,23 @@ def test_from_specs_builds_graph() -> None:
def test_from_specs_allows_forward_references() -> None:
# b depends on a, but a is declared after b — order should not matter.
graph = px.Graph.from_specs([
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("a", _fn),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("a", _fn),
]
)
assert graph.layers() == [["a"], ["b"]]
def test_duplicate_task_raises() -> None:
with pytest.raises(DuplicateTaskError):
_ = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("a", _fn),
])
_ = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("a", _fn),
]
)
def test_missing_dependency_raises() -> None:
@@ -52,20 +58,24 @@ def test_missing_dependency_raises() -> None:
def test_cycle_detection() -> None:
with pytest.raises(CycleError):
_ = px.Graph.from_specs([
px.TaskSpec("a", _fn, depends_on=("c",)),
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("c", _fn, depends_on=("b",)),
])
_ = px.Graph.from_specs(
[
px.TaskSpec("a", _fn, depends_on=("c",)),
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("c", _fn, depends_on=("b",)),
]
)
def test_layers_grouping() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn),
px.TaskSpec("c", _fn, depends_on=("a", "b")),
px.TaskSpec("d", _fn, depends_on=("c",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn),
px.TaskSpec("c", _fn, depends_on=("a", "b")),
px.TaskSpec("d", _fn, depends_on=("c",)),
]
)
layers = graph.layers()
assert layers == [["a", "b"], ["c"], ["d"]]
@@ -76,10 +86,12 @@ def test_self_dependency_rejected() -> None:
def test_to_mermaid() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
]
)
mermaid = graph.to_mermaid()
assert mermaid.startswith("graph TD")
assert 'a["a"]' in mermaid
@@ -93,11 +105,13 @@ def test_to_mermaid_invalid_orientation() -> None:
def test_subgraph_by_tags() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn, tags=("ingest",)),
px.TaskSpec("b", _fn, depends_on=("a",), tags=("ingest",)),
px.TaskSpec("c", _fn, depends_on=("b",), tags=("report",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn, tags=("ingest",)),
px.TaskSpec("b", _fn, depends_on=("a",), tags=("ingest",)),
px.TaskSpec("c", _fn, depends_on=("b",), tags=("report",)),
]
)
sub = graph.subgraph(["ingest"])
assert set(sub.names) == {"a", "b"}
# Edge to dropped task c is removed; b no longer waits for anything
@@ -106,11 +120,13 @@ def test_subgraph_by_tags() -> None:
def test_subgraph_by_names() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("c", _fn, depends_on=("b",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
px.TaskSpec("c", _fn, depends_on=("b",)),
]
)
sub = graph.subgraph_by_names(["a", "b"])
assert set(sub.names) == {"a", "b"}
# c is dropped, so b's dep on c (none here) — but a->b edge preserved.
@@ -124,10 +140,12 @@ def test_subgraph_by_names_unknown() -> None:
def test_describe() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
]
)
desc = graph.describe()
assert "Layer 1" in desc
assert "Layer 2" in desc
@@ -164,11 +182,13 @@ def test_all_specs_returns_view() -> None:
def test_all_deps_combines_hard_and_soft() -> None:
"""all_deps 应返回硬依赖 + 软依赖的组合。"""
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn),
px.TaskSpec("c", _fn, depends_on=("a",), soft_depends_on=("b",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn),
px.TaskSpec("c", _fn, depends_on=("a",), soft_depends_on=("b",)),
]
)
all_deps = graph.all_deps("c")
assert set(all_deps) == {"a", "b"}
# 硬依赖在前，软依赖在后
@@ -183,10 +203,12 @@ def test_spec_accessor() -> None:
def test_dependencies_accessor() -> None:
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, depends_on=("a",)),
]
)
assert graph.dependencies("a") == ()
assert graph.dependencies("b") == ("a",)
@@ -205,16 +227,18 @@ def test_empty_graph_layers() -> None:
def test_subgraph_preserves_metadata() -> None:
"""子图应保留原任务的 retry/timeout/tags 等元数据。"""
graph = px.Graph.from_specs([
px.TaskSpec(
"a",
_fn,
tags=("x",),
retry=px.RetryPolicy(max_attempts=3),
timeout=5.0,
),
px.TaskSpec("b", _fn, depends_on=("a",), tags=("y",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"a",
_fn,
tags=("x",),
retry=px.RetryPolicy(max_attempts=3),
timeout=5.0,
),
px.TaskSpec("b", _fn, depends_on=("a",), tags=("y",)),
]
)
sub = graph.subgraph(["x"])
spec = sub.spec("a")
assert spec.retry.max_attempts == 3
@@ -250,10 +274,12 @@ def test_from_specs_with_invalid_type() -> None:
# ---------------------------------------------------------------------- #
def test_to_mermaid_soft_depends_on() -> None:
"""to_mermaid 应正确绘制软依赖为虚线."""
graph = px.Graph.from_specs([
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, soft_depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", _fn),
px.TaskSpec("b", _fn, soft_depends_on=("a",)),
]
)
mermaid = graph.to_mermaid()
assert "a -.-> b" in mermaid # 软依赖用虚线
@@ -355,11 +381,13 @@ def test_graph_composer_expand_refs_ref_returns_empty() -> None:
def test_graph_composer_expand_refs_multiple_original_specs_serialized() -> None:
"""expand_refs 多个 original_specs 应串行依赖，且首个依赖 ref 末任务."""
graph_a = px.Graph.from_specs([px.TaskSpec("a1", _fn)])
graph_b = px.Graph.from_specs([
px.TaskSpec("b1", _fn),
px.TaskSpec("b2", _fn),
px.TaskSpec("b3", _fn),
])
graph_b = px.Graph.from_specs(
[
px.TaskSpec("b1", _fn),
px.TaskSpec("b2", _fn),
px.TaskSpec("b3", _fn),
]
)
graph_b._pending_refs = ["cmd_a"]
composer = GraphComposer({"cmd_a": graph_a, "cmd_b": graph_b})
tests/test_profiling.py
+62 -44
View File
@@ -95,11 +95,13 @@ class TestProfileReportConstruction:
report.results["a"] = _result("a", start, 1.0)
report.results["b"] = _result("b", start + timedelta(seconds=1), 2.0)
report.results["c"] = _result("c", start + timedelta(seconds=3), 1.5)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
_spec("c", deps=("b",)),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b", deps=("a",)),
_spec("c", deps=("b",)),
]
)
profile = ProfileReport.from_report(report, graph)
@@ -136,11 +138,13 @@ class TestProfileReportConstruction:
report.results["a"] = _result("a", start, 1.0)
report.results["b"] = _result("b", start, 3.0)
report.results["c"] = _result("c", start + timedelta(seconds=3), 1.0)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b"),
_spec("c", deps=("a", "b")),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b"),
_spec("c", deps=("a", "b")),
]
)
profile = ProfileReport.from_report(report, graph)
@@ -188,10 +192,12 @@ class TestWaitTime:
report = px.RunReport()
report.results["a"] = _result("a", start, 1.0)
report.results["b"] = _result("b", start + timedelta(seconds=1.5), 1.0)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b", deps=("a",)),
]
)
profile = ProfileReport.from_report(report, graph)
@@ -204,10 +210,12 @@ class TestWaitTime:
report.results["a"] = _result("a", start, 2.0)
# b 在 a 还没完成时就开始（不应该但可能发生）
report.results["b"] = _result("b", start + timedelta(seconds=1), 1.0)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b", deps=("a",)),
]
)
profile = ProfileReport.from_report(report, graph)
@@ -220,10 +228,12 @@ class TestWaitTime:
report = px.RunReport()
report.results["a"] = _result("a", start, 1.0)
report.results["b"] = _skipped_result("b")
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b", deps=("a",)),
]
)
profile = ProfileReport.from_report(report, graph)
@@ -241,12 +251,14 @@ class TestCriticalPath:
report.results["b"] = _result("b", start + timedelta(seconds=1), 3.0)
report.results["c"] = _result("c", start + timedelta(seconds=1), 1.0)
report.results["d"] = _result("d", start + timedelta(seconds=4), 1.0)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
_spec("c", deps=("a",)),
_spec("d", deps=("b", "c")),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b", deps=("a",)),
_spec("c", deps=("a",)),
_spec("d", deps=("b", "c")),
]
)
profile = ProfileReport.from_report(report, graph)
@@ -393,12 +405,14 @@ class TestQueries:
report.results["b"] = _result("b", start + timedelta(seconds=1), 3.0)
report.results["c"] = _result("c", start + timedelta(seconds=1), 1.0)
report.results["d"] = _result("d", start + timedelta(seconds=4), 1.0)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
_spec("c", deps=("a",)),
_spec("d", deps=("b", "c")),
])
graph = px.Graph.from_specs(
[
_spec("a"),
_spec("b", deps=("a",)),
_spec("c", deps=("a",)),
_spec("d", deps=("b", "c")),
]
)
profile = ProfileReport.from_report(report, graph)
@@ -537,11 +551,13 @@ class TestIntegrationWithRun:
time.sleep(0.01) # 确保任务有实际耗时，避免 duration 极小导致并行度计算为 0
return 1
graph = px.Graph.from_specs([
px.TaskSpec("a", slow),
px.TaskSpec("b", slow, depends_on=("a",)),
px.TaskSpec("c", slow, depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", slow),
px.TaskSpec("b", slow, depends_on=("a",)),
px.TaskSpec("c", slow, depends_on=("a",)),
]
)
report = px.run(graph, strategy="sequential")
profile = ProfileReport.from_report(report, graph)
@@ -560,11 +576,13 @@ class TestIntegrationWithRun:
time.sleep(0.05)
return 1
graph = px.Graph.from_specs([
px.TaskSpec("a", slow),
px.TaskSpec("b", slow),
px.TaskSpec("c", slow),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", slow),
px.TaskSpec("b", slow),
px.TaskSpec("c", slow),
]
)
report = px.run(graph, strategy="thread", max_workers=3)
profile = ProfileReport.from_report(report, graph)
tests/test_report.py
-1
View File
@@ -172,4 +172,3 @@ class TestRunReportQueries:
report.results["a"] = TaskResult[Any](spec=spec, status=TaskStatus.PENDING)
durs = report.durations()
assert durs["a"] == 0.0
tests/test_runner.py
+55 -41
View File
@@ -29,20 +29,24 @@ def _echo_graph(name: str = "echo_task", msg: str = "hello") -> px.Graph:
def _failing_graph() -> px.Graph:
"""构造一个必定失败的单任务图."""
return px.Graph.from_specs([
px.TaskSpec(
"fail",
cmd=["python", "-c", "import sys; sys.exit(1)"],
)
])
return px.Graph.from_specs(
[
px.TaskSpec(
"fail",
cmd=["python", "-c", "import sys; sys.exit(1)"],
)
]
)
def _multi_task_graph() -> px.Graph:
"""构造一个带依赖的多任务图."""
return px.Graph.from_specs([
px.TaskSpec("a", cmd=[*ECHO_CMD, "a"]),
px.TaskSpec("b", cmd=[*ECHO_CMD, "b"], depends_on=("a",)),
])
return px.Graph.from_specs(
[
px.TaskSpec("a", cmd=[*ECHO_CMD, "a"]),
px.TaskSpec("b", cmd=[*ECHO_CMD, "b"], depends_on=("a",)),
]
)
# ---------------------------------------------------------------------- #
@@ -314,13 +318,15 @@ class TestCliRunnerVerbose:
def test_verbose_prints_skip_lifecycle(self, capsys: pytest.CaptureFixture[str]) -> None:
"""verbose 模式下跳过的任务应打印跳过信息."""
graph = px.Graph.from_specs([
px.TaskSpec(
"skip_me",
cmd=[*ECHO_CMD, "skip"],
conditions=(lambda _ctx: False,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"skip_me",
cmd=[*ECHO_CMD, "skip"],
conditions=(lambda _ctx: False,),
),
]
)
runner = px.CliRunner(aliases={"skip": graph})
_ = runner.run(["skip"])
captured = capsys.readouterr()
@@ -517,26 +523,30 @@ class TestCliRunnerIntegration:
def test_condition_skipped_command_succeeds(self) -> None:
"""条件不满足时任务跳过, 整体仍成功."""
graph = px.Graph.from_specs([
px.TaskSpec(
"skip_me",
cmd=[*ECHO_CMD, "should not run"],
conditions=(lambda _ctx: False,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"skip_me",
cmd=[*ECHO_CMD, "should not run"],
conditions=(lambda _ctx: False,),
),
]
)
runner = px.CliRunner(aliases={"skip": graph})
exit_code = runner.run(["skip"])
assert exit_code == CliExitCode.SUCCESS.value
def test_condition_met_command_succeeds(self) -> None:
"""条件满足时任务执行, 整体成功."""
graph = px.Graph.from_specs([
px.TaskSpec(
"run_me",
cmd=[*ECHO_CMD, "should run"],
conditions=(lambda _ctx: True,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"run_me",
cmd=[*ECHO_CMD, "should run"],
conditions=(lambda _ctx: True,),
),
]
)
runner = px.CliRunner(aliases={"run": graph})
exit_code = runner.run(["run"])
assert exit_code == CliExitCode.SUCCESS.value
@@ -552,12 +562,14 @@ class TestCliRunnerIntegration:
return fn
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
px.TaskSpec("c", make("c"), depends_on=("a",)),
px.TaskSpec("d", make("d"), depends_on=("b", "c")),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
px.TaskSpec("c", make("c"), depends_on=("a",)),
px.TaskSpec("d", make("d"), depends_on=("b", "c")),
]
)
runner = px.CliRunner(aliases={"diamond": graph})
exit_code = runner.run(["diamond"])
assert exit_code == CliExitCode.SUCCESS.value
@@ -661,10 +673,12 @@ class TestCliRunnerNewApi:
def test_aliases_graph_value(self) -> None:
"""aliases 值为 Graph 时原样使用（复杂场景：conditions 等）."""
graph = px.Graph.from_specs([
px.TaskSpec("a", cmd=[*ECHO_CMD, "a"]),
px.TaskSpec("b", cmd=[*ECHO_CMD, "b"], depends_on=("a",)),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("a", cmd=[*ECHO_CMD, "a"]),
px.TaskSpec("b", cmd=[*ECHO_CMD, "b"], depends_on=("a",)),
]
)
runner = px.CliRunner(aliases={"g": graph})
assert set(runner.graphs["g"].all_specs().keys()) == {"a", "b"}
tests/test_taskspec_commands.py
+178 -142
View File
@@ -21,9 +21,11 @@ else:
def test_taskspec_with_cmd_list():
"""测试使用命令列表的 TaskSpec."""
graph = px.Graph.from_specs([
px.TaskSpec("echo_test", cmd=[*ECHO_CMD, "hello"]),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("echo_test", cmd=[*ECHO_CMD, "hello"]),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -38,9 +40,11 @@ def test_taskspec_with_cmd_string():
else:
shell_cmd = "echo 'hello from shell'"
graph = px.Graph.from_specs([
px.TaskSpec("shell_test", cmd=shell_cmd),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("shell_test", cmd=shell_cmd),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -55,13 +59,15 @@ def test_taskspec_with_conditions_skip():
def never_true(_ctx):
return False
graph = px.Graph.from_specs([
px.TaskSpec(
"should_skip",
cmd=[*ECHO_CMD, "this should not run"],
conditions=(never_true,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"should_skip",
cmd=[*ECHO_CMD, "this should not run"],
conditions=(never_true,),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -76,13 +82,15 @@ def test_taskspec_with_conditions_execute():
def always_true(_ctx):
return True
graph = px.Graph.from_specs([
px.TaskSpec(
"should_run",
cmd=[*ECHO_CMD, "this should run"],
conditions=(always_true,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"should_run",
cmd=[*ECHO_CMD, "this should run"],
conditions=(always_true,),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -99,23 +107,25 @@ def test_platform_conditions():
win_cmd = ["echo", "Windows"]
posix_cmd = ["echo", "POSIX"]
graph = px.Graph.from_specs([
px.TaskSpec(
"win_task",
cmd=win_cmd,
conditions=(lambda _ctx: Constants.IS_WINDOWS,),
),
px.TaskSpec(
"linux_task",
cmd=posix_cmd,
conditions=(lambda _ctx: Constants.IS_LINUX,),
),
px.TaskSpec(
"macos_task",
cmd=posix_cmd,
conditions=(lambda _ctx: Constants.IS_MACOS,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"win_task",
cmd=win_cmd,
conditions=(lambda _ctx: Constants.IS_WINDOWS,),
),
px.TaskSpec(
"linux_task",
cmd=posix_cmd,
conditions=(lambda _ctx: Constants.IS_LINUX,),
),
px.TaskSpec(
"macos_task",
cmd=posix_cmd,
conditions=(lambda _ctx: Constants.IS_MACOS,),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -141,13 +151,15 @@ def test_app_installed_conditions():
python_cmd = [sys.executable, "--version"]
py_name = "python" if sys.platform == "win32" else "python3"
graph = px.Graph.from_specs([
px.TaskSpec(
"python_check",
cmd=python_cmd,
conditions=(BuiltinConditions.HAS_INSTALLED(py_name),),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"python_check",
cmd=python_cmd,
conditions=(BuiltinConditions.HAS_INSTALLED(py_name),),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -173,23 +185,25 @@ def test_combined_conditions():
# NOT 条件
not_condition = BuiltinConditions.NOT(lambda _ctx: False)
graph = px.Graph.from_specs([
px.TaskSpec(
"and_test",
cmd=[*ECHO_CMD, "AND"],
conditions=(and_condition,),
),
px.TaskSpec(
"or_test",
cmd=[*ECHO_CMD, "OR"],
conditions=(or_condition,),
),
px.TaskSpec(
"not_test",
cmd=[*ECHO_CMD, "NOT"],
conditions=(not_condition,),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"and_test",
cmd=[*ECHO_CMD, "AND"],
conditions=(and_condition,),
),
px.TaskSpec(
"or_test",
cmd=[*ECHO_CMD, "OR"],
conditions=(or_condition,),
),
px.TaskSpec(
"not_test",
cmd=[*ECHO_CMD, "NOT"],
conditions=(not_condition,),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -205,13 +219,15 @@ def test_taskspec_with_cwd():
else:
ls_cmd = ["ls", "-la"]
graph = px.Graph.from_specs([
px.TaskSpec(
"list_current",
cmd=ls_cmd,
cwd=Path.cwd(),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"list_current",
cmd=ls_cmd,
cwd=Path.cwd(),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -222,14 +238,16 @@ def test_taskspec_with_cwd():
@pytest.mark.slow
def test_taskspec_with_timeout():
"""测试超时设置."""
graph = px.Graph.from_specs([
# 短时间任务应该成功
px.TaskSpec(
"short_task",
cmd=[sys.executable, "-c", "import time; time.sleep(0.1)"],
timeout=1.0,
),
])
graph = px.Graph.from_specs(
[
# 短时间任务应该成功
px.TaskSpec(
"short_task",
cmd=[sys.executable, "-c", "import time; time.sleep(0.1)"],
timeout=1.0,
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -239,24 +257,26 @@ def test_taskspec_with_timeout():
def test_taskspec_dependency_with_conditions():
"""测试依赖和条件的组合."""
graph = px.Graph.from_specs([
px.TaskSpec(
"first",
cmd=[*ECHO_CMD, "first"],
conditions=(lambda _ctx: True,),
),
px.TaskSpec(
"second",
cmd=[*ECHO_CMD, "second"],
depends_on=("first",),
conditions=(lambda _ctx: True,),
),
px.TaskSpec(
"third",
cmd=[*ECHO_CMD, "third"],
depends_on=("second",),
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"first",
cmd=[*ECHO_CMD, "first"],
conditions=(lambda _ctx: True,),
),
px.TaskSpec(
"second",
cmd=[*ECHO_CMD, "second"],
depends_on=("first",),
conditions=(lambda _ctx: True,),
),
px.TaskSpec(
"third",
cmd=[*ECHO_CMD, "third"],
depends_on=("second",),
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -271,10 +291,12 @@ def test_taskspec_mixed_fn_and_cmd():
def my_function():
return "result from function"
graph = px.Graph.from_specs([
px.TaskSpec("fn_task", fn=my_function),
px.TaskSpec("cmd_task", cmd=[*ECHO_CMD, "from command"]),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fn_task", fn=my_function),
px.TaskSpec("cmd_task", cmd=[*ECHO_CMD, "from command"]),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -289,13 +311,15 @@ def test_taskspec_cmd_overrides_fn():
def my_function():
return "should not run"
graph = px.Graph.from_specs([
px.TaskSpec(
"cmd_priority",
fn=my_function,
cmd=[*ECHO_CMD, "cmd takes priority"],
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"cmd_priority",
fn=my_function,
cmd=[*ECHO_CMD, "cmd takes priority"],
),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -310,9 +334,11 @@ def test_taskspec_callable_cmd():
def my_callable():
return "callable result"
graph = px.Graph.from_specs([
px.TaskSpec("callable_cmd", cmd=my_callable),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("callable_cmd", cmd=my_callable),
]
)
report = px.run(graph, strategy="sequential")
assert report.success
@@ -373,13 +399,15 @@ class TestTaskSpecVerbose:
"""verbose=True 时失败也应打印返回码."""
from pyflowx.errors import TaskFailedError
graph = px.Graph.from_specs([
px.TaskSpec(
"fail",
cmd=[sys.executable, "-c", "import sys; sys.exit(1)"],
verbose=True,
)
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"fail",
cmd=[sys.executable, "-c", "import sys; sys.exit(1)"],
verbose=True,
)
]
)
with pytest.raises(TaskFailedError):
_ = px.run(graph, strategy="sequential")
captured = capsys.readouterr()
@@ -408,16 +436,18 @@ class TestTaskSpecCmdErrors:
"""命令失败时错误信息应包含 stderr."""
from pyflowx.errors import TaskFailedError
graph = px.Graph.from_specs([
px.TaskSpec(
"fail",
cmd=[
sys.executable,
"-c",
"import sys; sys.stderr.write('error-msg'); sys.exit(1)",
],
)
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"fail",
cmd=[
sys.executable,
"-c",
"import sys; sys.stderr.write('error-msg'); sys.exit(1)",
],
)
]
)
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
# 非 verbose 模式下, stderr 应包含在错误信息中
@@ -435,9 +465,11 @@ class TestTaskSpecCmdErrors:
"""shell 命令失败时应抛出 RuntimeError."""
from pyflowx.errors import TaskFailedError
graph = px.Graph.from_specs([
px.TaskSpec("fail", cmd=f'{sys.executable} -c "import sys; sys.exit(1)"'),
])
graph = px.Graph.from_specs(
[
px.TaskSpec("fail", cmd=f'{sys.executable} -c "import sys; sys.exit(1)"'),
]
)
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
assert "Shell 命令执行失败" in str(exc_info.value.cause)
@@ -447,13 +479,15 @@ class TestTaskSpecCmdErrors:
"""命令超时应抛出 RuntimeError."""
from pyflowx.errors import TaskFailedError
graph = px.Graph.from_specs([
px.TaskSpec(
"slow",
cmd=[sys.executable, "-c", "import time; time.sleep(5)"],
timeout=0.1,
)
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"slow",
cmd=[sys.executable, "-c", "import time; time.sleep(5)"],
timeout=0.1,
)
]
)
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
assert "超时" in str(exc_info.value.cause)
@@ -463,13 +497,15 @@ class TestTaskSpecCmdErrors:
"""shell 命令超时应抛出 RuntimeError."""
from pyflowx.errors import TaskFailedError
graph = px.Graph.from_specs([
px.TaskSpec(
"slow",
cmd=f'{sys.executable} -c "import time; time.sleep(5)"',
timeout=0.1,
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(
"slow",
cmd=f'{sys.executable} -c "import time; time.sleep(5)"',
timeout=0.1,
),
]
)
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
assert "超时" in str(exc_info.value.cause)