get_interchain: per_chain + chains options (default first anchor per chain); bump 0.5.0

transaction.get_interchain / block.get_interchain take per_chain= and chains=
kwargs mapping to prime-node's ?perChain=&chains= params. Default (no kwargs)
returns one anchor per chain. Shared interchain_query() helper, exported; pytest.

.gitea/workflows/publish.yml

@@ -0,0 +1,37 @@
+name: Publish to PyPI Registry
+
+on:
+  release:
+    types: [created, published]
+  workflow_dispatch:
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install build tooling
+        run: python -m pip install --upgrade pip build twine
+
+      - name: Run tests
+        run: |
+          python -m pip install -e '.[dev]'
+          python -m pytest -q
+
+      - name: Build package
+        run: python -m build
+
+      - name: Publish to Gitea PyPI registry
+        env:
+          TWINE_USERNAME: sa-dc-build
+          TWINE_PASSWORD: ${{ secrets.SA_DC_BUILD_API_TOKEN }}
+          TWINE_REPOSITORY_URL: https://git.dragonchain.com/api/packages/dragonchain/pypi
+        run: twine upload --non-interactive dist/*

README.md

@@ -146,6 +146,63 @@ client = DragonchainSDK(
 ### Block
 - `block.get(block_id)` — Get block by ID
 
+### Proof Measure (public, unauthenticated)
+- `proof_measure.get_security(network, since=None)` — A network's accumulated security since `since` (unix seconds), as a raw measure + USD valuation. `network` = `"BTC"` or `"ETH"`.
+- `proof_measure.report(req)` — Per-transaction "securedBy" report over interchain anchors.
+- `proof_measure.health()` — Service liveness.
+
+## Interchain trace
+
+`transaction.get_interchain` / `block.get_interchain` trace a prime block to the
+public-chain anchors covering it. By default they return the **first anchor per
+chain** (anchor proofs are chained, so the earliest per chain is the meaningful
+one):
+
+```python
+# Default: first anchor per chain (ETH, BTC, …)
+trace = client.block.get_interchain("42")
+
+# Up to 3 anchors per chain
+trace = client.block.get_interchain("42", per_chain=3)
+
+# All anchors, only the ETH-mainnet chain ("1"); "0" = BTC
+trace = client.block.get_interchain("42", per_chain=0, chains=["1"])
+```
+
+## Proof Measure
+
+`proof-measure` is a separate, **public, unauthenticated** Dragonchain service
+(the measured-immutability / "securedBy" metric) at
+`https://proof-measure.dragonchain.com`. It needs no credentials. Decimal fields
+are returned as strings (full precision) and timestamps as int unix seconds.
+
+```python
+import time
+from prime_sdk import ProofMeasureClient, ReportRequest, ReportAnchorInput
+
+# Via the main SDK (targets the default public endpoint):
+sec = client.proof_measure.get_security("BTC", int(time.time()) - 3600)
+print(f"BTC secured by {sec.value_usd_formatted} ({sec.raw.value} {sec.raw.unit})")
+
+# Or standalone — no prime credentials needed:
+pm = ProofMeasureClient()  # or ProofMeasureClient("http://localhost:9481")
+
+report = pm.report(
+    ReportRequest(
+        transaction_id="tx-123",
+        prime_id="my-prime",
+        block_id="42",
+        anchors=[
+            ReportAnchorInput(tx_hash="0x...", timestamp=anchor_unix, network="BTC"),
+            ReportAnchorInput(tx_hash="0x...", timestamp=anchor_unix, network="ETH"),
+        ],
+    )
+)
+print(f"Secured by {report.total_value_usd_formatted} across {len(report.anchors)} anchors")
+
+pm.health()
+```
+
 ## Authentication
 
 The SDK uses HMAC-SHA256 authentication. You need to provide:

prime_sdk/__init__.py

@@ -71,11 +71,25 @@ from .models import (
     TransactionTypeCreateResponse,
     VerificationBlock,
 )
+from .models import (
+    AnchorSecurity,
+    HashPower,
+    HealthResponse,
+    RawMeasure,
+    ReportAnchorInput,
+    ReportRequest,
+    SecurityResult,
+    TransactionReport,
+)
+from .interchain import interchain_query
+from .proof_measure import DEFAULT_BASE_URL as PROOF_MEASURE_DEFAULT_BASE_URL
+from .proof_measure import ProofMeasureClient
 from .system import SystemClient
 from .transaction import TransactionClient
 from .transaction_type import TransactionTypeClient
+from .unauthenticated_client import UnauthenticatedClient
 
-__version__ = "0.2.0"
+__version__ = "0.5.0"
 
 
 class DragonchainSDK:
@@ -91,6 +105,7 @@ class DragonchainSDK:
         auth_key_id: str,
         auth_key: str,
         base_url: str,
+        session: "Optional[requests.Session]" = None,
     ):
         """Create a new Dragonchain SDK client.
 
@@ -99,13 +114,21 @@ class DragonchainSDK:
             auth_key_id: Your authentication key ID.
             auth_key: Your authentication key.
             base_url: Base URL of your node (e.g. "https://chains.dragonchain.com").
+            session: Optional pre-configured ``requests.Session``. Inject one
+                with an SSRF-guarded transport adapter when ``base_url`` is
+                attacker-influenced (a tenant's prime_endpoint); the default
+                session is unguarded and intended for trusted/CLI use.
         """
-        self._client = Client(public_id, auth_key_id, auth_key, base_url)
+        self._client = Client(public_id, auth_key_id, auth_key, base_url, session=session)
         self.transaction = TransactionClient(self._client)
         self.transaction_type = TransactionTypeClient(self._client)
         self.contract = ContractClient(self._client)
         self.block = BlockClient(self._client)
         self.system = SystemClient(self._client)
+        # proof-measure is a separate, unauthenticated public service; this
+        # handle targets its default public endpoint. For a custom endpoint
+        # construct a ``ProofMeasureClient`` directly.
+        self.proof_measure = ProofMeasureClient()
 
     def get_client(self) -> Client:
         """Return the underlying HTTP client (advanced use)."""
@@ -123,6 +146,10 @@ __all__ = [
     "SystemClient",
     "TransactionClient",
     "TransactionTypeClient",
+    "ProofMeasureClient",
+    "UnauthenticatedClient",
+    "PROOF_MEASURE_DEFAULT_BASE_URL",
+    "interchain_query",
     # models
     "Block",
     "BlockHeader",
@@ -150,4 +177,13 @@ __all__ = [
     "TransactionTypeCreateRequest",
     "TransactionTypeCreateResponse",
     "VerificationBlock",
+    # proof-measure models
+    "RawMeasure",
+    "SecurityResult",
+    "ReportAnchorInput",
+    "ReportRequest",
+    "AnchorSecurity",
+    "HashPower",
+    "TransactionReport",
+    "HealthResponse",
 ]

prime_sdk/block.py

@@ -1,6 +1,9 @@
 """Block endpoints."""
 
+from typing import List, Optional
+
 from .client import Client
+from .interchain import interchain_query
 from .models import Block, InterchainTrace
 
 
@@ -11,10 +14,21 @@ class BlockClient:
     def get(self, block_id: str) -> Block:
         return self._client.get(f"/api/v1/block/{block_id}", Block)
 
-    def get_interchain(self, block_id: str) -> InterchainTrace:
+    def get_interchain(
+        self,
+        block_id: str,
+        per_chain: Optional[int] = None,
+        chains: Optional[List[str]] = None,
+    ) -> InterchainTrace:
         """Trace a block to the validator (verification) blocks that validated it
         and the public-chain interchain anchors those validator blocks were
-        bundled into."""
+        bundled into.
+
+        By default returns the first anchor per public chain (anchor proofs are
+        chained, so the earliest per chain is the meaningful one). ``per_chain``
+        caps anchors per chain (1 = first per chain; 0 = all); ``chains``
+        restricts to specific chain ids."""
         return self._client.get(
-            f"/api/v1/block/{block_id}/interchain", InterchainTrace
+            f"/api/v1/block/{block_id}/interchain{interchain_query(per_chain, chains)}",
+            InterchainTrace,
         )

prime_sdk/client.py

@@ -39,13 +39,20 @@ class Client:
         auth_key: str,
         base_url: str,
         timeout: int = DEFAULT_TIMEOUT,
+        session: Optional[requests.Session] = None,
     ):
         self.public_id = public_id
         self.auth_key_id = auth_key_id
         self.auth_key = auth_key
         self.base_url = base_url.rstrip("/")
         self.timeout = timeout
-        self._session = requests.Session()
+        # A caller may inject a pre-configured Session — e.g. one with a
+        # transport adapter that refuses to connect to internal IPs — when the
+        # base_url is attacker-influenced (a tenant's prime_endpoint). The
+        # default Session is unguarded, which is fine for trusted/CLI use; the
+        # SSRF policy belongs in the server that points this client at
+        # untrusted endpoints, not baked into the client library.
+        self._session = session if session is not None else requests.Session()
 
     # -- introspection helpers (mirror the Go accessors) --------------------
 
@@ -120,8 +127,17 @@ class Client:
             headers["Content-Type"] = content_type
 
         url = self.base_url + path
+        # allow_redirects=False: Prime is a JSON API that never legitimately
+        # 3xx-redirects, and following redirects is an SSRF vector (a public
+        # endpoint that 302s to an internal address). A redirect surfaces as a
+        # 3xx the caller can inspect rather than being silently followed.
         resp = self._session.request(
-            method, url, data=body_bytes, headers=headers, timeout=self.timeout
+            method,
+            url,
+            data=body_bytes,
+            headers=headers,
+            timeout=self.timeout,
+            allow_redirects=False,
         )
 
         resp_text = resp.text

prime_sdk/interchain.py

@@ -0,0 +1,27 @@
+"""Query builder for the interchain-trace endpoints.
+
+Anchor proofs are chained, so by default the trace returns the first anchor per
+public chain; these options change how many and which chains are returned.
+"""
+
+from typing import List, Optional
+from urllib.parse import quote
+
+
+def interchain_query(
+    per_chain: Optional[int] = None, chains: Optional[List[str]] = None
+) -> str:
+    """Build the "?perChain=...&chains=..." suffix for the interchain trace
+    endpoints.
+
+    Returns "" when nothing is set (the server then applies its defaults: one
+    anchor per chain, all chains). ``per_chain`` caps anchors per public chain,
+    earliest-first (1 = first per chain; 0 = all). ``chains`` restricts to the
+    given chain ids ("1" = ETH mainnet, "0" = BTC; testnet ids differ).
+    """
+    parts: List[str] = []
+    if per_chain is not None:
+        parts.append(f"perChain={int(per_chain)}")
+    if chains:
+        parts.append("chains=" + ",".join(quote(str(c), safe="") for c in chains))
+    return "?" + "&".join(parts) if parts else ""

prime_sdk/models.py

@@ -492,3 +492,169 @@ class ListResponse:
     @classmethod
     def from_dict(cls, d: Dict[str, Any]) -> "ListResponse":
         return cls(items=d.get("items") or [], total_count=d.get("total_count", 0))
+
+
+# --------------------------------------------------------------------------- #
+# Proof Measure (measured immutability / "securedBy")
+# --------------------------------------------------------------------------- #
+#
+# Decimal-valued fields are kept as strings (full precision); timestamps are int
+# unix seconds. JSON keys are camelCase, preserved verbatim from the service.
+
+
+@dataclass
+class RawMeasure:
+    """A network's native cumulative security measure: cumulative hashes for
+    PoW, stake-seconds for PoS."""
+
+    value: str = ""  # human-scaled mantissa, e.g. "484.44"
+    unit: str = ""  # e.g. "Zettahashes", "ETH·s"
+    base: str = ""  # unscaled base amount (hashes or stake-seconds)
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "RawMeasure":
+        return cls(
+            value=d.get("value", ""),
+            unit=d.get("unit", ""),
+            base=d.get("base", ""),
+        )
+
+
+@dataclass
+class SecurityResult:
+    """Security a single network accrued for a window/anchor, as the raw native
+    measure AND a USD valuation, plus a normalized 0..1 score."""
+
+    network: str = ""
+    consensus: str = ""  # "pow" | "pos"
+    raw: RawMeasure = field(default_factory=RawMeasure)
+    value_usd: str = ""
+    value_usd_formatted: str = ""
+    label: str = ""
+    normalized_score: str = ""
+    since: int = 0  # window/anchor start (unix seconds)
+    as_of: int = 0  # latest sample time (unix seconds)
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "SecurityResult":
+        return cls(
+            network=d.get("network", ""),
+            consensus=d.get("consensus", ""),
+            raw=RawMeasure.from_dict(d.get("raw") or {}),
+            value_usd=d.get("valueUsd", ""),
+            value_usd_formatted=d.get("valueUsdFormatted", ""),
+            label=d.get("label", ""),
+            normalized_score=d.get("normalizedScore", ""),
+            since=d.get("since", 0),
+            as_of=d.get("asOf", 0),
+        )
+
+
+@dataclass
+class ReportAnchorInput:
+    """One anchor supplied in a report request. Provide either ``network``
+    ("BTC"/"ETH") or the public-chain numeric ``chain_id``; network wins."""
+
+    tx_hash: str
+    timestamp: int  # anchor time, unix seconds
+    network: Optional[str] = None
+    chain_id: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        d: Dict[str, Any] = {"txHash": self.tx_hash, "timestamp": self.timestamp}
+        if self.network:
+            d["network"] = self.network
+        if self.chain_id:
+            d["chainId"] = self.chain_id
+        return d
+
+
+@dataclass
+class ReportRequest:
+    """Body of ``ProofMeasureClient.report``: a transaction's interchain anchors."""
+
+    transaction_id: str
+    prime_id: str
+    block_id: str
+    anchors: List[ReportAnchorInput] = field(default_factory=list)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "transactionId": self.transaction_id,
+            "primeId": self.prime_id,
+            "blockId": self.block_id,
+            "anchors": [a.to_dict() for a in self.anchors],
+        }
+
+
+@dataclass
+class AnchorSecurity:
+    """One interchain anchor with the security its public network has accumulated
+    since the anchor was placed."""
+
+    network: str = ""
+    anchor_timestamp: int = 0  # unix seconds
+    anchor_tx_hash: str = ""
+    security: SecurityResult = field(default_factory=SecurityResult)
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "AnchorSecurity":
+        return cls(
+            network=d.get("network", ""),
+            anchor_timestamp=d.get("anchorTimestamp", 0),
+            anchor_tx_hash=d.get("anchorTxHash", ""),
+            security=SecurityResult.from_dict(d.get("security") or {}),
+        )
+
+
+@dataclass
+class HashPower:
+    """Combined raw hash power across a report's PoW anchors."""
+
+    value: str = ""
+    units: str = ""
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "HashPower":
+        return cls(value=d.get("value", ""), units=d.get("units", ""))
+
+
+@dataclass
+class TransactionReport:
+    """Per-transaction "securedBy" report: every public-chain anchor covering the
+    transaction's block with both raw and USD security, plus combined totals.
+    ``hash_power`` is None when there are no PoW anchors."""
+
+    transaction_id: str = ""
+    prime_id: str = ""
+    block_id: str = ""
+    anchors: List[AnchorSecurity] = field(default_factory=list)
+    total_value_usd: str = ""
+    total_value_usd_formatted: str = ""
+    hash_power: Optional[HashPower] = None
+    total_normalized_score: str = ""
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "TransactionReport":
+        hp = d.get("hashPower")
+        return cls(
+            transaction_id=d.get("transactionId", ""),
+            prime_id=d.get("primeId", ""),
+            block_id=d.get("blockId", ""),
+            anchors=[AnchorSecurity.from_dict(a) for a in (d.get("anchors") or [])],
+            total_value_usd=d.get("totalValueUsd", ""),
+            total_value_usd_formatted=d.get("totalValueUsdFormatted", ""),
+            hash_power=HashPower.from_dict(hp) if hp else None,
+            total_normalized_score=d.get("totalNormalizedScore", ""),
+        )
+
+
+@dataclass
+class HealthResponse:
+    """proof-measure liveness payload."""
+
+    status: str = ""
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "HealthResponse":
+        return cls(status=d.get("status", ""))

prime_sdk/proof_measure.py

@@ -0,0 +1,64 @@
+"""Client for the Dragonchain proof-measure service.
+
+proof-measure is the measured-immutability / "securedBy" metric for L1–L5
+verification chains. It is a separate, public, UNauthenticated service (no API
+keys), so this client needs only a base URL. It exposes a network's accumulated
+security as both a raw measure (cumulative hashes / stake-seconds) and a USD
+valuation, plus a per-transaction "securedBy" report over interchain anchors.
+"""
+
+from typing import Optional
+
+import requests
+
+from .client import DEFAULT_TIMEOUT
+from .models import (
+    CONTENT_TYPE_JSON,
+    HealthResponse,
+    ReportRequest,
+    SecurityResult,
+    TransactionReport,
+)
+from .unauthenticated_client import UnauthenticatedClient
+
+#: Public production proof-measure endpoint.
+DEFAULT_BASE_URL = "https://proof-measure.dragonchain.com"
+
+
+class ProofMeasureClient:
+    """Calls the proof-measure HTTP API.
+
+    With no arguments it targets the public production endpoint, so
+    ``ProofMeasureClient()`` works with zero configuration.
+    """
+
+    def __init__(
+        self,
+        base_url: str = DEFAULT_BASE_URL,
+        timeout: int = DEFAULT_TIMEOUT,
+        session: "Optional[requests.Session]" = None,
+    ):
+        self._client = UnauthenticatedClient(
+            base_url or DEFAULT_BASE_URL, timeout=timeout, session=session
+        )
+
+    def get_security(self, network: str, since: Optional[int] = None) -> SecurityResult:
+        """Return the security a public network (``network`` = "BTC" or "ETH")
+        has accumulated since ``since`` (unix seconds), as a raw measure + USD
+        valuation. Omit ``since`` to use the service's default window."""
+        path = f"/api/v1/security/{network}"
+        if since and since > 0:
+            path += f"?since={since}"
+        return self._client.get(path, SecurityResult)
+
+    def report(self, req: ReportRequest) -> TransactionReport:
+        """Compute the per-transaction "securedBy" report for the supplied
+        interchain anchors: each anchor's raw + USD security since it was placed,
+        plus combined totals."""
+        return self._client.post(
+            "/api/v1/report", CONTENT_TYPE_JSON, req, TransactionReport
+        )
+
+    def health(self) -> HealthResponse:
+        """Report service liveness and DB reachability."""
+        return self._client.get("/api/v1/health", HealthResponse)

prime_sdk/transaction.py

@@ -1,6 +1,9 @@
 """Transaction endpoints."""
 
+from typing import List, Optional
+
 from .client import CONTENT_TYPE_JSON, Client
+from .interchain import interchain_query
 from .models import (
     InterchainTrace,
     ListTransactionsResponse,
@@ -40,13 +43,25 @@ class TransactionClient:
     def get(self, transaction_id: str) -> Transaction:
         return self._client.get(f"/api/v1/transaction/{transaction_id}", Transaction)
 
-    def get_interchain(self, transaction_id: str) -> InterchainTrace:
+    def get_interchain(
+        self,
+        transaction_id: str,
+        per_chain: Optional[int] = None,
+        chains: Optional[List[str]] = None,
+    ) -> InterchainTrace:
         """Trace a transaction to the validator (verification) blocks that
         validated its prime block and the public-chain interchain anchors those
         validator blocks were bundled into. If the transaction is still pending
-        (not yet in a block) the trace's lists are empty."""
+        (not yet in a block) the trace's lists are empty.
+
+        By default returns the first anchor per public chain (anchor proofs are
+        chained, so the earliest per chain is the meaningful one). ``per_chain``
+        caps anchors per chain (1 = first per chain; 0 = all); ``chains``
+        restricts to specific chain ids."""
         return self._client.get(
-            f"/api/v1/transaction/{transaction_id}/interchain", InterchainTrace
+            f"/api/v1/transaction/{transaction_id}/interchain"
+            f"{interchain_query(per_chain, chains)}",
+            InterchainTrace,
         )
 
     def list(self) -> ListTransactionsResponse:

prime_sdk/unauthenticated_client.py

@@ -0,0 +1,89 @@
+"""HTTP client for public Dragonchain services that require no credentials.
+
+Mirrors :class:`prime_sdk.client.Client`'s request plumbing — body marshaling,
+``allow_redirects=False``, status handling, and ``from_dict`` decoding — but
+sends no DC1-HMAC-SHA256 Authorization/Dragonchain/Timestamp headers. Used by the
+proof-measure client, which talks to a separate, public, unauthenticated service.
+"""
+
+import json
+from typing import Any, Optional, Type, TypeVar
+
+import requests
+
+from .client import CONTENT_TYPE_JSON, DEFAULT_TIMEOUT, _to_jsonable
+from .errors import DragonchainAPIError
+
+T = TypeVar("T")
+
+
+class UnauthenticatedClient:
+    """Unauthenticated HTTP client for a single public service base URL."""
+
+    def __init__(
+        self,
+        base_url: str,
+        timeout: int = DEFAULT_TIMEOUT,
+        session: Optional[requests.Session] = None,
+    ):
+        self.base_url = base_url.rstrip("/")
+        self.timeout = timeout
+        # A caller may inject a pre-configured Session (e.g. one with an
+        # SSRF-guarded transport adapter) when base_url is attacker-influenced.
+        self._session = session if session is not None else requests.Session()
+
+    def endpoint(self) -> str:
+        return self.base_url
+
+    def _do_request(
+        self,
+        method: str,
+        path: str,
+        content_type: str = "",
+        body: Any = None,
+        response_cls: Optional[Type[T]] = None,
+    ) -> Optional[T]:
+        if isinstance(body, (bytes, bytearray)):
+            body_bytes = bytes(body)
+        elif body is not None:
+            body_bytes = json.dumps(_to_jsonable(body)).encode("utf-8")
+            content_type = CONTENT_TYPE_JSON
+        else:
+            body_bytes = b""
+
+        headers = {"Accept": CONTENT_TYPE_JSON}
+        if content_type:
+            headers["Content-Type"] = content_type
+
+        url = self.base_url + path
+        # allow_redirects=False for the same reason as the authenticated client:
+        # a JSON API never legitimately 3xx-redirects, and following one is an
+        # SSRF vector.
+        resp = self._session.request(
+            method,
+            url,
+            data=body_bytes,
+            headers=headers,
+            timeout=self.timeout,
+            allow_redirects=False,
+        )
+
+        resp_text = resp.text
+        if resp.status_code >= 400:
+            raise DragonchainAPIError(resp.status_code, resp_text.strip())
+
+        if response_cls is not None and resp_text:
+            return response_cls.from_dict(json.loads(resp_text))  # type: ignore[attr-defined]
+        return None
+
+    def get(self, path: str, response_cls: Optional[Type[T]] = None) -> Optional[T]:
+        return self._do_request("GET", path, response_cls=response_cls)
+
+    def post(
+        self,
+        path: str,
+        content_type: str,
+        body: Any,
+        response_cls: Optional[Type[T]] = None,
+    ) -> Optional[T]:
+        return self._do_request("POST", path, content_type, body, response_cls)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "prime-sdk-python"
-version = "0.2.0"
+version = "0.5.0"
 description = "A self-contained Python SDK for interacting with Dragonchain nodes."
 readme = "README.md"
 requires-python = ">=3.9"

tests/test_client.py

@@ -93,8 +93,16 @@ class _FakeSession:
         self.response = response
         self.calls = []
 
-    def request(self, method, url, data=None, headers=None, timeout=None):
+    def request(self, method, url, data=None, headers=None, timeout=None, allow_redirects=None):
-        self.calls.append({"method": method, "url": url, "data": data, "headers": headers})
+        self.calls.append(
+            {
+                "method": method,
+                "url": url,
+                "data": data,
+                "headers": headers,
+                "allow_redirects": allow_redirects,
+            }
+        )
         return self.response
 
 

tests/test_interchain.py

@@ -0,0 +1,22 @@
+"""Offline tests for the interchain-trace query builder."""
+
+from prime_sdk import interchain_query
+
+
+def test_empty_when_nothing_set():
+    assert interchain_query() == ""
+    assert interchain_query(None, None) == ""
+    assert interchain_query(chains=[]) == ""
+
+
+def test_per_chain_including_zero():
+    assert interchain_query(per_chain=1) == "?perChain=1"
+    assert interchain_query(per_chain=0) == "?perChain=0"
+
+
+def test_chains_joined_with_commas():
+    assert interchain_query(chains=["1", "0"]) == "?chains=1,0"
+
+
+def test_combined():
+    assert interchain_query(per_chain=2, chains=["1", "0"]) == "?perChain=2&chains=1,0"

tests/test_proof_measure.py

@@ -0,0 +1,145 @@
+"""Offline tests for the proof-measure client. No network access required."""
+
+import json
+
+import pytest
+
+from prime_sdk import (
+    DragonchainAPIError,
+    ProofMeasureClient,
+    ReportAnchorInput,
+    ReportRequest,
+)
+from prime_sdk.proof_measure import DEFAULT_BASE_URL
+
+
+class _FakeResponse:
+    def __init__(self, status_code, text):
+        self.status_code = status_code
+        self.text = text
+
+
+class _FakeSession:
+    def __init__(self, response):
+        self.response = response
+        self.calls = []
+
+    def request(self, method, url, data=None, headers=None, timeout=None, allow_redirects=None):
+        self.calls.append(
+            {
+                "method": method,
+                "url": url,
+                "data": data,
+                "headers": headers,
+                "allow_redirects": allow_redirects,
+            }
+        )
+        return self.response
+
+
+def _client_with(session) -> ProofMeasureClient:
+    pm = ProofMeasureClient()
+    pm._client._session = session
+    return pm
+
+
+def test_default_base_url():
+    assert DEFAULT_BASE_URL == "https://proof-measure.dragonchain.com"
+    assert ProofMeasureClient()._client.endpoint() == DEFAULT_BASE_URL
+
+
+def test_get_security_builds_path_and_parses():
+    body = json.dumps(
+        {
+            "network": "BTC",
+            "consensus": "pow",
+            "raw": {"value": "71.63", "unit": "Yottahashes", "base": "7.16e25"},
+            "valueUsd": "13928170.26",
+            "valueUsdFormatted": "$13,928,170.26",
+            "label": "$13,928,170.26 energy consumed",
+            "normalizedScore": "0.9234",
+            "since": 1780504894,
+            "asOf": 1780591253,
+        }
+    )
+    fake = _FakeSession(_FakeResponse(200, body))
+    res = _client_with(fake).get_security("BTC", 1780504894)
+
+    call = fake.calls[0]
+    assert call["method"] == "GET"
+    assert call["url"] == DEFAULT_BASE_URL + "/api/v1/security/BTC?since=1780504894"
+    assert call["allow_redirects"] is False
+    # no auth headers on the unauthenticated client
+    assert "Authorization" not in call["headers"]
+    assert res.network == "BTC"
+    assert res.consensus == "pow"
+    assert res.raw.unit == "Yottahashes"
+    assert res.value_usd == "13928170.26"
+    assert res.value_usd_formatted == "$13,928,170.26"
+    assert res.since == 1780504894 and res.as_of == 1780591253
+
+
+def test_get_security_omits_since_when_not_positive():
+    fake = _FakeSession(_FakeResponse(200, json.dumps({"network": "ETH", "consensus": "pos"})))
+    _client_with(fake).get_security("ETH")
+    assert fake.calls[0]["url"] == DEFAULT_BASE_URL + "/api/v1/security/ETH"
+
+
+def test_report_posts_request_and_parses():
+    resp_body = json.dumps(
+        {
+            "transactionId": "tx-1",
+            "primeId": "p-1",
+            "blockId": "42",
+            "anchors": [
+                {
+                    "network": "BTC",
+                    "anchorTimestamp": 1712345678,
+                    "anchorTxHash": "0xabc",
+                    "security": {"network": "BTC", "consensus": "pow", "valueUsd": "100.00"},
+                }
+            ],
+            "totalValueUsd": "100.00",
+            "totalValueUsdFormatted": "$100.00",
+            "hashPower": {"value": "40.06", "units": "Yottahashes"},
+            "totalNormalizedScore": "0.9",
+        }
+    )
+    fake = _FakeSession(_FakeResponse(200, resp_body))
+    req = ReportRequest(
+        transaction_id="tx-1",
+        prime_id="p-1",
+        block_id="42",
+        anchors=[ReportAnchorInput(tx_hash="0xabc", timestamp=1712345678, network="BTC")],
+    )
+    rep = _client_with(fake).report(req)
+
+    call = fake.calls[0]
+    assert call["method"] == "POST"
+    assert call["url"] == DEFAULT_BASE_URL + "/api/v1/report"
+    assert call["headers"]["Content-Type"] == "application/json"
+    sent = json.loads(call["data"])
+    assert sent == {
+        "transactionId": "tx-1",
+        "primeId": "p-1",
+        "blockId": "42",
+        "anchors": [{"txHash": "0xabc", "timestamp": 1712345678, "network": "BTC"}],
+    }
+    assert rep.total_value_usd_formatted == "$100.00"
+    assert rep.hash_power is not None and rep.hash_power.units == "Yottahashes"
+    assert len(rep.anchors) == 1 and rep.anchors[0].security.network == "BTC"
+
+
+def test_health_parses():
+    fake = _FakeSession(_FakeResponse(200, json.dumps({"status": "ok"})))
+    res = _client_with(fake).health()
+    assert fake.calls[0]["url"] == DEFAULT_BASE_URL + "/api/v1/health"
+    assert res.status == "ok"
+
+
+def test_error_status_raises():
+    fake = _FakeSession(_FakeResponse(400, "  bad request  "))
+    with pytest.raises(DragonchainAPIError) as exc:
+        _client_with(fake).health()
+    assert exc.value.status_code == 400
+    assert exc.value.body == "bad request"