from __future__ import annotations
import inspect
import os
import pytest
from ._async_utils import is_async_callable
dcap_qvl = pytest.importorskip("dcap_qvl")
RUN_NETWORK = os.getenv("DCAP_QVL_RUN_NETWORK_TESTS") == "1"
def test_async_functions_are_exported() -> None:
expected = [
"get_collateral",
"get_collateral_from_pcs",
"get_collateral_and_verify",
]
for name in expected:
assert hasattr(dcap_qvl, name), f"{name} is not exported"
assert callable(dcap_qvl.get_collateral)
assert is_async_callable(dcap_qvl.get_collateral_from_pcs, b"short")
assert is_async_callable(dcap_qvl.get_collateral_and_verify, b"short")
@pytest.mark.asyncio
async def test_get_collateral_returns_awaitable() -> None:
ret = dcap_qvl.get_collateral(
pccs_url="://invalid-url",
raw_quote=b"short",
)
assert inspect.isawaitable(ret)
with pytest.raises(ValueError):
await ret
@pytest.mark.asyncio
async def test_get_collateral_rejects_non_bytes_quote() -> None:
with pytest.raises(TypeError, match="raw_quote"):
await dcap_qvl.get_collateral("http://example.com", "not bytes")
@pytest.mark.asyncio
async def test_get_collateral_from_pcs_rejects_invalid_quote() -> None:
with pytest.raises(ValueError, match="Failed to parse quote"):
await dcap_qvl.get_collateral_from_pcs(b"short")
@pytest.mark.asyncio
async def test_get_collateral_and_verify_rejects_invalid_quote() -> None:
with pytest.raises(ValueError, match="Failed to parse quote"):
await dcap_qvl.get_collateral_and_verify(b"short")
@pytest.mark.asyncio
@pytest.mark.skipif(
not RUN_NETWORK,
reason="Network test disabled (set DCAP_QVL_RUN_NETWORK_TESTS=1 to enable)",
)
async def test_get_collateral_from_pcs_network_smoke(sample_sgx_quote_bytes: bytes) -> None:
collateral = await dcap_qvl.get_collateral_from_pcs(sample_sgx_quote_bytes)
assert collateral is not None
@pytest.fixture
def sample_sgx_quote_bytes() -> bytes:
with open("sample/sgx_quote", "rb") as f:
return f.read()