scripts/__tests__/test_sync_broker_sdk_refs.py

Rama main · réplica de lectura del repositorio original.

from __future__ import annotations

import hashlib
import json
import textwrap

import pytest

from scripts import sync_broker_sdk_refs as syncer


def test_active_sdk_entries_ignores_placeholders_and_unknowns() -> None:
    entries = [
        {"name": "dhanhq", "version": "2.2.0"},
        {"name": "future-sdk", "version": "PLACEHOLDER"},
        {"name": "unknown-sdk", "version": "1.0.0"},
        {"name": "kotakneoapi", "version": "3.0.7", "release_version": "3.0.7"},
    ]

    assert syncer.active_sdk_entries(entries) == [
        {"name": "dhanhq", "version": "2.2.0", "source_commit": ""},
        {"name": "kotakneoapi", "version": "3.0.7", "release_version": "3.0.7", "source_commit": ""},
    ]


@pytest.mark.parametrize("release_version", [None, "", "v3.0.7", "3.0.7rc1"])
def test_kotak_reference_sync_requires_an_explicit_stable_release_version(release_version):
    entry = {"name": "kotakneoapi", "version": "3.0.8"}
    if release_version is not None:
        entry["release_version"] = release_version

    with pytest.raises(ValueError, match="release_version"):
        syncer.active_sdk_entries([entry])


def test_select_pypi_artifact_prefers_universal_py3_wheel() -> None:
    files = [
        {"filename": "pkg-1.0.0.tar.gz", "packagetype": "sdist", "python_version": "source"},
        {"filename": "pkg-1.0.0-cp312-macosx.whl", "packagetype": "bdist_wheel", "python_version": "cp312"},
        {"filename": "pkg-1.0.0-py3-none-any.whl", "packagetype": "bdist_wheel", "python_version": "py3"},
    ]

    assert syncer.select_pypi_artifact(files)["filename"] == "pkg-1.0.0-py3-none-any.whl"


def test_download_pypi_artifact_verifies_digest(tmp_path) -> None:
    body = b"wheel-bytes"
    digest = hashlib.sha256(body).hexdigest()
    release = {
        "releases": {
            "1.0.0": [
                {
                    "filename": "pkg-1.0.0-py3-none-any.whl",
                    "packagetype": "bdist_wheel",
                    "python_version": "py3",
                    "url": "https://files.example/pkg.whl",
                    "digests": {"sha256": digest},
                }
            ]
        }
    }

    def opener(url: str) -> tuple[int, bytes]:
        if url == "https://pypi.org/pypi/pkg/json":
            return 200, json.dumps(release).encode("utf-8")
        if url == "https://files.example/pkg.whl":
            return 200, body
        raise AssertionError(url)

    result = syncer.download_pypi_artifact("pkg", "1.0.0", tmp_path, opener)

    assert result["filename"] == "pkg-1.0.0-py3-none-any.whl"
    assert result["sha256"] == digest
    assert (tmp_path / "pkg-1.0.0-py3-none-any.whl").read_bytes() == body


def test_latest_pypi_version_reads_info_version() -> None:
    def opener(url: str) -> tuple[int, bytes]:
        assert url == "https://pypi.org/pypi/pkg/json"
        return 200, json.dumps({"info": {"version": "1.2.3"}}).encode("utf-8")

    assert syncer.latest_pypi_version("pkg", opener) == "1.2.3"


def test_load_broker_lock_reads_toml_entries(tmp_path) -> None:
    lock = tmp_path / "brokers.lock"
    lock.write_text(
        textwrap.dedent(
            """
            [[broker]]
            name = "dhanhq"
            version = "2.2.0"

            [[broker]]
            name = "growwapi"
            version = "1.5.0"
            """
        ),
        encoding="utf-8",
    )

    assert [entry["name"] for entry in syncer.load_broker_lock(lock)] == ["dhanhq", "growwapi"]


def test_sync_records_pypi_drift(tmp_path, monkeypatch) -> None:
    body = b"wheel-bytes"
    digest = hashlib.sha256(body).hexdigest()
    release = {
        "info": {"version": "2.3.0"},
        "releases": {
            "2.2.0": [
                {
                    "filename": "dhanhq-2.2.0-py3-none-any.whl",
                    "packagetype": "bdist_wheel",
                    "python_version": "py3",
                    "url": "https://files.example/dhanhq.whl",
                    "digests": {"sha256": digest},
                }
            ]
        },
    }

    def opener(url: str) -> tuple[int, bytes]:
        if url == "https://pypi.org/pypi/dhanhq/json":
            return 200, json.dumps(release).encode("utf-8")
        if url == "https://files.example/dhanhq.whl":
            return 200, body
        raise AssertionError(url)

    monkeypatch.setattr(syncer, "load_broker_lock", lambda _path: [{"name": "dhanhq", "version": "2.2.0"}])
    monkeypatch.setattr(
        syncer,
        "sync_git_mirror",
        lambda _source, _audit_root: {"path": "repo", "remote": "origin", "head": "abc", "describe": "abc"},
    )

    manifest = syncer.sync(audit_root=tmp_path / "audit", check_missing=False, opener=opener)

    assert manifest["sdks"][0]["pypi_latest_version"] == "2.3.0"
    assert manifest["sdks"][0]["pypi_current"] is False
    assert "uv run python scripts/sync_broker_sdk_refs.py" in (tmp_path / "audit" / "README.md").read_text(
        encoding="utf-8"
    )
    assert manifest["drift"] == [{"package": "dhanhq", "source": "pypi", "locked": "2.2.0", "upstream": "2.3.0"}]


def test_non_kotak_git_drift_still_tracks_remote_head(tmp_path, monkeypatch) -> None:
    digest = hashlib.sha256(b"wheel-bytes").hexdigest()
    release = {
        "info": {"version": "2.3.0"},
        "releases": {
            "2.3.0": [
                {
                    "filename": "dhanhq-2.3.0-py3-none-any.whl",
                    "packagetype": "bdist_wheel",
                    "python_version": "py3",
                    "url": "https://files.example/dhanhq.whl",
                    "digests": {"sha256": digest},
                }
            ]
        },
    }

    def opener(url: str) -> tuple[int, bytes]:
        if url == "https://pypi.org/pypi/dhanhq/json":
            return 200, json.dumps(release).encode("utf-8")
        if url == "https://files.example/dhanhq.whl":
            return 200, b"wheel-bytes"
        raise AssertionError(url)

    locked_commit = "a" * 40
    upstream_head = "b" * 40
    monkeypatch.setattr(
        syncer,
        "load_broker_lock",
        lambda _path: [{"name": "dhanhq", "version": "2.3.0", "source_commit": locked_commit, "sha256": digest}],
    )
    monkeypatch.setattr(
        syncer,
        "sync_git_mirror",
        lambda _source, _audit_root: {
            "path": "repo",
            "remote": "origin",
            "head": upstream_head,
            "describe": "v2.3.0",
        },
    )

    manifest = syncer.sync(audit_root=tmp_path / "audit", check_missing=False, opener=opener)

    assert manifest["sdks"][0]["git_current"] is False
    assert manifest["drift"] == [
        {"package": "dhanhq", "source": "git", "locked": locked_commit, "upstream": upstream_head}
    ]
    assert "locked behind upstream HEAD" in (tmp_path / "audit" / "README.md").read_text(encoding="utf-8")


@pytest.mark.parametrize(
    ("git_change", "pypi_version", "expected_source"),
    [
        ({}, "3.0.7", None),
        ({"head": "b" * 40, "head_tree": "1" * 40}, "3.0.7", None),
        ({"latest_stable_tag": "v3.0.8"}, "3.0.8", "release-tag"),
        ({"release_commit": "c" * 40}, "3.0.7", "release-commit"),
        ({"latest_stable_tag": "v3.0.10"}, "3.0.10", "release-tag"),
    ],
)
def test_kotak_dual_track_drift_is_independent(
    tmp_path,
    monkeypatch,
    git_change: dict[str, str],
    pypi_version: str,
    expected_source: str | None,
) -> None:
    wheel = b"release wheel"
    sdist = b"release sdist"
    wheel_hash = hashlib.sha256(wheel).hexdigest()
    sdist_hash = hashlib.sha256(sdist).hexdigest()
    locked = {
        "name": "kotakneoapi",
        "version": "3.0.8",
        "release_version": "3.0.7",
        "source_commit": "a" * 40,
        "source_tree": "d" * 40,
        "release_tag": "v3.0.7",
        "release_commit": "e" * 40,
        "release_tree": "f" * 40,
        "release_wheel_sha256": wheel_hash,
        "release_sdist_sha256": sdist_hash,
        "sha256": wheel_hash,
    }
    git = {
        "path": "mirror",
        "remote": "origin",
        "head": locked["source_commit"],
        "head_tree": locked["source_tree"],
        "describe": "v3.0.7",
        "latest_stable_tag": locked["release_tag"],
        "release_commit": locked["release_commit"],
        "release_tree": locked["release_tree"],
    }
    git.update(git_change)
    monkeypatch.setattr(syncer, "load_broker_lock", lambda _path: [locked])
    monkeypatch.setattr(syncer, "sync_git_mirror", lambda _source, _root: git)

    def opener(url: str) -> tuple[int, bytes]:
        if url == "https://pypi.org/pypi/kotakneoapi/json":
            data = {
                "info": {"version": pypi_version},
                "releases": {
                    "3.0.7": [
                        {
                            "filename": "kotakneoapi-3.0.7-py3-none-any.whl",
                            "packagetype": "bdist_wheel",
                            "python_version": "py3",
                            "url": "https://files.example/wheel",
                            "digests": {"sha256": wheel_hash},
                        },
                        {
                            "filename": "kotakneoapi-3.0.7.tar.gz",
                            "packagetype": "sdist",
                            "python_version": "source",
                            "url": "https://files.example/sdist",
                            "digests": {"sha256": sdist_hash},
                        },
                    ]
                },
            }
            return 200, json.dumps(data).encode()
        if url == "https://files.example/wheel":
            return 200, wheel
        if url == "https://files.example/sdist":
            return 200, sdist
        raise AssertionError(url)

    manifest = syncer.sync(audit_root=tmp_path / "audit", check_missing=False, opener=opener)

    sdk = manifest["sdks"][0]
    if expected_source is None:
        assert manifest["drift"] == []
        assert sdk["git_current"] is True
    else:
        assert any(d["package"] == "kotakneoapi" and d["source"] == expected_source for d in manifest["drift"])
        if expected_source == "release-commit":
            assert sdk["git_current"] is False
    if "head" in git_change:
        readme = (tmp_path / "audit" / "README.md").read_text(encoding="utf-8")
        assert "locked behind upstream HEAD" not in readme
        assert "git status: current" in readme
    assert sdk["version"] == "3.0.8"
    assert manifest["sdks"][0]["release_version"] == "3.0.7"
    assert manifest["sdks"][0]["pypi"]["filename"] == "kotakneoapi-3.0.7-py3-none-any.whl"