Netbox-Store/host_agent/tests/test_catalog.py

from __future__ import annotations

import hashlib
import io
import tarfile
import tempfile
import unittest
from pathlib import Path

from support import FakeTransport, make_config, plugin_json, release_json, wheel_bytes

from netbox_store_agent.catalog import CatalogError, SecureHTTPTransport, StoreClient
from netbox_store_agent.config import StoreSettings


class CatalogTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary.name)
        self.config = make_config(self.root)

    def tearDown(self) -> None:
        self.temporary.cleanup()

    def test_trailing_slash_fallback_and_valid_wheel(self) -> None:
        artifact = wheel_bytes(self.root)
        release = release_json(
            sha256=hashlib.sha256(artifact).hexdigest(), artifact_size=len(artifact)
        )
        responses = {
            "http://store.test/api/v1/plugins/demo-plugin/": plugin_json(),
            "http://store.test/api/v1/plugins/demo-plugin/releases/1.2.3/": release,
        }
        transport = FakeTransport(responses, artifact)
        client = StoreClient(self.config, transport)
        plan = client.get_release("demo-plugin", "1.2.3")
        # Destination directory is caller-owned; use an existing operation directory.
        operation_dir = self.root / "operation"
        operation_dir.mkdir()
        downloaded = client.download_release(plan, operation_dir)
        self.assertTrue(downloaded.is_file())
        self.assertEqual(transport.json_urls[0][-1], "n")
        self.assertEqual(transport.json_urls[1][-1], "/")

    def test_release_falls_back_to_plugin_releases_array(self) -> None:
        release = release_json()
        transport = FakeTransport(
            {"http://store.test/api/v1/plugins/demo-plugin": plugin_json([release])}
        )
        plan = StoreClient(self.config, transport).get_release("demo-plugin", "1.2.3")
        self.assertEqual(plan.version, "1.2.3")

    def test_commit_bound_source_archive_is_verified(self) -> None:
        buffer = io.BytesIO()
        with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
            content = b"[build-system]\nrequires = []\n"
            info = tarfile.TarInfo("demo/pyproject.toml")
            info.size = len(content)
            archive.addfile(info, io.BytesIO(content))
        artifact = buffer.getvalue()
        commit = "d" * 40
        release = release_json(
            download_url=f"http://artifacts.test/demo/archive/{commit}.tar.gz",
            artifact_kind="source_archive",
            artifact_filename="demo-plugin-1.2.3-source.tar.gz",
            commit_sha=commit,
            sha256=hashlib.sha256(artifact).hexdigest(),
            artifact_size=len(artifact),
        )
        transport = FakeTransport(
            {
                "http://store.test/api/v1/plugins/demo-plugin": plugin_json(),
                "http://store.test/api/v1/plugins/demo-plugin/releases/1.2.3": release,
            },
            artifact,
        )
        client = StoreClient(self.config, transport)
        plan = client.get_release("demo-plugin", "1.2.3")
        directory = self.root / "source-operation"
        directory.mkdir()
        self.assertEqual(client.download_release(plan, directory).read_bytes(), artifact)

    def test_unapproved_or_mutable_release_rejected(self) -> None:
        for change in ({"approved": False}, {"immutable": False}, {"status": "pending"}):
            with self.subTest(change=change):
                transport = FakeTransport(
                    {
                        "http://store.test/api/v1/plugins/demo-plugin": plugin_json(),
                        "http://store.test/api/v1/plugins/demo-plugin/releases/1.2.3": release_json(
                            **change
                        ),
                    }
                )
                with self.assertRaises(CatalogError):
                    StoreClient(self.config, transport).get_release("demo-plugin", "1.2.3")

    def test_unknown_catalog_field_fails_closed(self) -> None:
        payload = plugin_json()
        payload["internal_id"] = 7
        transport = FakeTransport({"http://store.test/api/v1/plugins/demo-plugin": payload})
        with self.assertRaises(CatalogError):
            StoreClient(self.config, transport).get_plugin("demo-plugin")

    def test_netbox_incompatibility_rejected(self) -> None:
        transport = FakeTransport(
            {
                "http://store.test/api/v1/plugins/demo-plugin": plugin_json(
                    min_netbox_version="4.7.0"
                )
            }
        )
        with self.assertRaises(CatalogError):
            StoreClient(self.config, transport).get_plugin("demo-plugin")

    def test_wheel_distribution_must_match(self) -> None:
        artifact = wheel_bytes(self.root)
        release = release_json(
            sha256=hashlib.sha256(artifact).hexdigest(), artifact_size=len(artifact)
        )
        payload = plugin_json(package_name="another-package")
        transport = FakeTransport(
            {
                "http://store.test/api/v1/plugins/demo-plugin": payload,
                "http://store.test/api/v1/plugins/demo-plugin/releases/1.2.3": release,
            },
            artifact,
        )
        client = StoreClient(self.config, transport)
        plan = client.get_release("demo-plugin", "1.2.3")
        directory = self.root / "mismatch"
        directory.mkdir()
        with self.assertRaises(CatalogError):
            client.download_release(plan, directory)

    def test_bearer_token_is_not_sent_to_artifact_origin(self) -> None:
        settings = StoreSettings(
            **{
                **self.config.store.__dict__,
                "base_url": "https://store.test:8443",
                "allow_http_for_testing": False,
            }
        )
        transport = SecureHTTPTransport(settings)
        transport._token = lambda: "secret"  # type: ignore[method-assign]
        self.assertEqual(
            transport._headers("https://store.test:8443/api")["Authorization"],
            "Bearer secret",
        )
        self.assertNotIn(
            "Authorization", transport._headers("https://artifacts.test/release.whl")
        )
        self.assertNotIn("Authorization", transport._headers("https://store.test/api"))


if __name__ == "__main__":
    unittest.main()