← back to Terminal Status
test_owner_change_carry.py
558 lines
"""TK-11317 (Steve's TK-11620 guard ruling, 2026-09-26): when a live terminal's
stored record belongs to a PRIOR owner (tty reused by a new claude/codex pid, e.g.
after /res), the engine must never silently blank or green-floor a prior
needs-Steve colour. These tests exercise Store.prior_attention,
Store.settle_owner_change, the backfill/start wiring, row() rendering, and the
stickiness guarantees (deferential green, the `ticket` command, explicit set)
against a tempdir Store with a fake in-memory painter -- never a real tty, never
the shared ~/.local/state/abrams-terminal-status store.
"""
import dataclasses
import datetime as dt
import json
import os
import tempfile
import time
import unittest
from pathlib import Path
import terminal_status as ts
class OwnerChangeCarryTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.old_process = ts.Process(111, 1, "ttys050",
"Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
self.new_process = ts.Process(222, 1, "ttys050",
"Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
self.old_owner = self.old_process.owner()
self.new_owner = self.new_process.owner()
self.paints = []
self.rows = {222: self.new_process}
self.store = ts.Store(self.root, lambda **_: self.rows,
lambda owner, record: self.paints.append(record.copy()))
def write_legacy(self, runtime, title):
path = self.store.legacy[runtime] / (self.old_owner.tty + ".dot")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(title)
return path
def age_old_record(self, seconds_ago):
"""Rewrite the OLD owner's on-disk `updated_at` to look `seconds_ago` in
the past -- the freshness clock prior_attention() reads for a JSON
source -- AND age its legacy .dot mirrors to match (Store.set() always
writes ALL runtimes' mirrors, so a same-tty realistic "this leftover
state is N days old" scenario ages both together; aging only the JSON
would leave the still-fresh .dot mirror as an unaged fallback source,
silently defeating the point of the test). Bypasses Store.set() (which
always stamps "now") so tests can exercise CARRY_MAX_AGE_S deterministically."""
path = self.store.root / (self.old_owner.tty + ".json")
data = json.loads(path.read_text())
aged = dt.datetime.now(dt.timezone.utc) - dt.timedelta(seconds=seconds_ago)
data["updated_at"] = aged.isoformat()
path.write_text(json.dumps(data, indent=2) + "\n")
when = time.time() - seconds_ago
for directory in self.store.legacy.values():
dot_path = directory / (self.old_owner.tty + ".dot")
if dot_path.exists():
os.utime(dot_path, (when, when))
def age_legacy(self, path, seconds_ago):
when = time.time() - seconds_ago
os.utime(path, (when, when))
def make_old_record(self, color, label="reason"):
"""Write a canonical record for the OLD owner directly (no lock/assert_owner
contention with the new owner, since the fixture's process table only
contains the NEW owner -- exactly the real-world shape at the moment
owner_changed is first observed)."""
old_rows = {111: self.old_process}
old_store = ts.Store(self.root, lambda **_: old_rows,
lambda owner, record: None)
return old_store.set(self.old_owner, color, label, rows=old_rows)
# ---- prior_attention: JSON source ----------------------------------
def test_prior_purple_record_carried(self):
self.make_old_record("purple", "gated memo")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "purple")
self.assertEqual(prior["label"], "gated memo")
self.assertEqual(prior["ticket"], "")
# TK-11317 contrarian follow-up: carried_from/prior_revision provenance.
self.assertEqual(prior["carried_from"]["pid"], self.old_owner.pid)
self.assertEqual(prior["carried_from"]["started"], self.old_owner.started)
self.assertEqual(prior["carried_from"]["hops"], 1)
self.assertIsInstance(prior["carried_from"]["at"], (int, float))
self.assertEqual(prior["prior_revision"], 1)
def test_prior_orange_record_carried(self):
self.make_old_record("orange", "paste waiting")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "orange")
self.assertEqual(prior["label"], "paste waiting")
def test_prior_yellow_record_carried(self):
self.make_old_record("yellow", "which fork?")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "yellow")
self.assertEqual(prior["label"], "which fork?")
def test_prior_lightblue_record_carried(self):
self.make_old_record("lightblue", "needs Steve")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "lightblue")
self.assertEqual(prior["label"], "needs Steve")
def test_prior_record_ticket_is_carried_from_json_source(self):
self.make_old_record("purple", "TK-11317 · gated memo")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["ticket"], "TK-11317")
self.assertEqual(prior["label"], "gated memo")
# ---- prior_attention: legacy .dot source ----------------------------
def test_legacy_dot_only_attention_carried_when_json_prior_is_green(self):
self.make_old_record("green")
self.write_legacy("claude", "\U0001F7E3 TK-99999 · legacy gated only")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "purple")
self.assertEqual(prior["label"], "legacy gated only")
# The ticket is NEVER invented from a legacy-.dot-only carry (rule: the
# carried ticket is kept ONLY when the carried colour came from the JSON
# record itself).
self.assertEqual(prior["ticket"], "")
def test_legacy_dot_only_attention_carried_when_no_json_record_at_all(self):
# No JSON record was ever written for the old owner -- only a legacy mirror
# exists (the pre-migration shape this rollout still has to tolerate).
self.write_legacy("claude", "\U0001F7E0 waiting on a paste")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "orange")
self.assertEqual(prior["label"], "waiting on a paste")
self.assertEqual(prior["ticket"], "")
# ---- priority arbitration between the two sources --------------------
def test_priority_choice_when_json_and_legacy_disagree(self):
# JSON = purple (PRIORITY 2), legacy .dot = orange (PRIORITY 1) -> orange
# outranks purple, so orange wins even though it is NOT the JSON source.
self.make_old_record("purple", "gated memo")
self.write_legacy("claude", "\U0001F7E0 paste waiting")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "orange")
self.assertEqual(prior["label"], "paste waiting")
self.assertEqual(prior["ticket"], "") # legacy source -> no invented ticket
def test_priority_choice_lightblue_outranks_everything(self):
self.make_old_record("yellow", "which fork?")
self.write_legacy("claude", "\U0001F535 needs Steve")
prior = self.store.prior_attention(self.new_owner)
self.assertEqual(prior["color"], "lightblue")
# ---- no attention pending -> yellow status-needed --------------------
def test_prior_green_yields_none_and_settles_to_yellow_status_needed(self):
self.make_old_record("green")
self.assertIsNone(self.store.prior_attention(self.new_owner))
record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(record["state"], "yellow")
self.assertEqual(record["label"], ts.OWNER_STATUS_LABEL)
self.assertTrue(record["owner_status_unverified"])
self.assertNotIn("carried_from", record)
def test_prior_pink_yields_none_and_settles_to_yellow_status_needed(self):
self.make_old_record("pink")
self.assertIsNone(self.store.prior_attention(self.new_owner))
record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(record["state"], "yellow")
self.assertTrue(record["owner_status_unverified"])
def test_no_prior_record_at_all_settles_to_yellow_status_needed(self):
# Reason "missing" on a fresh owner is handled separately (green floor) --
# this is prior_attention()/settle_owner_change() called directly on a tty
# with NOTHING behind it, which must still never fabricate an attention
# colour out of nothing.
self.assertIsNone(self.store.prior_attention(self.new_owner))
record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(record["state"], "yellow")
self.assertTrue(record["owner_status_unverified"])
# ---- missing (no prior owner at all) keeps its plain green floor -----
def test_missing_reason_still_green_floors_via_backfill(self):
lone = ts.Process(333, 1, "ttys051", "Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
rows = {333: lone}
store = ts.Store(self.root, lambda **_: rows, lambda owner, record: None)
painted, floored, skipped = store.backfill(rows)
record, reason = store.load(lone.owner())
self.assertEqual(reason, "canonical")
self.assertEqual(record["state"], "green")
self.assertNotIn("owner_status_unverified", record)
self.assertEqual((painted, floored, skipped), (0, 1, 0))
# ---- carried_from provenance -----------------------------------------
def test_carried_from_records_prior_owner_identity(self):
self.make_old_record("purple", "gated memo")
record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(record["carried_from"]["pid"], self.old_owner.pid)
self.assertEqual(record["carried_from"]["started"], self.old_owner.started)
self.assertEqual(record["carried_from"]["hops"], 1)
self.assertIsInstance(record["carried_from"]["at"], (int, float))
def test_load_accepts_records_with_and_without_carried_from(self):
# load() must validate a record carrying the new optional fields exactly as
# readily as one that predates this change (neither field present).
self.store.set(self.new_owner, "green", rows=self.rows)
record, reason = self.store.load(self.new_owner)
self.assertEqual(reason, "canonical")
self.assertNotIn("carried_from", record)
self.make_old_record("purple", "gated memo")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
record2, reason2 = self.store.load(self.new_owner)
self.assertEqual(reason2, "canonical")
self.assertIn("carried_from", record2)
self.assertEqual(record2, settled)
# ---- row() rendering during the owner_changed transient window -------
def test_row_for_owner_changed_shows_carried_colour_not_none(self):
self.make_old_record("purple", "gated memo")
row = self.store.row(self.new_owner)
self.assertEqual(row["status_source"], "owner_changed")
self.assertEqual(row["color"], "purple")
self.assertIn("gated memo", row["label"])
self.assertIn("owner_status_unverified", row["warnings"])
def test_row_for_owner_changed_with_no_prior_attention_shows_yellow(self):
self.make_old_record("green")
row = self.store.row(self.new_owner)
self.assertEqual(row["status_source"], "owner_changed")
self.assertEqual(row["color"], "yellow")
self.assertIn(ts.OWNER_STATUS_LABEL, row["label"])
self.assertIn("owner_status_unverified", row["warnings"])
def test_row_never_renders_none_for_owner_changed(self):
for color in ("purple", "orange", "yellow", "lightblue", "green", "pink"):
with self.subTest(color=color):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
root = Path(temp.name)
old_rows = {111: self.old_process}
old_store = ts.Store(root, lambda **_: old_rows, lambda o, r: None)
old_store.set(self.old_owner, color, "some reason", rows=old_rows)
store = ts.Store(root, lambda **_: self.rows, lambda o, r: None)
row = store.row(self.new_owner)
self.assertNotEqual(row["color"], "none")
def test_row_for_canonical_owner_status_unverified_record_carries_warning(self):
record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(record["state"], "yellow")
row = self.store.row(self.new_owner)
self.assertEqual(row["status_source"], "canonical")
self.assertIn("owner_status_unverified", row["warnings"])
# ---- start command wiring ---------------------------------------------
def test_start_command_settles_owner_changed_via_carry(self):
self.make_old_record("orange", "paste waiting")
record, reason = self.store.load(self.new_owner)
self.assertIsNone(record)
self.assertEqual(reason, "owner_changed")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(settled["state"], "orange")
# ---- backfill wiring ----------------------------------------------------
def test_backfill_carries_purple_through_owner_changed(self):
self.make_old_record("purple", "gated memo")
painted, floored, skipped = self.store.backfill(self.rows)
self.assertEqual(floored, 1)
record, reason = self.store.load(self.new_owner)
self.assertEqual(reason, "canonical")
self.assertEqual(record["state"], "purple")
self.assertIn("carried_from", record)
def test_backfill_never_flattens_carried_attention_to_green(self):
self.make_old_record("lightblue", "needs Steve")
self.store.backfill(self.rows)
record, _ = self.store.load(self.new_owner)
self.assertNotEqual(record["state"], "green")
self.assertEqual(record["state"], "lightblue")
# ---- stickiness: deferential green preserves the carry -----------------
def test_deferential_green_keeps_carried_attention(self):
self.make_old_record("purple", "gated memo")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(settled["state"], "purple")
result = self.store.set(self.new_owner, "green", deferential=True, rows=self.rows)
self.assertEqual(result["state"], "purple")
self.assertEqual(result["carried_from"], settled["carried_from"])
self.assertEqual(result, settled)
def test_deferential_green_keeps_owner_status_unverified(self):
self.make_old_record("green")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(settled["state"], "yellow")
result = self.store.set(self.new_owner, "green", deferential=True, rows=self.rows)
self.assertEqual(result["state"], "yellow")
self.assertTrue(result["owner_status_unverified"])
self.assertEqual(result, settled)
# ---- stickiness: the `ticket` command's set() call shape ---------------
def test_ticket_binding_preserves_carried_attention_and_flag(self):
self.make_old_record("purple", "gated memo")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
previous, _ = self.store.read(self.new_owner)
# This mirrors EXACTLY the `ticket` command handler's store.set() call shape
# (terminal_status.py, `if args.command == "ticket":`).
record = self.store.set(
self.new_owner, previous["state"],
ts.label_ticket(previous["label"])[0],
ticket_update="TK-11317",
variant=previous.get("variant", ""),
carried_from=previous.get("carried_from"),
owner_status_unverified=previous.get("owner_status_unverified"),
rows=self.rows)
self.assertEqual(record["state"], "purple")
self.assertEqual(record["ticket"], "TK-11317")
self.assertEqual(record["carried_from"], settled["carried_from"])
def test_ticket_binding_preserves_owner_status_unverified(self):
self.make_old_record("green")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
previous, _ = self.store.read(self.new_owner)
record = self.store.set(
self.new_owner, previous["state"],
ts.label_ticket(previous["label"])[0],
ticket_update="TK-11317",
variant=previous.get("variant", ""),
carried_from=previous.get("carried_from"),
owner_status_unverified=previous.get("owner_status_unverified"),
rows=self.rows)
self.assertEqual(record["state"], "yellow")
self.assertTrue(record["owner_status_unverified"])
self.assertEqual(record["ticket"], "TK-11317")
# ---- explicit non-deferential set CLEARS the flag -----------------------
def test_explicit_set_green_clears_owner_status_unverified(self):
self.make_old_record("green")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertTrue(settled["owner_status_unverified"])
record = self.store.set(self.new_owner, "green", rows=self.rows)
self.assertNotIn("owner_status_unverified", record)
def test_explicit_set_green_clears_carried_from(self):
self.make_old_record("purple", "gated memo")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertIn("carried_from", settled)
record = self.store.set(self.new_owner, "green", rows=self.rows)
self.assertNotIn("carried_from", record)
# ---- old owner cannot write (existing owner check, still true) ---------
def test_old_owner_cannot_write_after_tty_reuse(self):
self.make_old_record("purple", "gated memo")
with self.assertRaises(ts.StatusError):
self.store.set(self.old_owner, "orange", "stale write", rows=self.rows)
# ---- CONTRARIAN REVIEW FOLLOW-UP (2026-09-26): CRITICAL hole #1 --------
# working-state.sh's per-prompt UserPromptSubmit hook fires
# `set green WORKING --deferential --boot` on EVERY prompt. Before this fix,
# set()'s deferential-yield guard required a truthy `previous`, but
# load(owner) returns None for reason owner_changed -- so the FIRST prompt
# after a tty reuse silently overwrote a carried purple/orange/yellow/
# lightblue with plain green. These tests are the exact repro.
def test_deferential_first_prompt_carries_purple_not_green(self):
self.make_old_record("purple", "gated memo")
record = self.store.set(self.new_owner, "green", "WORKING",
deferential=True, rows=self.rows)
self.assertEqual(record["state"], "purple")
self.assertIn("gated memo", record["label"])
self.assertIn("carried_from", record)
# The record actually ON DISK (not just the return value) is the carry.
on_disk, reason = self.store.load(self.new_owner)
self.assertEqual(reason, "canonical")
self.assertEqual(on_disk["state"], "purple")
def test_deferential_first_prompt_unverified_yellow_stays_yellow(self):
# Prior owner had nothing pending (plain green) -- the deferential green
# call must land on the yellow "New owner - status needed" floor, not
# silently succeed at plain green.
self.make_old_record("green")
record = self.store.set(self.new_owner, "green", "WORKING",
deferential=True, rows=self.rows)
self.assertEqual(record["state"], "yellow")
self.assertTrue(record["owner_status_unverified"])
on_disk, reason = self.store.load(self.new_owner)
self.assertEqual(on_disk["state"], "yellow")
def test_explicit_non_deferential_set_still_wins_over_owner_changed(self):
# "A NON-deferential explicit set by the new owner still wins" -- the
# presettle branch only engages for deferential=True. An explicit
# /greendot-style call is the new owner's own real status and must NOT
# be redirected into a carry.
self.make_old_record("purple", "gated memo")
record = self.store.set(self.new_owner, "green", "WORKING",
deferential=False, rows=self.rows)
self.assertEqual(record["state"], "green")
self.assertNotIn("carried_from", record)
# ---- CONTRARIAN REVIEW FOLLOW-UP: age ceiling + provenance --------------
def test_stale_prior_purple_does_not_carry_falls_to_yellow(self):
self.make_old_record("purple", "TK-11317 · gated memo")
self.age_old_record(21 * 86400) # 21 days -- far past the 24h ceiling
self.assertIsNone(self.store.prior_attention(self.new_owner))
record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(record["state"], "yellow")
self.assertTrue(record["owner_status_unverified"])
self.assertEqual(record["ticket"], "")
def test_fresh_prior_purple_just_under_ceiling_still_carries(self):
self.make_old_record("purple", "gated memo")
self.age_old_record(ts.CARRY_MAX_AGE_S - 60)
prior = self.store.prior_attention(self.new_owner)
self.assertIsNotNone(prior)
self.assertEqual(prior["color"], "purple")
def test_prior_just_over_ceiling_does_not_carry(self):
self.make_old_record("purple", "gated memo")
self.age_old_record(ts.CARRY_MAX_AGE_S + 60)
self.assertIsNone(self.store.prior_attention(self.new_owner))
def test_fresh_legacy_dot_carried(self):
self.write_legacy("claude", "\U0001F7E0 paste waiting") # freshly written = now
prior = self.store.prior_attention(self.new_owner)
self.assertIsNotNone(prior)
self.assertEqual(prior["color"], "orange")
def test_stale_legacy_dot_not_carried(self):
path = self.write_legacy("claude", "\U0001F7E0 paste waiting")
self.age_legacy(path, 21 * 86400)
self.assertIsNone(self.store.prior_attention(self.new_owner))
def test_three_generation_chain_keeps_original_carried_from_and_hops(self):
# Gen 1: owner A (old_owner) gets gated purple.
self.make_old_record("purple", "gated memo")
# Gen 2: owner B (new_owner, tty reused) settles, carrying A's attention.
gen2 = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(gen2["carried_from"]["pid"], self.old_owner.pid)
self.assertEqual(gen2["carried_from"]["hops"], 1)
# Gen 3: the SAME tty is reused again (owner B departs, owner C arrives).
gen3_process = ts.Process(333, 1, "ttys050", "Fri Sep 11 10:00:00 2026",
"/usr/bin/claude")
gen3_owner = gen3_process.owner()
gen3_rows = {333: gen3_process}
store3 = ts.Store(self.root, lambda **_: gen3_rows, lambda o, r: None)
prior = store3.prior_attention(gen3_owner)
self.assertIsNotNone(prior)
self.assertEqual(prior["color"], "purple")
# ORIGINAL provenance preserved -- pid/started/at are A's, not B's --
# only hops increments.
self.assertEqual(prior["carried_from"]["pid"], self.old_owner.pid)
self.assertEqual(prior["carried_from"]["started"], self.old_owner.started)
self.assertEqual(prior["carried_from"]["at"], gen2["carried_from"]["at"])
self.assertEqual(prior["carried_from"]["hops"], 2)
gen3 = store3.settle_owner_change(gen3_owner, rows=gen3_rows)
self.assertEqual(gen3["carried_from"]["pid"], self.old_owner.pid)
self.assertEqual(gen3["carried_from"]["started"], self.old_owner.started)
self.assertEqual(gen3["carried_from"]["hops"], 2)
self.assertEqual(gen3["revision"], 3) # 1 (gen1) -> 2 (gen2) -> 3 (gen3)
# ---- CONTRARIAN REVIEW FOLLOW-UP: revision continuity (LOW) -------------
def test_revision_continues_across_carry_not_reset_to_one(self):
old_rec = self.make_old_record("purple", "gated memo")
self.assertEqual(old_rec["revision"], 1)
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(settled["revision"], 2) # continues, not reset to 1
def test_revision_resets_to_one_when_nothing_carried(self):
self.make_old_record("green")
settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(settled["revision"], 1)
# ---- CONTRARIAN REVIEW FOLLOW-UP: automatic-writer audit -----------------
# The `ticket` command handler AND auto-bind's _default_ticket_set both did
# `previous, _ = store.read(owner)` and silently defaulted to color "none"
# whenever reason was actually owner_changed -- the SECOND identified hole.
# effective_previous() closes it for both call sites.
def test_effective_previous_returns_settlement_preview_for_owner_changed(self):
self.make_old_record("purple", "gated memo")
record, reason = self.store.load(self.new_owner)
self.assertIsNone(record)
self.assertEqual(reason, "owner_changed")
previous, eff_reason = self.store.effective_previous(self.new_owner)
self.assertEqual(eff_reason, "owner_changed")
self.assertEqual(previous["state"], "purple")
self.assertIn("carried_from", previous)
self.assertEqual(previous["next_revision"], 2)
def test_effective_previous_passes_through_canonical_and_missing(self):
self.store.set(self.new_owner, "green", rows=self.rows)
previous, reason = self.store.effective_previous(self.new_owner)
self.assertEqual(reason, "canonical")
self.assertEqual(previous["state"], "green")
lone = ts.Process(444, 1, "ttys077", "Sat Sep 12 08:00:00 2026", "/usr/bin/claude")
previous2, reason2 = self.store.effective_previous(lone.owner())
self.assertEqual(reason2, "missing")
self.assertIsNone(previous2)
def test_ticket_command_preserves_carry_when_settle_has_not_run_yet(self):
# Reproduces the SECOND identified hole directly: binding a ticket on a
# reused tty BEFORE backfill/the deferential hook ever settled it must
# not blank the carried colour to "none".
self.make_old_record("purple", "gated memo")
record, reason = self.store.load(self.new_owner)
self.assertIsNone(record)
self.assertEqual(reason, "owner_changed") # settle has NOT run yet
previous, eff_reason = self.store.effective_previous(self.new_owner)
self.assertEqual(eff_reason, "owner_changed")
bound = self.store.set(
self.new_owner, previous["state"],
ts.label_ticket(previous["label"])[0],
ticket_update="TK-22222",
variant=previous.get("variant", ""),
carried_from=previous.get("carried_from"),
owner_status_unverified=previous.get("owner_status_unverified"),
revision_override=previous.get("next_revision"),
rows=self.rows)
self.assertEqual(bound["state"], "purple")
self.assertEqual(bound["ticket"], "TK-22222")
self.assertIn("carried_from", bound)
self.assertEqual(bound["revision"], 2) # continued, not reset to 1
def test_default_ticket_set_preserves_carried_purple_on_owner_changed(self):
self.make_old_record("purple", "gated memo")
ts._default_ticket_set(self.store, self.new_owner, "TK-33333")
record, reason = self.store.load(self.new_owner)
self.assertEqual(reason, "canonical")
self.assertEqual(record["state"], "purple")
self.assertEqual(record["ticket"], "TK-33333")
self.assertIn("carried_from", record)
self.assertEqual(record["revision"], 2)
def test_default_ticket_set_floors_to_yellow_when_nothing_carried(self):
self.make_old_record("green")
ts._default_ticket_set(self.store, self.new_owner, "TK-44444")
record, reason = self.store.load(self.new_owner)
self.assertEqual(reason, "canonical")
self.assertEqual(record["state"], "yellow")
self.assertTrue(record["owner_status_unverified"])
self.assertEqual(record["ticket"], "TK-44444")
if __name__ == "__main__":
unittest.main()