tegwick
82cfe3da5a
Some checks failed
tamq-ci / test (push) Failing after 6s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc
52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
import json
|
|
|
|
from tamq.cli import main
|
|
from tamq.store import Store
|
|
|
|
|
|
def test_purge_dry_run_does_not_delete(tmp_path, monkeypatch, capsys):
|
|
state = tmp_path / "state"
|
|
monkeypatch.setenv("TAMQ_STATE_DIR", str(state))
|
|
store = Store(state / "tamq.sqlite3")
|
|
store.add("a", "b", "body")
|
|
store.close()
|
|
assert main(["purge"]) == 0
|
|
assert "Dry run" in capsys.readouterr().out
|
|
store = Store(state / "tamq.sqlite3")
|
|
assert len(store.list()) == 1
|
|
|
|
|
|
def test_purge_confirmed_deletes(tmp_path, monkeypatch, capsys):
|
|
state = tmp_path / "state"
|
|
monkeypatch.setenv("TAMQ_STATE_DIR", str(state))
|
|
store = Store(state / "tamq.sqlite3")
|
|
store.add("a", "b", "body")
|
|
store.db.execute("UPDATE messages SET created_at=created_at-400*86400")
|
|
store.db.commit()
|
|
store.close()
|
|
assert main(["purge", "--yes"]) == 0
|
|
assert "Purged" in capsys.readouterr().out
|
|
assert len(Store(state / "tamq.sqlite3").list()) == 0
|
|
|
|
|
|
def test_feedback_chain_purge_selects_only_reflected_direction(
|
|
tmp_path, monkeypatch, capsys
|
|
):
|
|
state = tmp_path / "state"
|
|
monkeypatch.setenv("TAMQ_STATE_DIR", str(state))
|
|
store = Store(state / "tamq.sqlite3")
|
|
root = store.add("a", "b", "body")
|
|
child = store.add("b", "a", f"body [{root}]")
|
|
grandchild = store.add("a", "b", f"body [{root}] [{child}]")
|
|
unrelated = store.add("a", "b", f"not reflected [{root}]")
|
|
store.close()
|
|
|
|
args = ["purge", "--feedback-chain", root]
|
|
assert main(args) == 0
|
|
assert "Dry run: 3 message(s)" in capsys.readouterr().out
|
|
assert len(Store(state / "tamq.sqlite3").list()) == 4
|
|
|
|
assert main([*args, "--yes"]) == 0
|
|
assert "Purged 3" in capsys.readouterr().out
|
|
remaining = Store(state / "tamq.sqlite3").list()
|
|
assert [row["message_id"] for row in remaining] == [unrelated]
|