mirror of
https://github.com/Av0idD4rk/firegex.git
synced 2026-10-23 15:30:59 +00:00
863422f202
Expose 28 tools with shared REST operations, role-scoped bearer authentication, revisions, durable retries, snapshots, jobs and timed rollback. Add Docker verification harnesses and setup/workflow documentation in docs/mcp.md and inst.md. Validated with 330 unit tests, two consecutive 25-case MCP/import runs, and restart recovery; the full upstream regression run remains incomplete.
48 lines
1.9 KiB
Python
48 lines
1.9 KiB
Python
import time
|
|
import sqlite3
|
|
import pytest
|
|
from mcp_api.store import Store
|
|
|
|
|
|
def test_tokens_are_hashed_revocable_and_expire(tmp_path):
|
|
store = Store(tmp_path / "mcp.sqlite3")
|
|
store.open()
|
|
credential = store.token_create("laptop", "operator", 1)
|
|
assert store.token_verify(credential["token"])["name"] == "laptop"
|
|
assert not store.token_verify("incorrect")
|
|
assert credential["token"] not in repr(store.token_list())
|
|
assert store.token_revoke(credential["id"])
|
|
assert store.token_verify(credential["token"]) is None
|
|
expired = store.token_create("expired")
|
|
with store.conn:
|
|
store.conn.execute("UPDATE tokens SET expires=? WHERE id=?", (time.time() - 1, expired["id"]))
|
|
assert store.token_verify(expired["token"]) is None
|
|
store.close()
|
|
assert credential["token"].encode() not in (tmp_path / "mcp.sqlite3").read_bytes()
|
|
|
|
|
|
def test_pending_changes_and_request_receipts_survive_reopening(tmp_path):
|
|
path = tmp_path / "control.sqlite3"
|
|
store = Store(path)
|
|
store.open()
|
|
store.put("change", "c", {"id": "c", "status": "awaiting_confirmation"})
|
|
store.request_put("actor", "retry-key", "fingerprint", "c")
|
|
store.close()
|
|
store.open()
|
|
assert store.pending("awaiting_confirmation") == [{"id": "c", "status": "awaiting_confirmation"}]
|
|
assert store.request_get("actor", "retry-key")["change_id"] == "c"
|
|
assert store.request_get("other", "retry-key") is None
|
|
store.close()
|
|
|
|
|
|
def test_retry_key_and_change_are_committed_atomically(tmp_path):
|
|
store = Store(tmp_path / "control.sqlite3")
|
|
store.open()
|
|
request = {"actor": "a", "id": "retry", "fingerprint": "f"}
|
|
store.queue_change({"id": "one"}, request)
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
store.queue_change({"id": "two"}, request)
|
|
assert store.get("change", "two") is None
|
|
assert store.request_get("a", "retry")["change_id"] == "one"
|
|
store.close()
|