test_harmony_cli_phase4.py
python
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
131 days ago
| 1 | """Tests for Phase 4 CLI additions to ``muse harmony``. |
| 2 | |
| 3 | New subcommands: |
| 4 | ``muse harmony escalate <pattern_id>`` — record an escalation |
| 5 | ``muse harmony escalations`` — list escalations |
| 6 | ``muse harmony resolve-escalation <esc_id>`` — close an escalation |
| 7 | ``muse harmony engine … --auto-escalate`` — engine auto-records on Tier 4 |
| 8 | |
| 9 | Coverage tiers |
| 10 | -------------- |
| 11 | I Unit — TypedDicts for escalate / escalations / resolve-escalation |
| 12 | II Success — escalate recorded; escalations list; resolve-escalation closes |
| 13 | III Errors — invalid IDs; missing records; resolution-id required |
| 14 | IV E2E — engine --auto-escalate → escalations; full lifecycle |
| 15 | V Integrity — all JSON fields always present; timestamps present |
| 16 | VI Security — path-traversal IDs rejected |
| 17 | VII Perf — all subcommands <300 ms |
| 18 | """ |
| 19 | from __future__ import annotations |
| 20 | from collections.abc import Mapping |
| 21 | |
| 22 | from muse.core._types import fake_id |
| 23 | import json |
| 24 | import pathlib |
| 25 | import time |
| 26 | import typing |
| 27 | |
| 28 | import pytest |
| 29 | |
| 30 | from tests.cli_test_helper import CliRunner |
| 31 | |
| 32 | runner = CliRunner() |
| 33 | |
| 34 | |
| 35 | # --------------------------------------------------------------------------- |
| 36 | # Helpers |
| 37 | # --------------------------------------------------------------------------- |
| 38 | |
| 39 | |
| 40 | |
| 41 | @pytest.fixture() |
| 42 | def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 43 | muse_dir = tmp_path / ".muse" |
| 44 | muse_dir.mkdir() |
| 45 | (muse_dir / "config.toml").write_text('[repo]\nname = "test"\nid = "abc"\n') |
| 46 | monkeypatch.chdir(tmp_path) |
| 47 | return tmp_path |
| 48 | |
| 49 | |
| 50 | def _record_pattern( |
| 51 | path: str = "track.mid", |
| 52 | domain: str = "midi", |
| 53 | conflict_type: str = "content", |
| 54 | ours: str = "ours", |
| 55 | theirs: str = "theirs", |
| 56 | ) -> str: |
| 57 | r = runner.invoke(None, [ |
| 58 | "harmony", "record", |
| 59 | "--path", path, "--domain", domain, |
| 60 | "--conflict-type", conflict_type, |
| 61 | "--ours-id", fake_id(ours), |
| 62 | "--theirs-id", fake_id(theirs), |
| 63 | "--json", |
| 64 | ]) |
| 65 | assert r.exit_code == 0, r.output |
| 66 | return json.loads(r.output)["pattern_id"] |
| 67 | |
| 68 | |
| 69 | def _save_resolution(pattern_id: str, confidence: str = "0.9") -> str: |
| 70 | r = runner.invoke(None, [ |
| 71 | "harmony", "resolve", |
| 72 | "--pattern-id", pattern_id, |
| 73 | "--strategy", "manual", |
| 74 | "--outcome-blob", fake_id("outcome"), |
| 75 | "--confidence", confidence, |
| 76 | "--json", |
| 77 | ]) |
| 78 | assert r.exit_code == 0, r.output |
| 79 | return json.loads(r.output)["resolution_id"] |
| 80 | |
| 81 | |
| 82 | def _escalate(pattern_id: str, reason: str = "No match found") -> str: |
| 83 | r = runner.invoke(None, [ |
| 84 | "harmony", "escalate", pattern_id, |
| 85 | "--reason", reason, |
| 86 | "--json", |
| 87 | ]) |
| 88 | assert r.exit_code == 0, r.output |
| 89 | return json.loads(r.output)["escalation_id"] |
| 90 | |
| 91 | |
| 92 | # =========================================================================== |
| 93 | # Tier I — Unit: TypedDict schemas |
| 94 | # =========================================================================== |
| 95 | |
| 96 | |
| 97 | class TestTypedDictSchemas: |
| 98 | """I: new TypedDicts declare expected keys.""" |
| 99 | |
| 100 | def _hints(self, name: str) -> Mapping[str, object]: |
| 101 | import muse.cli.commands.harmony as h |
| 102 | td = getattr(h, name) |
| 103 | return typing.get_type_hints(td) |
| 104 | |
| 105 | def test_escalate_json_has_escalation_id(self) -> None: |
| 106 | assert "escalation_id" in self._hints("_HarmonyEscalateJson") |
| 107 | |
| 108 | def test_escalate_json_has_pattern_id(self) -> None: |
| 109 | assert "pattern_id" in self._hints("_HarmonyEscalateJson") |
| 110 | |
| 111 | def test_escalate_json_has_already_existed(self) -> None: |
| 112 | assert "already_existed" in self._hints("_HarmonyEscalateJson") |
| 113 | |
| 114 | def test_escalation_entry_has_status(self) -> None: |
| 115 | assert "status" in self._hints("_HarmonyEscalationEntryJson") |
| 116 | |
| 117 | def test_escalation_entry_has_escalation_id(self) -> None: |
| 118 | assert "escalation_id" in self._hints("_HarmonyEscalationEntryJson") |
| 119 | |
| 120 | def test_escalation_entry_has_pattern_id(self) -> None: |
| 121 | assert "pattern_id" in self._hints("_HarmonyEscalationEntryJson") |
| 122 | |
| 123 | def test_escalations_json_has_total(self) -> None: |
| 124 | assert "total" in self._hints("_HarmonyEscalationsJson") |
| 125 | |
| 126 | def test_escalations_json_has_escalations(self) -> None: |
| 127 | assert "escalations" in self._hints("_HarmonyEscalationsJson") |
| 128 | |
| 129 | def test_resolve_escalation_json_has_escalation_id(self) -> None: |
| 130 | assert "escalation_id" in self._hints("_HarmonyResolveEscalationJson") |
| 131 | |
| 132 | def test_resolve_escalation_json_has_resolved(self) -> None: |
| 133 | assert "resolved" in self._hints("_HarmonyResolveEscalationJson") |
| 134 | |
| 135 | |
| 136 | class TestRegistration: |
| 137 | """I: new subcommands are reachable.""" |
| 138 | |
| 139 | def test_escalate_help(self, repo: pathlib.Path) -> None: |
| 140 | r = runner.invoke(None, ["harmony", "escalate", "--help"]) |
| 141 | assert r.exit_code == 0 |
| 142 | |
| 143 | def test_escalations_help(self, repo: pathlib.Path) -> None: |
| 144 | r = runner.invoke(None, ["harmony", "escalations", "--help"]) |
| 145 | assert r.exit_code == 0 |
| 146 | |
| 147 | def test_resolve_escalation_help(self, repo: pathlib.Path) -> None: |
| 148 | r = runner.invoke(None, ["harmony", "resolve-escalation", "--help"]) |
| 149 | assert r.exit_code == 0 |
| 150 | |
| 151 | |
| 152 | # =========================================================================== |
| 153 | # Tier II — Integration: success paths |
| 154 | # =========================================================================== |
| 155 | |
| 156 | |
| 157 | class TestEscalateSuccess: |
| 158 | """II: muse harmony escalate — success paths.""" |
| 159 | |
| 160 | def test_escalate_returns_escalation_id(self, repo: pathlib.Path) -> None: |
| 161 | pid = _record_pattern() |
| 162 | r = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 163 | assert r.exit_code == 0 |
| 164 | data = json.loads(r.output) |
| 165 | assert "escalation_id" in data |
| 166 | assert data["escalation_id"].startswith("sha256:") |
| 167 | |
| 168 | def test_escalate_returns_pattern_id(self, repo: pathlib.Path) -> None: |
| 169 | pid = _record_pattern() |
| 170 | r = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 171 | data = json.loads(r.output) |
| 172 | assert data["pattern_id"] == pid |
| 173 | |
| 174 | def test_escalate_already_existed_false_on_first(self, repo: pathlib.Path) -> None: |
| 175 | pid = _record_pattern() |
| 176 | r = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 177 | assert json.loads(r.output)["already_existed"] is False |
| 178 | |
| 179 | def test_escalate_idempotent_already_existed_true(self, repo: pathlib.Path) -> None: |
| 180 | pid = _record_pattern() |
| 181 | runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 182 | r2 = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 183 | assert r2.exit_code == 0 |
| 184 | assert json.loads(r2.output)["already_existed"] is True |
| 185 | |
| 186 | def test_escalate_same_id_both_calls(self, repo: pathlib.Path) -> None: |
| 187 | pid = _record_pattern() |
| 188 | r1 = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 189 | r2 = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 190 | eid1 = json.loads(r1.output)["escalation_id"] |
| 191 | eid2 = json.loads(r2.output)["escalation_id"] |
| 192 | assert eid1 == eid2 |
| 193 | |
| 194 | def test_escalate_custom_reason(self, repo: pathlib.Path) -> None: |
| 195 | pid = _record_pattern() |
| 196 | r = runner.invoke(None, [ |
| 197 | "harmony", "escalate", pid, |
| 198 | "--reason", "Custom escalation reason", |
| 199 | "--json", |
| 200 | ]) |
| 201 | assert r.exit_code == 0 |
| 202 | assert "escalation_id" in json.loads(r.output) |
| 203 | |
| 204 | def test_escalate_text_output(self, repo: pathlib.Path) -> None: |
| 205 | pid = _record_pattern() |
| 206 | r = runner.invoke(None, ["harmony", "escalate", pid]) |
| 207 | assert r.exit_code == 0 |
| 208 | assert pid[:12] in r.output |
| 209 | |
| 210 | def test_escalate_unknown_pattern_still_works(self, repo: pathlib.Path) -> None: |
| 211 | """Escalation can reference a pattern not in the store.""" |
| 212 | unknown_pid = fake_id("unknown-pattern") |
| 213 | r = runner.invoke(None, ["harmony", "escalate", unknown_pid, "--json"]) |
| 214 | assert r.exit_code == 0 |
| 215 | assert json.loads(r.output)["pattern_id"] == unknown_pid |
| 216 | |
| 217 | def test_escalate_with_agent_id(self, repo: pathlib.Path) -> None: |
| 218 | pid = _record_pattern() |
| 219 | r = runner.invoke(None, [ |
| 220 | "harmony", "escalate", pid, |
| 221 | "--agent-id", "claude-code", |
| 222 | "--json", |
| 223 | ]) |
| 224 | assert r.exit_code == 0 |
| 225 | |
| 226 | |
| 227 | class TestEscalationsSuccess: |
| 228 | """II: muse harmony escalations — success paths.""" |
| 229 | |
| 230 | def test_empty_store_returns_zero(self, repo: pathlib.Path) -> None: |
| 231 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 232 | assert r.exit_code == 0 |
| 233 | data = json.loads(r.output) |
| 234 | assert data["total"] == 0 |
| 235 | assert data["escalations"] == [] |
| 236 | |
| 237 | def test_lists_all_by_default(self, repo: pathlib.Path) -> None: |
| 238 | pid1, pid2 = _record_pattern("a.mid"), _record_pattern("b.mid", ours="b_ours", theirs="b_theirs") |
| 239 | _escalate(pid1) |
| 240 | _escalate(pid2) |
| 241 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 242 | assert r.exit_code == 0 |
| 243 | data = json.loads(r.output) |
| 244 | assert data["total"] == 2 |
| 245 | |
| 246 | def test_filter_open(self, repo: pathlib.Path) -> None: |
| 247 | pid1 = _record_pattern() |
| 248 | pid2 = _record_pattern("b.mid", ours="bo", theirs="bt") |
| 249 | eid1 = _escalate(pid1) |
| 250 | _escalate(pid2) |
| 251 | # resolve first |
| 252 | res_id = _save_resolution(pid1) |
| 253 | runner.invoke(None, [ |
| 254 | "harmony", "resolve-escalation", eid1, |
| 255 | "--resolution-id", res_id, |
| 256 | "--json", |
| 257 | ]) |
| 258 | r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"]) |
| 259 | data = json.loads(r.output) |
| 260 | assert data["total"] == 1 |
| 261 | |
| 262 | def test_filter_resolved(self, repo: pathlib.Path) -> None: |
| 263 | pid = _record_pattern() |
| 264 | eid = _escalate(pid) |
| 265 | res_id = _save_resolution(pid) |
| 266 | runner.invoke(None, [ |
| 267 | "harmony", "resolve-escalation", eid, |
| 268 | "--resolution-id", res_id, "--json", |
| 269 | ]) |
| 270 | r = runner.invoke(None, ["harmony", "escalations", "--status", "resolved", "--json"]) |
| 271 | assert json.loads(r.output)["total"] == 1 |
| 272 | |
| 273 | def test_text_output(self, repo: pathlib.Path) -> None: |
| 274 | pid = _record_pattern() |
| 275 | _escalate(pid) |
| 276 | r = runner.invoke(None, ["harmony", "escalations"]) |
| 277 | assert r.exit_code == 0 |
| 278 | |
| 279 | |
| 280 | class TestResolveEscalationSuccess: |
| 281 | """II: muse harmony resolve-escalation — success paths.""" |
| 282 | |
| 283 | def test_resolved_true_when_found(self, repo: pathlib.Path) -> None: |
| 284 | pid = _record_pattern() |
| 285 | eid = _escalate(pid) |
| 286 | res_id = _save_resolution(pid) |
| 287 | r = runner.invoke(None, [ |
| 288 | "harmony", "resolve-escalation", eid, |
| 289 | "--resolution-id", res_id, "--json", |
| 290 | ]) |
| 291 | assert r.exit_code == 0 |
| 292 | data = json.loads(r.output) |
| 293 | assert data["resolved"] is True |
| 294 | |
| 295 | def test_resolve_escalation_id_in_response(self, repo: pathlib.Path) -> None: |
| 296 | pid = _record_pattern() |
| 297 | eid = _escalate(pid) |
| 298 | res_id = _save_resolution(pid) |
| 299 | r = runner.invoke(None, [ |
| 300 | "harmony", "resolve-escalation", eid, |
| 301 | "--resolution-id", res_id, "--json", |
| 302 | ]) |
| 303 | assert json.loads(r.output)["escalation_id"] == eid |
| 304 | |
| 305 | def test_resolved_false_when_not_found(self, repo: pathlib.Path) -> None: |
| 306 | eid = fake_id("missing-esc") |
| 307 | res_id = fake_id("res") |
| 308 | r = runner.invoke(None, [ |
| 309 | "harmony", "resolve-escalation", eid, |
| 310 | "--resolution-id", res_id, "--json", |
| 311 | ]) |
| 312 | assert r.exit_code == 0 |
| 313 | assert json.loads(r.output)["resolved"] is False |
| 314 | |
| 315 | def test_text_output(self, repo: pathlib.Path) -> None: |
| 316 | pid = _record_pattern() |
| 317 | eid = _escalate(pid) |
| 318 | res_id = _save_resolution(pid) |
| 319 | r = runner.invoke(None, [ |
| 320 | "harmony", "resolve-escalation", eid, |
| 321 | "--resolution-id", res_id, |
| 322 | ]) |
| 323 | assert r.exit_code == 0 |
| 324 | |
| 325 | |
| 326 | class TestEngineAutoEscalate: |
| 327 | """II: engine --auto-escalate creates escalation record on Tier 4.""" |
| 328 | |
| 329 | def test_auto_escalate_creates_record(self, repo: pathlib.Path) -> None: |
| 330 | pid = _record_pattern() |
| 331 | runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate", "--json"]) |
| 332 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 333 | data = json.loads(r.output) |
| 334 | assert data["total"] >= 1 |
| 335 | |
| 336 | def test_auto_escalate_only_on_escalated(self, repo: pathlib.Path) -> None: |
| 337 | """If the engine resolves (applied), no escalation record is created.""" |
| 338 | pid = _record_pattern() |
| 339 | _save_resolution(pid, confidence="0.95") |
| 340 | runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate", "--json"]) |
| 341 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 342 | assert json.loads(r.output)["total"] == 0 |
| 343 | |
| 344 | def test_auto_escalate_pattern_id_in_record(self, repo: pathlib.Path) -> None: |
| 345 | pid = _record_pattern() |
| 346 | runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate"]) |
| 347 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 348 | recs = json.loads(r.output)["escalations"] |
| 349 | assert any(e["pattern_id"] == pid for e in recs) |
| 350 | |
| 351 | def test_engine_without_flag_no_escalation_record(self, repo: pathlib.Path) -> None: |
| 352 | """Without --auto-escalate, escalate tier writes audit but NOT a record.""" |
| 353 | pid = _record_pattern() |
| 354 | runner.invoke(None, ["harmony", "engine", pid, "--json"]) |
| 355 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 356 | assert json.loads(r.output)["total"] == 0 |
| 357 | |
| 358 | |
| 359 | # =========================================================================== |
| 360 | # Tier III — Error paths |
| 361 | # =========================================================================== |
| 362 | |
| 363 | |
| 364 | class TestEscalateErrors: |
| 365 | """III: muse harmony escalate — error paths.""" |
| 366 | |
| 367 | def test_invalid_id_exits_1(self, repo: pathlib.Path) -> None: |
| 368 | r = runner.invoke(None, ["harmony", "escalate", "bad-id", "--json"]) |
| 369 | assert r.exit_code == 1 |
| 370 | |
| 371 | def test_traversal_id_exits_1(self, repo: pathlib.Path) -> None: |
| 372 | r = runner.invoke(None, ["harmony", "escalate", "../../evil", "--json"]) |
| 373 | assert r.exit_code == 1 |
| 374 | |
| 375 | |
| 376 | class TestResolveEscalationErrors: |
| 377 | """III: muse harmony resolve-escalation — error paths.""" |
| 378 | |
| 379 | def test_invalid_escalation_id_exits_1(self, repo: pathlib.Path) -> None: |
| 380 | r = runner.invoke(None, [ |
| 381 | "harmony", "resolve-escalation", "bad-id", |
| 382 | "--resolution-id", fake_id("r"), "--json", |
| 383 | ]) |
| 384 | assert r.exit_code == 1 |
| 385 | |
| 386 | def test_invalid_resolution_id_exits_1(self, repo: pathlib.Path) -> None: |
| 387 | eid = fake_id("esc") |
| 388 | r = runner.invoke(None, [ |
| 389 | "harmony", "resolve-escalation", eid, |
| 390 | "--resolution-id", "bad-res", "--json", |
| 391 | ]) |
| 392 | assert r.exit_code == 1 |
| 393 | |
| 394 | def test_missing_resolution_id_flag_exits_non_zero(self, repo: pathlib.Path) -> None: |
| 395 | eid = fake_id("esc") |
| 396 | r = runner.invoke(None, ["harmony", "resolve-escalation", eid, "--json"]) |
| 397 | assert r.exit_code != 0 |
| 398 | |
| 399 | |
| 400 | # =========================================================================== |
| 401 | # Tier IV — End-to-end |
| 402 | # =========================================================================== |
| 403 | |
| 404 | |
| 405 | class TestEndToEnd: |
| 406 | """IV: Full lifecycle via CLI.""" |
| 407 | |
| 408 | def test_escalate_resolve_escalation_audit(self, repo: pathlib.Path) -> None: |
| 409 | pid = _record_pattern() |
| 410 | eid = _escalate(pid) |
| 411 | |
| 412 | # Verify escalation is open |
| 413 | r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"]) |
| 414 | assert json.loads(r.output)["total"] == 1 |
| 415 | |
| 416 | # Save a resolution |
| 417 | res_id = _save_resolution(pid) |
| 418 | |
| 419 | # Resolve the escalation |
| 420 | r = runner.invoke(None, [ |
| 421 | "harmony", "resolve-escalation", eid, |
| 422 | "--resolution-id", res_id, "--json", |
| 423 | ]) |
| 424 | assert json.loads(r.output)["resolved"] is True |
| 425 | |
| 426 | # Now zero open escalations |
| 427 | r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"]) |
| 428 | assert json.loads(r.output)["total"] == 0 |
| 429 | |
| 430 | # Audit log has escalation_resolved event |
| 431 | r = runner.invoke(None, ["harmony", "audit", "--json"]) |
| 432 | event_types = [e["event_type"] for e in json.loads(r.output)["entries"]] |
| 433 | assert "escalation_resolved" in event_types |
| 434 | |
| 435 | def test_engine_auto_escalate_then_resolve(self, repo: pathlib.Path) -> None: |
| 436 | pid = _record_pattern() |
| 437 | runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate"]) |
| 438 | |
| 439 | r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"]) |
| 440 | open_recs = json.loads(r.output)["escalations"] |
| 441 | assert len(open_recs) >= 1 |
| 442 | eid = open_recs[0]["escalation_id"] |
| 443 | |
| 444 | res_id = _save_resolution(pid) |
| 445 | r = runner.invoke(None, [ |
| 446 | "harmony", "resolve-escalation", eid, |
| 447 | "--resolution-id", res_id, "--json", |
| 448 | ]) |
| 449 | assert json.loads(r.output)["resolved"] is True |
| 450 | |
| 451 | def test_multiple_escalate_filter_independently(self, repo: pathlib.Path) -> None: |
| 452 | pids = [_record_pattern(f"{i}.mid", ours=f"o{i}", theirs=f"t{i}") for i in range(4)] |
| 453 | eids = [_escalate(p) for p in pids] |
| 454 | |
| 455 | # Resolve first two |
| 456 | for i, (pid, eid) in enumerate(zip(pids[:2], eids[:2])): |
| 457 | res_id = _save_resolution(pid, confidence=f"0.{80 + i}") |
| 458 | runner.invoke(None, [ |
| 459 | "harmony", "resolve-escalation", eid, |
| 460 | "--resolution-id", res_id, "--json", |
| 461 | ]) |
| 462 | |
| 463 | r_open = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"]) |
| 464 | r_resolved = runner.invoke(None, ["harmony", "escalations", "--status", "resolved", "--json"]) |
| 465 | assert json.loads(r_open.output)["total"] == 2 |
| 466 | assert json.loads(r_resolved.output)["total"] == 2 |
| 467 | |
| 468 | |
| 469 | # =========================================================================== |
| 470 | # Tier V — Data integrity |
| 471 | # =========================================================================== |
| 472 | |
| 473 | |
| 474 | class TestDataIntegrity: |
| 475 | """V: All JSON fields always present; types correct.""" |
| 476 | |
| 477 | def test_escalate_all_fields_present(self, repo: pathlib.Path) -> None: |
| 478 | pid = _record_pattern() |
| 479 | r = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 480 | data = json.loads(r.output) |
| 481 | for field in ("escalation_id", "pattern_id", "already_existed"): |
| 482 | assert field in data, f"missing: {field}" |
| 483 | |
| 484 | def test_escalations_entry_all_fields_present(self, repo: pathlib.Path) -> None: |
| 485 | pid = _record_pattern() |
| 486 | _escalate(pid) |
| 487 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 488 | entry = json.loads(r.output)["escalations"][0] |
| 489 | for field in ("escalation_id", "pattern_id", "reason", "status", |
| 490 | "escalated_at", "escalated_by"): |
| 491 | assert field in entry, f"missing: {field}" |
| 492 | |
| 493 | def test_escalation_id_is_prefixed_sha256(self, repo: pathlib.Path) -> None: |
| 494 | pid = _record_pattern() |
| 495 | r = runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 496 | eid = json.loads(r.output)["escalation_id"] |
| 497 | assert eid.startswith("sha256:") |
| 498 | assert len(eid) == 71 # "sha256:" + 64 hex chars |
| 499 | assert all(c in "0123456789abcdef" for c in eid[7:]) |
| 500 | |
| 501 | def test_resolve_escalation_all_fields_present(self, repo: pathlib.Path) -> None: |
| 502 | pid = _record_pattern() |
| 503 | eid = _escalate(pid) |
| 504 | res_id = _save_resolution(pid) |
| 505 | r = runner.invoke(None, [ |
| 506 | "harmony", "resolve-escalation", eid, |
| 507 | "--resolution-id", res_id, "--json", |
| 508 | ]) |
| 509 | data = json.loads(r.output) |
| 510 | for field in ("escalation_id", "resolved"): |
| 511 | assert field in data, f"missing: {field}" |
| 512 | |
| 513 | def test_escalations_empty_list_not_null(self, repo: pathlib.Path) -> None: |
| 514 | r = runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 515 | data = json.loads(r.output) |
| 516 | assert isinstance(data["escalations"], list) |
| 517 | |
| 518 | |
| 519 | # =========================================================================== |
| 520 | # Tier VI — Security |
| 521 | # =========================================================================== |
| 522 | |
| 523 | |
| 524 | class TestSecurity: |
| 525 | """VI: Path-traversal IDs rejected at all Phase 4 entry points.""" |
| 526 | |
| 527 | def test_escalate_traversal_rejected(self, repo: pathlib.Path) -> None: |
| 528 | r = runner.invoke(None, ["harmony", "escalate", "../../evil", "--json"]) |
| 529 | assert r.exit_code == 1 |
| 530 | |
| 531 | def test_escalate_null_byte_rejected(self, repo: pathlib.Path) -> None: |
| 532 | r = runner.invoke(None, ["harmony", "escalate", "a" * 63 + "\x00", "--json"]) |
| 533 | assert r.exit_code == 1 |
| 534 | |
| 535 | def test_resolve_escalation_traversal_eid(self, repo: pathlib.Path) -> None: |
| 536 | r = runner.invoke(None, [ |
| 537 | "harmony", "resolve-escalation", "../../evil", |
| 538 | "--resolution-id", fake_id("r"), "--json", |
| 539 | ]) |
| 540 | assert r.exit_code == 1 |
| 541 | |
| 542 | def test_resolve_escalation_traversal_res_id(self, repo: pathlib.Path) -> None: |
| 543 | eid = fake_id("esc") |
| 544 | r = runner.invoke(None, [ |
| 545 | "harmony", "resolve-escalation", eid, |
| 546 | "--resolution-id", "../../evil", "--json", |
| 547 | ]) |
| 548 | assert r.exit_code == 1 |
| 549 | |
| 550 | |
| 551 | # =========================================================================== |
| 552 | # Tier VII — Performance |
| 553 | # =========================================================================== |
| 554 | |
| 555 | |
| 556 | class TestPerformance: |
| 557 | """VII: all Phase 4 subcommands <300 ms.""" |
| 558 | |
| 559 | def test_escalate_under_300ms(self, repo: pathlib.Path) -> None: |
| 560 | pid = _record_pattern() |
| 561 | start = time.monotonic() |
| 562 | runner.invoke(None, ["harmony", "escalate", pid, "--json"]) |
| 563 | elapsed = (time.monotonic() - start) * 1000 |
| 564 | assert elapsed < 300, f"escalate took {elapsed:.0f}ms" |
| 565 | |
| 566 | def test_escalations_under_300ms(self, repo: pathlib.Path) -> None: |
| 567 | start = time.monotonic() |
| 568 | runner.invoke(None, ["harmony", "escalations", "--json"]) |
| 569 | elapsed = (time.monotonic() - start) * 1000 |
| 570 | assert elapsed < 300, f"escalations took {elapsed:.0f}ms" |
| 571 | |
| 572 | def test_resolve_escalation_under_300ms(self, repo: pathlib.Path) -> None: |
| 573 | pid = _record_pattern() |
| 574 | eid = _escalate(pid) |
| 575 | res_id = _save_resolution(pid) |
| 576 | start = time.monotonic() |
| 577 | runner.invoke(None, [ |
| 578 | "harmony", "resolve-escalation", eid, |
| 579 | "--resolution-id", res_id, "--json", |
| 580 | ]) |
| 581 | elapsed = (time.monotonic() - start) * 1000 |
| 582 | assert elapsed < 300, f"resolve-escalation took {elapsed:.0f}ms" |
File History
2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
131 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c
docs: docstring sprint for-each-ref→hotspots — idiomatic ru…
Sonnet 4.6
patch
137 days ago