stackchain-dashboard/tests/test_rotate_private_state.py

import base64
import json
import os
import sqlite3
import subprocess
import sys
from pathlib import Path

from src.completed_filed_review_store import CompletedFiledReviewStore
from src.queue_priority_store import DEFAULT_QUEUE_ORDER, QueuePriorityStore
from src.recent_work_store import RecentWorkStore
from src.saved_search_store import SavedSearchStore


ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "rotate_private_state.py"


def encoded(value: bytes) -> str:
    return base64.b64encode(value * 32).decode()


def run_rotation(state: Path, *, tamper: bool = False):
    env = {
        **os.environ,
        "STACKCHAIN_STATE_DIR": str(state),
        "STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEYS": json.dumps(
            {"old": encoded(b"o"), "next": encoded(b"n")}
        ),
        "STACKCHAIN_PRIVATE_STATE_ACTIVE_KEY_ID": "next",
    }
    if tamper:
        with sqlite3.connect(state / "saved-searches.sqlite3") as connection:
            connection.execute(
                "UPDATE saved_searches SET views = 'v1:tampered' WHERE login = 'timmy'"
            )
    return subprocess.run(
        [sys.executable, str(SCRIPT)], env=env, text=True, capture_output=True
    )


def test_rotation_command_rewraps_shared_store_rows_without_changing_metadata(tmp_path):
    state = tmp_path / "state"
    path = state / "saved-searches.sqlite3"
    store = SavedSearchStore(path, encryption_key=b"o" * 32)
    expected = store.replace(
        "timmy",
        0,
        [{"id": "mine", "name": "My private work", "query": "assignee:timmy"}],
    )

    completed = run_rotation(state)

    assert completed.returncode == 0, completed.stderr
    report = json.loads(completed.stdout)
    assert report["saved-searches"] == {
        "current": 0,
        "failed": 0,
        "migrated": 1,
        "total": 1,
    }
    with sqlite3.connect(path) as connection:
        revision, payload = connection.execute(
            "SELECT revision, views FROM saved_searches WHERE login = 'timmy'"
        ).fetchone()
    assert revision == expected["revision"]
    assert payload.startswith("v2:next:")
    assert SavedSearchStore(
        path, encryption_key=({"next": b"n" * 32}, "next")
    ).get("timmy") == expected


def test_rotation_command_fails_closed_and_never_prints_private_content(tmp_path):
    state = tmp_path / "state"
    secret = "private launch roadmap"
    store = SavedSearchStore(state / "saved-searches.sqlite3", encryption_key=b"o" * 32)
    store.replace("timmy", 0, [{"id": "secret", "name": secret, "query": "is:open"}])

    completed = run_rotation(state, tamper=True)

    assert completed.returncode == 1
    report = json.loads(completed.stdout)
    assert report["saved-searches"]["failed"] == 1
    assert secret not in completed.stdout
    assert "timmy" not in completed.stdout
    assert completed.stderr == ""


def test_rotation_command_rewraps_completed_filed_history_without_printing_it(tmp_path):
    state = tmp_path / "state"
    path = state / "completed-filed-reviews.sqlite3"
    private_receipt = {
        "repository": "private-rotation/canary",
        "number": 1252,
        "updated_at": "2026-08-22T12:00:00Z",
    }
    CompletedFiledReviewStore(path, encryption_key=b"o" * 32).merge("timmy", [private_receipt])

    completed = run_rotation(state)

    assert completed.returncode == 0, completed.stderr
    report = json.loads(completed.stdout)
    assert report["completed-filed-reviews"] == {
        "current": 0, "failed": 0, "migrated": 1, "total": 1
    }
    assert "private-rotation" not in completed.stdout
    assert "timmy" not in completed.stdout
    with sqlite3.connect(path) as connection:
        payload = connection.execute(
            "SELECT receipts FROM completed_filed_review_collections WHERE login = 'timmy'"
        ).fetchone()[0]
    assert payload.startswith("v2:next:")
    assert CompletedFiledReviewStore(
        path, encryption_key=({"next": b"n" * 32}, "next")
    ).get("timmy") == {"receipts": [private_receipt]}


def test_rotation_command_rewraps_mobile_queue_priority(tmp_path):
    state = tmp_path / "state"
    path = state / "queue-priority.sqlite3"
    preferred = list(DEFAULT_QUEUE_ORDER)
    preferred.remove("following")
    preferred.insert(1, "following")
    expected = QueuePriorityStore(path, encryption_key=b"o" * 32).replace(
        "timmy", 0, preferred
    )

    completed = run_rotation(state)

    assert completed.returncode == 0, completed.stderr
    assert json.loads(completed.stdout)["queue-priority"] == {
        "current": 0, "failed": 0, "migrated": 1, "total": 1
    }
    with sqlite3.connect(path) as connection:
        payload = connection.execute(
            "SELECT queue_order FROM queue_priorities WHERE login = 'timmy'"
        ).fetchone()[0]
    assert payload.startswith("v2:next:")
    assert QueuePriorityStore(
        path, encryption_key=({"next": b"n" * 32}, "next")
    ).get("timmy") == expected


def test_rotation_command_rewraps_recent_work_without_printing_titles(tmp_path):
    state = tmp_path / "state"
    path = state / "recent-work.sqlite3"
    private_item = {
        "kind": "issue",
        "repository": "private/canary",
        "number": 1475,
        "title": "Secret release investigation",
        "route": "#/my-work/issue/private/canary/1475",
    }
    expected = RecentWorkStore(path, encryption_key=b"o" * 32).record("timmy", private_item)

    completed = run_rotation(state)

    assert completed.returncode == 0, completed.stderr
    assert json.loads(completed.stdout)["recent-work"] == {
        "current": 0, "failed": 0, "migrated": 1, "total": 1
    }
    assert private_item["title"] not in completed.stdout
    assert "timmy" not in completed.stdout
    with sqlite3.connect(path) as connection:
        payload = connection.execute(
            "SELECT items FROM recent_work WHERE login = 'timmy'"
        ).fetchone()[0]
    assert payload.startswith("v2:next:")
    assert RecentWorkStore(
        path, encryption_key=({"next": b"n" * 32}, "next")
    ).get("timmy") == expected