605 lines
22 KiB
Python
605 lines
22 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
import httpx
|
|
|
|
from src import main
|
|
from src.state_encryption import PrivateStateEncryptionError
|
|
from src.today_store import TodayPlanFull, TodaySessionConflict, TodayStore
|
|
|
|
|
|
def test_task_bearing_today_journey_is_encrypted_at_rest_and_survives_restart(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
key = b"t" * 32
|
|
store = TodayStore(path, encryption_key=key, clock=lambda: 1234.5)
|
|
issue = "issue:private-roadmap/launch:4242:"
|
|
device = "private-phone-canary"
|
|
recap_session = "private-recap-canary"
|
|
|
|
plan = store.apply_batch("Timmy", [
|
|
{"operation_id": "add", "action": "add", "item_id": issue},
|
|
{"operation_id": "configure", "action": "configure", "item_id": "plan",
|
|
"capacity_minutes": 180, "estimates": {issue: 75}},
|
|
])
|
|
session = store.update_session(
|
|
"timmy", base_revision=0, device_id=device, identity=issue,
|
|
elapsed_ms=90_000, running=True,
|
|
)
|
|
recap = store.save_recap("timmy", recap_session, [
|
|
{"identity": issue, "estimate_minutes": 75, "actual_minutes": 91}
|
|
])
|
|
assert store.begin_time_log("timmy", recap_session, issue, 91) == "claimed"
|
|
store.finish_time_log("timmy", recap_session, issue, succeeded=True)
|
|
|
|
retained = b"".join(
|
|
candidate.read_bytes()
|
|
for candidate in (path, path.with_name(path.name + "-wal"))
|
|
if candidate.exists()
|
|
)
|
|
for canary in (issue, device, recap_session):
|
|
assert canary.encode() not in retained
|
|
assert retained.count(b"v1:") >= 4
|
|
|
|
reopened = TodayStore(path, encryption_key=key)
|
|
assert reopened.get("timmy") == {
|
|
key: value for key, value in plan.items()
|
|
if key not in {"accepted_operation_ids", "duplicate_operation_ids", "rejected_operations"}
|
|
}
|
|
assert reopened.get_session("timmy") == session
|
|
assert reopened.list_recaps("timmy") == [recap]
|
|
assert reopened.begin_time_log("timmy", recap_session, issue, 91) == "logged"
|
|
|
|
|
|
def test_legacy_today_rows_migrate_on_use_without_changing_behavior(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
key = b"m" * 32
|
|
store = TodayStore(path, encryption_key=key)
|
|
issue = "issue:legacy-private/project:7:"
|
|
with sqlite3.connect(path) as connection:
|
|
connection.execute(
|
|
"INSERT INTO today_plans VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
("timmy", 4, f'["{issue}"]', 120, f'{{"{issue}":45}}', "2026-08-19", "UTC"),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO today_sessions VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
("timmy", 3, "legacy-phone", issue, 5000, 1, None, 99.0),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO today_recaps VALUES (?, ?, ?, ?)",
|
|
("timmy", "legacy-session", 88.0,
|
|
f'[{ {"identity": issue, "estimate_minutes": 45, "actual_minutes": 50} }]'.replace("'", '"')),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO today_time_logs VALUES (?, ?, ?, ?, ?)",
|
|
("timmy", "legacy-session", issue, 50, "pending"),
|
|
)
|
|
|
|
assert store.get("timmy")["revision"] == 4
|
|
assert store.get_session("timmy")["revision"] == 3
|
|
assert store.list_recaps("timmy")[0]["items"][0]["identity"] == issue
|
|
assert store.begin_time_log("timmy", "legacy-session", issue, 50) == "pending"
|
|
|
|
retained = path.read_bytes()
|
|
for canary in (issue, "legacy-phone", "legacy-session"):
|
|
assert canary.encode() not in retained
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_tampered_or_account_substituted_today_state_fails_closed_with_retry(monkeypatch, tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(path, encryption_key=b"a" * 32)
|
|
store.apply("timmy", "seed", "add", "issue:private/repo:9:")
|
|
with sqlite3.connect(path) as connection:
|
|
payload = connection.execute(
|
|
"SELECT ids FROM today_plans WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
connection.execute(
|
|
"INSERT INTO today_plans(login, revision, ids, estimates) VALUES ('alexander', 1, ?, '{}')",
|
|
(payload,),
|
|
)
|
|
|
|
with pytest.raises(PrivateStateEncryptionError):
|
|
store.get("alexander")
|
|
|
|
async def user():
|
|
return {"login": "alexander"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: store)
|
|
with pytest.raises(main.HTTPException) as raised:
|
|
await main.get_today_plan()
|
|
assert raised.value.status_code == 503
|
|
assert raised.value.detail == "Today synchronization is unavailable"
|
|
assert raised.value.headers == {"Retry-After": "1"}
|
|
|
|
|
|
def test_active_session_is_durable_and_account_scoped(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(path, clock=lambda: 1234.5)
|
|
|
|
saved = store.update_session(
|
|
"Timmy", base_revision=0, device_id="phone-a",
|
|
identity="issue:stackchain/dashboard:42:", elapsed_ms=90_000, running=True,
|
|
)
|
|
|
|
assert saved == {
|
|
"revision": 1,
|
|
"device_id": "phone-a",
|
|
"identity": "issue:stackchain/dashboard:42:",
|
|
"elapsed_ms": 90_000,
|
|
"running": True,
|
|
"break_deadline_at": None,
|
|
"updated_at": 1234.5,
|
|
}
|
|
assert TodayStore(path).get_session("timmy") == saved
|
|
assert store.get_session("alexander") == {
|
|
"revision": 0,
|
|
"device_id": "",
|
|
"identity": "",
|
|
"elapsed_ms": 0,
|
|
"running": False,
|
|
"break_deadline_at": None,
|
|
"updated_at": None,
|
|
}
|
|
|
|
|
|
def test_timed_break_deadline_is_durable_in_the_account_session(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(path, clock=lambda: 1234.5)
|
|
|
|
saved = store.update_session(
|
|
"timmy", base_revision=0, device_id="phone-a",
|
|
identity="issue:r:42:", elapsed_ms=90_000, running=False,
|
|
break_deadline_at=1_800_000,
|
|
)
|
|
|
|
assert saved["break_deadline_at"] == 1_800_000
|
|
assert saved["running"] is False
|
|
assert TodayStore(path).get_session("timmy") == saved
|
|
|
|
|
|
def test_active_session_claim_rejects_a_stale_revision(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3")
|
|
first = store.update_session(
|
|
"timmy", base_revision=0, device_id="phone-a", identity="issue:r:1:",
|
|
elapsed_ms=1_000, running=True,
|
|
)
|
|
claimed = store.update_session(
|
|
"timmy", base_revision=first["revision"], device_id="phone-b",
|
|
identity=first["identity"], elapsed_ms=first["elapsed_ms"], running=True,
|
|
)
|
|
|
|
with pytest.raises(TodaySessionConflict) as raised:
|
|
store.update_session(
|
|
"timmy", base_revision=first["revision"], device_id="phone-a",
|
|
identity=first["identity"], elapsed_ms=5_000, running=True,
|
|
)
|
|
|
|
assert raised.value.session == claimed
|
|
assert store.get_session("timmy") == claimed
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_active_session_api_persists_a_bounded_break_deadline(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
store = TodayStore(tmp_path / "today.sqlite3")
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: store)
|
|
payload = main.TodaySessionUpdate(
|
|
base_revision=0, device_id="phone-a", identity="issue:r:1:",
|
|
elapsed_ms=5_000, running=False, break_deadline_at=1_800_000,
|
|
)
|
|
|
|
saved = await main.update_today_session(payload)
|
|
|
|
assert saved["break_deadline_at"] == 1_800_000
|
|
assert (await main.get_today_session()) == saved
|
|
|
|
|
|
def test_active_session_api_rejects_a_running_break():
|
|
with pytest.raises(ValueError, match="break must remain paused"):
|
|
main.TodaySessionUpdate(
|
|
base_revision=0, device_id="phone-a", identity="issue:r:1:",
|
|
elapsed_ms=5_000, running=True, break_deadline_at=1_800_000,
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_active_session_api_returns_current_state_on_claim_conflict(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
store = TodayStore(tmp_path / "today.sqlite3")
|
|
current = store.update_session(
|
|
"timmy", base_revision=0, device_id="phone-a", identity="issue:r:1:",
|
|
elapsed_ms=5_000, running=True,
|
|
)
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: store)
|
|
|
|
payload = main.TodaySessionUpdate(
|
|
base_revision=0, device_id="phone-b", identity="issue:r:1:",
|
|
elapsed_ms=5_000, running=True,
|
|
)
|
|
with pytest.raises(main.HTTPException) as raised:
|
|
await main.update_today_session(payload)
|
|
|
|
assert raised.value.status_code == 409
|
|
assert raised.value.detail == {"code": "session_changed", "session": current}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_transient_today_store_failure_tells_clients_when_to_retry(monkeypatch):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
class BusyStore:
|
|
def apply(self, *_args, **_kwargs):
|
|
raise sqlite3.OperationalError("database is locked")
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: BusyStore())
|
|
payload = main.TodayOperation(
|
|
operation_id="retry-me", action="add", item_id="issue:r:1:"
|
|
)
|
|
|
|
with pytest.raises(main.HTTPException) as raised:
|
|
await main.update_today_plan(payload)
|
|
|
|
assert raised.value.status_code == 503
|
|
assert raised.value.headers == {"Retry-After": "1"}
|
|
|
|
|
|
def test_operations_are_durable_ordered_idempotent_and_account_scoped(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(path, limit=3)
|
|
|
|
assert store.apply("timmy", "op-1", "add", "issue:stackchain/dashboard:1:") == {
|
|
"revision": 1,
|
|
"ids": ["issue:stackchain/dashboard:1:"],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
}
|
|
store.apply("timmy", "op-2", "add", "issue:stackchain/dashboard:2:")
|
|
store.apply("timmy", "op-3", "add", "issue:stackchain/dashboard:3:")
|
|
moved = store.apply(
|
|
"timmy", "op-4", "move", "issue:stackchain/dashboard:3:", direction="up"
|
|
)
|
|
duplicate = store.apply(
|
|
"timmy", "op-4", "move", "issue:stackchain/dashboard:3:", direction="up"
|
|
)
|
|
|
|
assert moved == duplicate == {
|
|
"revision": 4,
|
|
"ids": [
|
|
"issue:stackchain/dashboard:1:",
|
|
"issue:stackchain/dashboard:3:",
|
|
"issue:stackchain/dashboard:2:",
|
|
],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
}
|
|
assert TodayStore(path, limit=3).get("timmy") == moved
|
|
assert store.get("alexander") == {
|
|
"revision": 0,
|
|
"ids": [],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
}
|
|
|
|
|
|
def test_initialized_today_reads_remain_available_during_a_planning_write(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(path, timeout=0.05)
|
|
expected = store.apply("timmy", "seed", "add", "issue:r:1:")
|
|
|
|
writer = sqlite3.connect(path)
|
|
writer.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
assert TodayStore(path, timeout=0.05).get("timmy") == expected
|
|
finally:
|
|
writer.rollback()
|
|
writer.close()
|
|
|
|
|
|
def test_capacity_and_estimates_are_durable_account_scoped_and_follow_item_identity(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(path, limit=3)
|
|
store.apply("timmy", "add-1", "add", "issue:r:1:")
|
|
store.apply("timmy", "add-2", "add", "issue:r:2:")
|
|
|
|
configured = store.apply_batch("timmy", [{
|
|
"operation_id": "capacity-1",
|
|
"action": "configure",
|
|
"item_id": "plan",
|
|
"capacity_minutes": 180,
|
|
"estimates": {"issue:r:1:": 60, "issue:r:2:": 45, "issue:r:99:": 30},
|
|
"base_revision": 2,
|
|
}])
|
|
store.apply("timmy", "move-2", "move", "issue:r:2:", direction="up")
|
|
store.apply("timmy", "remove-1", "remove", "issue:r:1:")
|
|
|
|
assert configured["capacity_minutes"] == 180
|
|
assert configured["estimates"] == {"issue:r:1:": 60, "issue:r:2:": 45}
|
|
assert TodayStore(path, limit=3).get("timmy") == {
|
|
"revision": 5,
|
|
"ids": ["issue:r:2:"],
|
|
"capacity_minutes": 180,
|
|
"estimates": {"issue:r:2:": 45},
|
|
}
|
|
assert store.get("alexander") == {
|
|
"revision": 0,
|
|
"ids": [],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
}
|
|
|
|
|
|
def test_limit_is_atomic_and_remove_frees_capacity(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", limit=2)
|
|
store.apply("timmy", "one", "add", "issue:r:1:")
|
|
store.apply("timmy", "two", "add", "issue:r:2:")
|
|
|
|
with pytest.raises(TodayPlanFull):
|
|
store.apply("timmy", "three", "add", "issue:r:3:")
|
|
|
|
assert store.get("timmy") == {
|
|
"revision": 2,
|
|
"ids": ["issue:r:1:", "issue:r:2:"],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
}
|
|
store.apply("timmy", "remove", "remove", "issue:r:1:")
|
|
assert store.apply("timmy", "retry-three", "add", "issue:r:3:")["ids"] == [
|
|
"issue:r:2:",
|
|
"issue:r:3:",
|
|
]
|
|
|
|
|
|
def test_batch_applies_ordered_operations_once_and_rejects_only_capacity_conflicts(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", limit=2)
|
|
|
|
result = store.apply_batch(
|
|
"timmy",
|
|
[
|
|
{"operation_id": "one", "action": "add", "item_id": "issue:r:1:"},
|
|
{"operation_id": "two", "action": "add", "item_id": "issue:r:2:"},
|
|
{"operation_id": "full", "action": "add", "item_id": "issue:r:3:"},
|
|
{"operation_id": "remove", "action": "remove", "item_id": "issue:r:1:"},
|
|
{"operation_id": "three", "action": "add", "item_id": "issue:r:3:"},
|
|
],
|
|
)
|
|
|
|
assert result == {
|
|
"revision": 4,
|
|
"ids": ["issue:r:2:", "issue:r:3:"],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
"accepted_operation_ids": ["one", "two", "remove", "three"],
|
|
"duplicate_operation_ids": [],
|
|
"rejected_operations": [{"operation_id": "full", "reason": "today_full"}],
|
|
}
|
|
replay = store.apply_batch("timmy", [
|
|
{"operation_id": "two", "action": "add", "item_id": "issue:r:2:"},
|
|
{"operation_id": "three", "action": "add", "item_id": "issue:r:3:"},
|
|
])
|
|
assert replay["revision"] == 4
|
|
assert replay["accepted_operation_ids"] == []
|
|
assert replay["duplicate_operation_ids"] == ["two", "three"]
|
|
|
|
|
|
def test_operation_receipts_are_bounded_per_account_and_existing_schema_migrates(tmp_path):
|
|
path = tmp_path / "today.sqlite3"
|
|
with sqlite3.connect(path) as connection:
|
|
connection.execute(
|
|
"CREATE TABLE today_operations (login TEXT NOT NULL, operation_id TEXT NOT NULL, "
|
|
"PRIMARY KEY (login, operation_id))"
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO today_operations(login, operation_id) VALUES ('timmy', 'legacy')"
|
|
)
|
|
|
|
store = TodayStore(path, limit=20, operation_limit=3, clock=lambda: 1_000)
|
|
for index in range(5):
|
|
store.apply("timmy", f"op-{index}", "add", f"issue:r:{index}:")
|
|
store.apply("alexander", "other", "add", "issue:r:99:")
|
|
|
|
with sqlite3.connect(path) as connection:
|
|
columns = {row[1] for row in connection.execute("PRAGMA table_info(today_operations)")}
|
|
timmy = connection.execute(
|
|
"SELECT operation_id FROM today_operations WHERE login = 'timmy' ORDER BY rowid"
|
|
).fetchall()
|
|
alexander = connection.execute(
|
|
"SELECT operation_id FROM today_operations WHERE login = 'alexander'"
|
|
).fetchall()
|
|
|
|
assert "created_at" in columns
|
|
assert timmy == [("op-2",), ("op-3",), ("op-4",)]
|
|
assert alexander == [("other",)]
|
|
|
|
|
|
def test_pruned_stale_today_operation_cannot_reorder_a_newer_plan(tmp_path):
|
|
store = TodayStore(tmp_path / "today.sqlite3", limit=5, operation_limit=1)
|
|
store.apply_batch("timmy", [
|
|
{"operation_id": "seed-1", "action": "add", "item_id": "issue:r:1:", "base_revision": 0},
|
|
{"operation_id": "seed-2", "action": "add", "item_id": "issue:r:2:", "base_revision": 0},
|
|
])
|
|
store.apply_batch("timmy", [{
|
|
"operation_id": "offline-move", "action": "move", "item_id": "issue:r:2:",
|
|
"direction": "up", "base_revision": 2,
|
|
}])
|
|
newer = store.apply_batch("timmy", [{
|
|
"operation_id": "newer-move", "action": "move", "item_id": "issue:r:1:",
|
|
"direction": "up", "base_revision": 3,
|
|
}])
|
|
|
|
replay = store.apply_batch("timmy", [{
|
|
"operation_id": "offline-move", "action": "move", "item_id": "issue:r:2:",
|
|
"direction": "up", "base_revision": 2,
|
|
}])
|
|
|
|
assert replay["ids"] == newer["ids"] == ["issue:r:1:", "issue:r:2:"]
|
|
assert replay["revision"] == 4
|
|
assert replay["accepted_operation_ids"] == []
|
|
assert replay["rejected_operations"] == [
|
|
{"operation_id": "offline-move", "reason": "stale_intent"}
|
|
]
|
|
|
|
|
|
def test_today_receipt_age_pruning_is_account_scoped(tmp_path):
|
|
clock = [1_000.0]
|
|
path = tmp_path / "today.sqlite3"
|
|
store = TodayStore(
|
|
path, limit=5, operation_retention_seconds=60, clock=lambda: clock[0]
|
|
)
|
|
store.apply("timmy", "old", "add", "issue:r:1:")
|
|
store.apply("alexander", "other-old", "add", "issue:r:2:")
|
|
clock[0] += 61
|
|
store.apply("timmy", "fresh", "remove", "issue:r:1:")
|
|
|
|
with sqlite3.connect(path) as connection:
|
|
timmy = connection.execute(
|
|
"SELECT operation_id FROM today_operations WHERE login = 'timmy'"
|
|
).fetchall()
|
|
alexander = connection.execute(
|
|
"SELECT operation_id FROM today_operations WHERE login = 'alexander'"
|
|
).fetchall()
|
|
|
|
assert timmy == [("fresh",)]
|
|
assert alexander == [("other-old",)]
|
|
|
|
|
|
def test_today_api_model_preserves_client_base_revision():
|
|
operation = main.TodayOperation(
|
|
operation_id="offline", action="move", item_id="issue:r:2:",
|
|
direction="up", base_revision=7,
|
|
)
|
|
|
|
assert operation.model_dump()["base_revision"] == 7
|
|
|
|
|
|
def test_today_api_model_accepts_bounded_capacity_configuration():
|
|
operation = main.TodayOperation(
|
|
operation_id="capacity", action="configure", item_id="plan",
|
|
capacity_minutes=180,
|
|
estimates={"issue:r:1:": 60, "issue:r:2:": 45},
|
|
base_revision=2,
|
|
)
|
|
|
|
assert operation.model_dump() == {
|
|
"operation_id": "capacity",
|
|
"action": "configure",
|
|
"item_id": "plan",
|
|
"direction": None,
|
|
"base_revision": 2,
|
|
"capacity_minutes": 180,
|
|
"estimates": {"issue:r:1:": 60, "issue:r:2:": 45},
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_single_capacity_configuration_uses_atomic_batch_path(monkeypatch):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
calls = []
|
|
class Store:
|
|
def apply_batch(self, login, operations):
|
|
calls.append((login, operations))
|
|
return {"revision": 1, "ids": [], "capacity_minutes": 180, "estimates": {}}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "_today_store", lambda: Store())
|
|
payload = main.TodayOperation(
|
|
operation_id="capacity", action="configure", item_id="plan", capacity_minutes=180,
|
|
)
|
|
|
|
result = await main.update_today_plan(payload)
|
|
|
|
assert result["capacity_minutes"] == 180
|
|
assert calls == [("timmy", [payload.model_dump()])]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authenticated_today_api_uses_confirmed_account_and_csrf(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
|
|
monkeypatch.setenv(
|
|
"STACKCHAIN_DASHBOARD_SESSION_SECRET",
|
|
"a-separate-session-signing-secret-with-enough-entropy",
|
|
)
|
|
monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_TODAY_DB", str(tmp_path / "today.sqlite3"))
|
|
|
|
async def user():
|
|
return {"id": 1, "login": "Timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
forbidden = await client.patch(
|
|
"/api/v1/today",
|
|
json={
|
|
"operation_id": "mobile-1",
|
|
"action": "add",
|
|
"item_id": "issue:stackchain/dashboard:357:",
|
|
},
|
|
)
|
|
changed = await client.patch(
|
|
"/api/v1/today",
|
|
json={
|
|
"operations": [
|
|
{"operation_id": "mobile-1", "action": "add", "item_id": "issue:stackchain/dashboard:357:"},
|
|
{"operation_id": "mobile-2", "action": "add", "item_id": "issue:stackchain/dashboard:359:"},
|
|
],
|
|
},
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
fetched = await client.get("/api/v1/today")
|
|
changed_session = await client.patch(
|
|
"/api/v1/today/session",
|
|
json={
|
|
"base_revision": 0,
|
|
"device_id": "phone-a",
|
|
"identity": "issue:stackchain/dashboard:357:",
|
|
"elapsed_ms": 12_000,
|
|
"running": True,
|
|
},
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
fetched_session = await client.get("/api/v1/today/session")
|
|
|
|
assert forbidden.status_code == 403
|
|
assert changed.status_code == 200
|
|
assert changed.json() == {
|
|
"revision": 2,
|
|
"ids": ["issue:stackchain/dashboard:357:", "issue:stackchain/dashboard:359:"],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
"accepted_operation_ids": ["mobile-1", "mobile-2"],
|
|
"duplicate_operation_ids": [],
|
|
"rejected_operations": [],
|
|
}
|
|
assert fetched.json() == {
|
|
"revision": 2,
|
|
"ids": ["issue:stackchain/dashboard:357:", "issue:stackchain/dashboard:359:"],
|
|
"capacity_minutes": None,
|
|
"estimates": {},
|
|
}
|
|
assert fetched.headers["cache-control"] == "no-store"
|
|
assert changed_session.status_code == 200
|
|
assert fetched_session.json() == changed_session.json()
|
|
assert fetched_session.headers["cache-control"] == "no-store"
|