style(ws0+ws2+ws4): ruff format slack_listener, hook, test (CI ruff format --check)
This commit is contained in:
parent
2aa72d73f4
commit
83d17a5b7e
3 changed files with 34 additions and 9 deletions
|
|
@ -323,7 +323,9 @@ class SlackListener:
|
|||
loop stays alive.
|
||||
"""
|
||||
if self._new_task_callback is None:
|
||||
_LOG.debug("/new-task received but no new_task_callback configured; ignoring")
|
||||
_LOG.debug(
|
||||
"/new-task received but no new_task_callback configured; ignoring"
|
||||
)
|
||||
return None
|
||||
|
||||
text = str(raw_payload.get("text") or "").strip()
|
||||
|
|
|
|||
|
|
@ -67,7 +67,9 @@ def _load_hook_module():
|
|||
|
||||
|
||||
def test_plugin_root_exists() -> None:
|
||||
assert _PLUGIN_ROOT.is_dir(), f"sea-haven-claude-plugin/ not found at {_PLUGIN_ROOT}"
|
||||
assert _PLUGIN_ROOT.is_dir(), (
|
||||
f"sea-haven-claude-plugin/ not found at {_PLUGIN_ROOT}"
|
||||
)
|
||||
|
||||
|
||||
def test_claude_md_exists_and_has_required_sections() -> None:
|
||||
|
|
@ -80,7 +82,9 @@ def test_claude_md_exists_and_has_required_sections() -> None:
|
|||
|
||||
def test_settings_template_exists_and_is_valid_json() -> None:
|
||||
tmpl = _PLUGIN_ROOT / "settings.template.json"
|
||||
assert tmpl.is_file(), "settings.template.json must exist in sea-haven-claude-plugin/"
|
||||
assert tmpl.is_file(), (
|
||||
"settings.template.json must exist in sea-haven-claude-plugin/"
|
||||
)
|
||||
data = json.loads(tmpl.read_text(encoding="utf-8"))
|
||||
assert "hooks" in data, "settings.template.json must have a 'hooks' key"
|
||||
assert "UserPromptSubmit" in data["hooks"], (
|
||||
|
|
@ -107,8 +111,10 @@ def _make_listener(
|
|||
|
||||
if db_path is None:
|
||||
import tempfile
|
||||
|
||||
db_path = Path(tempfile.mkdtemp()) / "agent-team.db"
|
||||
from agent_team.db.schema import init_db
|
||||
|
||||
init_db(db_path)
|
||||
|
||||
transport = SlackTransport(channel="C_FAKE")
|
||||
|
|
@ -121,7 +127,9 @@ def _make_listener(
|
|||
)
|
||||
|
||||
|
||||
def _new_task_payload(text: str = "Add OAuth", user_id: str = _OWNER_ID) -> dict[str, Any]:
|
||||
def _new_task_payload(
|
||||
text: str = "Add OAuth", user_id: str = _OWNER_ID
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "slash_commands",
|
||||
"command": "/new-task",
|
||||
|
|
@ -135,7 +143,12 @@ def test_is_new_task_command_detects_slash_new_task() -> None:
|
|||
|
||||
|
||||
def test_is_new_task_command_ignores_other_commands() -> None:
|
||||
payload = {"type": "slash_commands", "command": "/other", "text": "foo", "user_id": "U1"}
|
||||
payload = {
|
||||
"type": "slash_commands",
|
||||
"command": "/other",
|
||||
"text": "foo",
|
||||
"user_id": "U1",
|
||||
}
|
||||
assert not _is_new_task_command(payload)
|
||||
|
||||
|
||||
|
|
@ -291,9 +304,14 @@ def test_delegate_empty_task_returns_usage_hint(hook: Any) -> None:
|
|||
|
||||
def test_missing_token_returns_block_with_hint(hook: Any) -> None:
|
||||
import os
|
||||
env_without_token = {k: v for k, v in os.environ.items() if k != "AGENT_TEAM_API_TOKEN"}
|
||||
|
||||
env_without_token = {
|
||||
k: v for k, v in os.environ.items() if k != "AGENT_TEAM_API_TOKEN"
|
||||
}
|
||||
with patch.dict(os.environ, env_without_token, clear=True):
|
||||
result = hook.run("/delegate Some task", api_url="http://localhost:8765", token="")
|
||||
result = hook.run(
|
||||
"/delegate Some task", api_url="http://localhost:8765", token=""
|
||||
)
|
||||
assert result.get("action") == "block"
|
||||
assert "AGENT_TEAM_API_TOKEN" in result["reason"]
|
||||
|
||||
|
|
@ -333,7 +351,10 @@ def test_api_network_error_returns_block(hook: Any) -> None:
|
|||
)
|
||||
|
||||
assert result.get("action") == "block"
|
||||
assert "run-team.py serve" in result["reason"] or "Connection refused" in result["reason"]
|
||||
assert (
|
||||
"run-team.py serve" in result["reason"]
|
||||
or "Connection refused" in result["reason"]
|
||||
)
|
||||
|
||||
|
||||
def test_unexpected_exception_returns_block(hook: Any) -> None:
|
||||
|
|
|
|||
|
|
@ -75,7 +75,9 @@ def _call_api(task: str, *, api_url: str, token: str) -> dict[str, Any]:
|
|||
return json.loads(resp.read().decode("utf-8"))
|
||||
|
||||
|
||||
def run(prompt: str, *, api_url: str = _DEFAULT_API_URL, token: str = "") -> dict[str, Any]:
|
||||
def run(
|
||||
prompt: str, *, api_url: str = _DEFAULT_API_URL, token: str = ""
|
||||
) -> dict[str, Any]:
|
||||
"""Core hook logic — pure function, fully testable without stdin/stdout.
|
||||
|
||||
Returns the hook output dict:
|
||||
|
|
|
|||
Reference in a new issue