import pytest from datetime import datetime from app.models import ComputeReceipt, JobType, WorkerRuntimeType, PrivacyMode, InferenceJob, WorkerState from app.receipts import create_receipt, verify_receipt_hash, verify_device_receipt_signature, verify_server_receipt_signature from app.security import generate_job_id, generate_worker_id, generate_session_id @pytest.fixture def sample_job(): return InferenceJob( job_id=generate_job_id(), session_id=generate_session_id(), job_type=JobType.TEXT_EMBEDDING, privacy_mode=PrivacyMode.RAW_INPUT_REMOTE, payload={"text": "hello world"}, ) @pytest.fixture def sample_worker(): return WorkerState( worker_id=generate_worker_id(), session_id=generate_session_id(), runtime_type=WorkerRuntimeType.SAFARI_WASM, ) @pytest.fixture def sample_result(): from app.models import JobResult return JobResult( job_id="job_test", worker_id="worker_test", output={"embedding": [0.1, 0.2]}, latency_ms=150, input_hash="abc123", output_hash="def456", ) def test_receipt_hash_is_stable(sample_job, sample_worker, sample_result): receipt = create_receipt(sample_job, sample_worker, sample_result) assert receipt.receipt_hash is not None assert len(receipt.receipt_hash) == 64 def test_receipt_changes_when_output_changes(sample_job, sample_worker, sample_result): r1 = create_receipt(sample_job, sample_worker, sample_result) sample_result.output = {"embedding": [0.3, 0.4]} r2 = create_receipt(sample_job, sample_worker, sample_result) assert r1.receipt_hash != r2.receipt_hash def test_receipt_rejects_bad_signature(sample_job, sample_worker, sample_result): receipt = create_receipt(sample_job, sample_worker, sample_result) receipt.device_signature = "bad" assert verify_device_receipt_signature(receipt) is False def test_server_signature_valid(sample_job, sample_worker, sample_result): receipt = create_receipt(sample_job, sample_worker, sample_result) assert verify_server_receipt_signature(receipt) is True