test_conformance.py

"""Byte-identical conformance against Swift-generated fixtures.

The fixture suite lives at QueueKit/Tests/QueueKitTests/Fixtures/ and
is generated by the Swift wire-format fixture generator (which constructs
Job and SignalFile values directly and encodes them with deterministic HLCs).
Python must produce byte-for-byte identical output.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from queuekit.job import (
    Job,
    SignalFile,
    HLC,
    ArtifactRef,
    ObservationStatus,
    encode_job,
    encode_signal,
    filename_for_job,
)


FIXTURES = (
    Path(__file__).resolve().parent.parent.parent
    / "Tests" / "QueueKitTests" / "Fixtures"
)


def _signed_node(unsigned: int) -> int:
    return unsigned if unsigned < 0x80000000 else unsigned - 0x1_0000_0000


def _load_job_input(name: str) -> Job:
    p = FIXTURES / f"{name}_input.json"
    obj = json.loads(p.read_bytes())
    payload_hex = obj.get("payload_bytes_hex", "")
    payload = bytes.fromhex(payload_hex) if payload_hex else b""
    sa = obj["submitted_at"]
    return Job(
        id=obj["id"],
        stream_id=obj["stream_id"],
        submitted_at=HLC(
            physical_time=sa["physical_time"],
            logical_count=sa["logical_count"],
            node_id=_signed_node(sa["node_id"]),
        ),
        priority=obj["priority"],
        payload=payload,
        extensions=obj.get("extensions", {}),
    )


def _load_signal_input(name: str) -> SignalFile:
    p = FIXTURES / f"signal_{name}_input.json"
    obj = json.loads(p.read_bytes())
    ca = obj["completed_at"]
    return SignalFile(
        job_id=obj["job_id"],
        status=ObservationStatus(obj["status"]),
        artifacts=[ArtifactRef(a["type"], a["value"])
                   for a in obj.get("artifacts", [])],
        completed_at=HLC(
            physical_time=ca["physical_time"],
            logical_count=ca["logical_count"],
            node_id=_signed_node(ca["node_id"]),
        ),
    )


@pytest.mark.skipif(
    not FIXTURES.exists(),
    reason=f"fixtures dir not present at {FIXTURES}",
)
@pytest.mark.parametrize(
    "name", ["job_001", "job_002", "job_003", "job_004", "job_005"])
def test_job_byte_identical(name):
    job = _load_job_input(name)
    expected = (FIXTURES / f"{name}_file.json").read_bytes()
    actual = encode_job(job)
    assert actual == expected, (
        f"Python encoded job differs from Swift fixture {name}:\n"
        f"  expected: {expected!r}\n"
        f"  actual:   {actual!r}")


@pytest.mark.skipif(
    not FIXTURES.exists(),
    reason=f"fixtures dir not present at {FIXTURES}",
)
@pytest.mark.parametrize(
    "name", ["job_001", "job_002", "job_003", "job_004", "job_005"])
def test_filename_byte_identical(name):
    job = _load_job_input(name)
    expected = (FIXTURES / f"{name}_filename.txt").read_text().strip()
    actual = filename_for_job(job)
    assert actual == expected


@pytest.mark.skipif(
    not FIXTURES.exists(),
    reason=f"fixtures dir not present at {FIXTURES}",
)
@pytest.mark.parametrize(
    "name", ["001", "002", "003", "004", "005"])
def test_signal_byte_identical(name):
    sig = _load_signal_input(name)
    expected = (FIXTURES / f"signal_{name}_output.json").read_bytes()
    actual = encode_signal(sig)
    assert actual == expected