Files
wehub-resource-sync 4b6817381b
Benchmark image — build + push to ECR (any adapter) / build + push (push) Waiting to run
CI / quality (ubuntu-latest) (push) Waiting to run
CI / test (tools-runtime) (push) Waiting to run
CI / test (e2e-general) (push) Waiting to run
CI / test (cli-runtime) (push) Waiting to run
CI / test (e2e-provider-and-openclaw) (push) Waiting to run
CI / test (integrations-and-misc) (push) Waiting to run
CI / coverage-report (push) Blocked by required conditions
CI / test-kubernetes (push) Waiting to run
CI / should-run-thorough (push) Waiting to run
CI / test-thorough (cloudwatch-demo) (push) Blocked by required conditions
CI / test-thorough (flink-ecs) (push) Blocked by required conditions
CI / test-thorough (upstream-lambda) (push) Blocked by required conditions
CI / test-thorough (prefect-ecs-fargate) (push) Blocked by required conditions
CodeQL / Analyze (python) (push) Waiting to run
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Blocked by required conditions
Release / publish-release (push) Blocked by required conditions
Release / publish-main-release (push) Blocked by required conditions
Release / prepare (push) Waiting to run
Release / verify (push) Blocked by required conditions
Release / build-python-dist (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Blocked by required conditions
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Waiting to run
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

420 lines
14 KiB
Python

from __future__ import annotations
import json
from types import SimpleNamespace
from surfaces.cli.tests import interactive
from surfaces.cli.tests.catalog import TestCatalog as Catalog
from surfaces.cli.tests.catalog import TestCatalogItem as CatalogItem
from surfaces.cli.tests.discover import load_test_catalog
def test_choose_interactive_item_returns_single_match_without_extra_prompt(monkeypatch) -> None:
catalog = Catalog(
items=(
CatalogItem(
id="make:test-cov",
kind="make_target",
display_name="Coverage Suite",
description="Run the coverage suite.",
command=("make", "test-cov"),
tags=("ci-safe",),
),
)
)
selected_prompts: list[str] = []
monkeypatch.setattr(interactive, "_choose_category", lambda: "ci-safe")
def _mock_select_item(items, *, prompt: str, allow_back: bool = False):
_ = allow_back
selected_prompts.append(prompt)
return items[0]
monkeypatch.setattr(interactive, "_select_item", _mock_select_item)
item, auto_selected = interactive.choose_interactive_item(catalog)
assert item.id == "make:test-cov"
assert auto_selected is True
assert selected_prompts == []
def test_choose_interactive_item_prompts_when_multiple_matches_exist(monkeypatch) -> None:
catalog = load_test_catalog()
selected_prompts: list[str] = []
selected_item_ids: list[list[str]] = []
monkeypatch.setattr(interactive, "_choose_category", lambda: "ci-safe")
def _mock_select_item_or_all(items, *, prompt: str, allow_back: bool = False):
_ = allow_back
selected_prompts.append(prompt)
selected_item_ids.append([item.id for item in items])
return items[0]
monkeypatch.setattr(interactive, "_select_item_or_all", _mock_select_item_or_all)
item, auto_selected = interactive.choose_interactive_item(catalog)
assert item.id == selected_item_ids[0][0]
assert auto_selected is False
assert selected_prompts == ["Choose a test or suite:"]
assert "make:test-cov" in selected_item_ids[0]
def test_choose_interactive_item_retries_after_empty_filter(monkeypatch) -> None:
catalog = Catalog(
items=(
CatalogItem(
id="make:test-cov",
kind="make_target",
display_name="Coverage Suite",
description="Run the coverage suite.",
command=("make", "test-cov"),
tags=("ci-safe",),
),
)
)
category_choices = iter(["rca", "ci-safe"])
monkeypatch.setattr(interactive, "_choose_category", lambda: next(category_choices))
item, auto_selected = interactive.choose_interactive_item(catalog)
assert item.id == "make:test-cov"
assert auto_selected is True
def test_choose_interactive_item_reselects_category_after_escape(monkeypatch) -> None:
catalog = Catalog(
items=(
CatalogItem(
id="make:test-cov",
kind="make_target",
display_name="Coverage Suite",
description="Run the coverage suite.",
command=("make", "test-cov"),
tags=("ci-safe",),
),
CatalogItem(
id="make:test-full",
kind="make_target",
display_name="Full Suite",
description="Run the full suite.",
command=("make", "test-full"),
tags=("ci-safe",),
),
CatalogItem(
id="make:demo",
kind="make_target",
display_name="Prefect ECS Demo",
description="Run the demo.",
command=("make", "demo"),
tags=("demo",),
),
)
)
category_choices = iter(["ci-safe", "demo"])
selected_prompts: list[str] = []
monkeypatch.setattr(interactive, "_choose_category", lambda: next(category_choices))
def _mock_select_item_or_all(items, *, prompt: str, allow_back: bool = False):
_ = allow_back
selected_prompts.append(prompt)
if len(selected_prompts) == 1:
raise interactive._GoBack
return items[0]
monkeypatch.setattr(interactive, "_select_item_or_all", _mock_select_item_or_all)
item, _ = interactive.choose_interactive_item(catalog)
assert item.id == "make:demo"
assert selected_prompts == ["Choose a test or suite:"]
def test_choose_interactive_item_returns_to_parent_list_after_escape(monkeypatch) -> None:
suite = CatalogItem(
id="suite:demo",
kind="suite",
display_name="Demo Suite",
description="A grouped demo suite.",
tags=("demo",),
children=(
CatalogItem(
id="scenario:demo:first",
kind="scenario",
display_name="First scenario",
description="First child.",
command=("make", "demo"),
tags=("demo",),
),
CatalogItem(
id="scenario:demo:second",
kind="scenario",
display_name="Second scenario",
description="Second child.",
command=("make", "demo"),
tags=("demo",),
),
),
)
leaf = CatalogItem(
id="make:demo",
kind="make_target",
display_name="Standalone Demo",
description="Run the demo.",
command=("make", "demo"),
tags=("demo",),
)
catalog = Catalog(items=(suite, leaf))
selected_prompts: list[str] = []
monkeypatch.setattr(interactive, "_choose_category", lambda: "demo")
def _mock_select_item(items, *, prompt: str, allow_back: bool = False):
_ = (items, allow_back)
selected_prompts.append(prompt)
if prompt == "Choose a test or suite:" and len(selected_prompts) == 1:
return suite
if prompt == "Select a scenario from Demo Suite:":
raise interactive._GoBack
return leaf
def _mock_select_item_or_all(items, *, prompt: str, allow_back: bool = False):
return _mock_select_item(items, prompt=prompt, allow_back=allow_back)
monkeypatch.setattr(interactive, "_select_item", _mock_select_item)
monkeypatch.setattr(interactive, "_select_item_or_all", _mock_select_item_or_all)
item, _ = interactive.choose_interactive_item(catalog)
assert item.id == "make:demo"
assert selected_prompts == [
"Choose a test or suite:",
"Select a scenario from Demo Suite:",
"Choose a test or suite:",
]
def test_select_item_or_all_allows_literal_run_all_item_id(monkeypatch) -> None:
item = CatalogItem(
id="__run_all__",
kind="make_target",
display_name="Literal Run All Test",
description="A test whose id matches the old sentinel string.",
command=("make", "test-cov"),
tags=("ci-safe",),
)
def _mock_select_prompt(*_args, **_kwargs):
return SimpleNamespace(ask=lambda: "__run_all__")
monkeypatch.setattr(interactive, "_require_interactive_dependencies", lambda: None)
monkeypatch.setattr(
interactive,
"_QuestionaryChoice",
lambda *, title, value: {"title": title, "value": value},
)
monkeypatch.setattr(interactive, "_select_prompt", _mock_select_prompt)
selection = interactive._select_item_or_all([item], prompt="Choose a test or suite:")
assert selection is item
def test_choose_interactive_item_expands_run_all_suites(monkeypatch) -> None:
suite = CatalogItem(
id="suite:demo",
kind="suite",
display_name="Demo Suite",
description="A grouped demo suite.",
tags=("demo",),
children=(
CatalogItem(
id="scenario:demo:first",
kind="scenario",
display_name="First scenario",
description="First child.",
command=("make", "demo-first"),
tags=("demo",),
),
CatalogItem(
id="scenario:demo:second",
kind="scenario",
display_name="Second scenario",
description="Second child.",
command=("make", "demo-second"),
tags=("demo",),
),
),
)
leaf = CatalogItem(
id="make:demo",
kind="make_target",
display_name="Standalone Demo",
description="Run the demo.",
command=("make", "demo"),
tags=("demo",),
)
catalog = Catalog(items=(suite, leaf))
def _mock_select_item_or_all(items, *, prompt: str, allow_back: bool = False):
_ = (prompt, allow_back)
return items
monkeypatch.setattr(interactive, "_choose_category", lambda: "demo")
monkeypatch.setattr(interactive, "_select_item_or_all", _mock_select_item_or_all)
selection, auto_selected = interactive.choose_interactive_item(catalog)
assert auto_selected is False
assert [item.id for item in selection] == [
"scenario:demo:first",
"scenario:demo:second",
"make:demo",
]
def test_run_interactive_picker_returns_zero_on_escape(monkeypatch) -> None:
catalog = load_test_catalog()
monkeypatch.setattr(interactive, "_require_interactive_dependencies", lambda: None)
monkeypatch.setattr(interactive.sys, "stdin", SimpleNamespace(isatty=lambda: True))
monkeypatch.setattr(interactive.sys, "stdout", SimpleNamespace(isatty=lambda: True))
monkeypatch.setattr(
interactive,
"choose_interactive_item",
lambda _catalog: (_ for _ in ()).throw(KeyboardInterrupt()),
)
assert interactive.run_interactive_picker(catalog) == 0
def test_run_interactive_picker_returns_to_selection_after_escape_from_confirm(monkeypatch) -> None:
first = CatalogItem(
id="make:test-cov",
kind="make_target",
display_name="Coverage Suite",
description="Run coverage.",
command=("make", "test-cov"),
tags=("ci-safe",),
)
second = CatalogItem(
id="make:test-full",
kind="make_target",
display_name="Full Suite",
description="Run full tests.",
command=("make", "test-full"),
tags=("ci-safe",),
)
selections = iter([(first, False), (second, False)])
monkeypatch.setattr(interactive, "_require_interactive_dependencies", lambda: None)
monkeypatch.setattr(interactive.sys, "stdin", SimpleNamespace(isatty=lambda: True))
monkeypatch.setattr(interactive.sys, "stdout", SimpleNamespace(isatty=lambda: True))
monkeypatch.setattr(interactive, "choose_interactive_item", lambda _catalog: next(selections))
confirm_calls = 0
def _mock_confirm(item):
_ = item
nonlocal confirm_calls
confirm_calls += 1
if confirm_calls == 1:
raise interactive._GoBack
return True
monkeypatch.setattr(interactive, "_confirm_run", _mock_confirm)
monkeypatch.setattr(
interactive, "run_catalog_item", lambda item: 7 if item.id == "make:test-full" else 1
)
assert interactive.run_interactive_picker(Catalog(items=(first, second))) == 7
def test_run_interactive_picker_writes_selection_for_background_mode(monkeypatch, tmp_path) -> None:
item = CatalogItem(
id="synthetic:001-replication-lag",
kind="cli_command",
display_name="001-replication-lag",
description="Run synthetic scenario.",
command=("opensre", "tests", "synthetic", "--scenario", "001-replication-lag"),
tags=("synthetic",),
)
selection_file = tmp_path / "selection.json"
monkeypatch.setenv("OPENSRE_TEST_PICKER_SELECTION_FILE", str(selection_file))
monkeypatch.setattr(interactive, "_require_interactive_dependencies", lambda: None)
monkeypatch.setattr(interactive.sys, "stdin", SimpleNamespace(isatty=lambda: True))
monkeypatch.setattr(interactive.sys, "stdout", SimpleNamespace(isatty=lambda: True))
monkeypatch.setattr(interactive, "choose_interactive_item", lambda _catalog: (item, False))
monkeypatch.setattr(interactive, "_confirm_run", lambda _item: True)
monkeypatch.setattr(
interactive,
"run_catalog_item",
lambda _item: (_ for _ in ()).throw(AssertionError("should not run inline")),
)
assert interactive.run_interactive_picker(Catalog(items=(item,))) == 0
assert json.loads(selection_file.read_text(encoding="utf-8")) == [
{
"id": "synthetic:001-replication-lag",
"display_name": "001-replication-lag",
"command": ["opensre", "tests", "synthetic", "--scenario", "001-replication-lag"],
"command_display": "opensre tests synthetic --scenario 001-replication-lag",
}
]
def test_run_catalog_items_skips_non_runnable_items() -> None:
runnable = CatalogItem(
id="make:test-cov",
kind="make_target",
display_name="Coverage Suite",
description="Run coverage.",
command=("make", "test-cov"),
tags=("ci-safe",),
)
suite = CatalogItem(
id="suite:ci-safe",
kind="suite",
display_name="Grouped CI Tests",
description="A suite without a direct command.",
tags=("ci-safe",),
children=(runnable,),
)
assert interactive.run_catalog_items([suite, runnable], dry_run=True) == 0
def test_confirm_run_prints_openclaw_preflight_messages(monkeypatch, capsys) -> None:
item = CatalogItem(
id="rca:openclaw_gateway_crashed",
kind="rca_file",
display_name="OpenClaw Gateway Crashed",
description="Run a bundled markdown RCA alert fixture.",
command=("make", "test-rca", "FILE=openclaw_gateway_crashed"),
tags=("rca", "fixture", "openclaw"),
)
monkeypatch.setattr(
interactive,
"get_preflight_messages",
lambda _item: ("OpenClaw preflight: unavailable.", "Fix: verify openclaw."),
)
monkeypatch.setattr(interactive, "_QuestionaryChoice", lambda *, title, value: (title, value))
monkeypatch.setattr(
interactive,
"_select_prompt",
lambda *_args, **_kwargs: SimpleNamespace(ask=lambda: True),
)
assert interactive._confirm_run(item) is True
output = capsys.readouterr().out
assert "OpenClaw preflight: unavailable." in output
assert "Fix: verify openclaw." in output