praxis/tests/test_vc_integration.py

"""VC integration test — issue → verify + key rotation (SLICE-09 TASK-09-06).

Issue a credential, verify it (valid: true, credentialTier: formative).
Revoke → verify (valid: false, status: revoked). Tamper payload → verify
fails. Key rotation: issue with key A, rotate to key B, issue with key B,
verify both (A against archived public key, B against active).
"""

from __future__ import annotations

import asyncio
import base64
import json
from pathlib import Path

import pytest

from db.migrate import apply_migrations
from db.store import PraxisStore
from server.vc import issuer, issuer_keys
from server.vc.verification import verify_credential, revoke_credential


def _await(coro):
    return asyncio.run(coro)


@pytest.fixture
def store(tmp_path: Path) -> PraxisStore:
    db = tmp_path / "test_vc_int.db"
    apply_migrations(db)
    return PraxisStore(db)


def test_issue_and_verify_valid(store: PraxisStore):
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    cred_id = _await(
        issuer.issue_credential(
            store=store,
            signing_key=kp.signing_key,
            key_id=kp.key_id,
            learner_id="learner-1",
            path="customer-service",
            scenarios_passed=["cs_refund_ca_v01", "cs_escalation_ca_v02", "cs_billing_v01"],
            rubric_score=4.2,
            completed_weeks=6,
            evidence=[{"type": "Evidence", "rubricMean": 4.2, "distinctScenarios": 3}],
        )
    )
    result = _await(verify_credential(store, cred_id))
    assert result is not None
    assert result["valid"] is True
    assert result["status"] == "active"
    assert result["credentialTier"] == "formative"
    assert result["mastery"]["completedWeeks"] == 6
    assert result["mastery"]["path"] == "customer-service"


def test_revoke_then_verify_invalid(store: PraxisStore):
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    cred_id = _await(
        issuer.issue_credential(
            store=store,
            signing_key=kp.signing_key,
            key_id=kp.key_id,
            learner_id="learner-1",
            path="customer-service",
            scenarios_passed=["s1", "s2", "s3"],
            rubric_score=4.0,
            completed_weeks=6,
            evidence=[],
        )
    )
    ok = _await(revoke_credential(store, cred_id))
    assert ok is True
    result = _await(verify_credential(store, cred_id))
    assert result is not None
    assert result["valid"] is False
    assert result["status"] == "revoked"


def test_tamper_payload_verify_fails(store: PraxisStore):
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    cred_id = _await(
        issuer.issue_credential(
            store=store,
            signing_key=kp.signing_key,
            key_id=kp.key_id,
            learner_id="learner-1",
            path="customer-service",
            scenarios_passed=["s1", "s2", "s3"],
            rubric_score=3.9,
            completed_weeks=6,
            evidence=[],
        )
    )
    row = _await(store.get_credential(cred_id))
    secured = json.loads(row["vc_payload_json"])
    secured["credentialSubject"]["scenariosPassed"] = ["forged"]
    vk = _await(issuer_keys.get_public_key_for_verification(store, kp.key_id))
    assert issuer.verify_proof(secured, vk) is False


def test_key_rotation_old_vc_still_verifies(store: PraxisStore):
    root = b"k" * 32
    kp_a = _await(issuer_keys.init_issuer_key(store, root))
    cred_a = _await(
        issuer.issue_credential(
            store=store,
            signing_key=kp_a.signing_key,
            key_id=kp_a.key_id,
            learner_id="learner-1",
            path="customer-service",
            scenarios_passed=["s1", "s2", "s3"],
            rubric_score=4.1,
            completed_weeks=6,
            evidence=[],
        )
    )
    kp_b = _await(issuer_keys.rotate_key(store, root))
    cred_b = _await(
        issuer.issue_credential(
            store=store,
            signing_key=kp_b.signing_key,
            key_id=kp_b.key_id,
            learner_id="learner-2",
            path="customer-service",
            scenarios_passed=["s1", "s2", "s3"],
            rubric_score=4.3,
            completed_weeks=6,
            evidence=[],
        )
    )
    res_a = _await(verify_credential(store, cred_a))
    res_b = _await(verify_credential(store, cred_b))
    assert res_a["valid"] is True
    assert res_b["valid"] is True
    row_a = _await(store.get_credential(cred_a))
    secured_a = json.loads(row_a["vc_payload_json"])
    vm_a = secured_a["proof"]["verificationMethod"]
    row_b = _await(store.get_credential(cred_b))
    secured_b = json.loads(row_b["vc_payload_json"])
    vm_b = secured_b["proof"]["verificationMethod"]
    assert vm_a != vm_b
    old_row = _await(store.get_public_key_row(kp_a.key_id))
    assert old_row["status"] == "superseded"


def test_verify_returns_none_for_unknown_id(store: PraxisStore):
    result = _await(verify_credential(store, "vc-doesnotexist"))
    assert result is None


def test_valid_until_is_three_years_out(store: PraxisStore):
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    cred_id = _await(
        issuer.issue_credential(
            store=store,
            signing_key=kp.signing_key,
            key_id=kp.key_id,
            learner_id="learner-1",
            path="customer-service",
            scenarios_passed=["s1", "s2", "s3"],
            rubric_score=4.0,
            completed_weeks=6,
            evidence=[],
        )
    )
    row = _await(store.get_credential(cred_id))
    secured = json.loads(row["vc_payload_json"])
    vf = secured["validFrom"]
    vu = secured["validUntil"]
    assert vf[:4] == "2026"
    assert vu[:4] == "2029"
    assert vu > vf