comfyui

1"""Integration tests against the live Comfy Cloud API.2 3These tests are auto-skipped when COMFY_CLOUD_API_KEY is not set.4They never SUBMIT workflows (would need a paid subscription) — they only5verify the read-only endpoints we rely on.6"""7 8from __future__ import annotations9 10import pytest11 12from _common import http_get, parse_model_list, resolve_url13 14 15pytestmark = pytest.mark.cloud16 17 18class TestCloudEndpointsLive:19    def test_system_stats_reachable(self, cloud_key):20        url = resolve_url("https://cloud.comfy.org", "/system_stats")21        r = http_get(url, headers={"X-API-Key": cloud_key})22        assert r.status == 20023        data = r.json()24        assert "system" in data25 26    def test_models_endpoint_routed_to_experiment(self, cloud_key):27        # We expect the skill to route /models/checkpoints → /api/experiment/models/checkpoints28        url = resolve_url("https://cloud.comfy.org", "/models/checkpoints")29        assert "/api/experiment/models/checkpoints" in url30        r = http_get(url, headers={"X-API-Key": cloud_key})31        assert r.status == 20032 33    def test_models_endpoint_returns_dicts(self, cloud_key):34        url = resolve_url("https://cloud.comfy.org", "/models/checkpoints")35        r = http_get(url, headers={"X-API-Key": cloud_key})36        data = r.json()37        assert isinstance(data, list)38        if data:39            # Cloud format: list of dicts with `name`40            assert isinstance(data[0], dict)41            assert "name" in data[0]42            # Our parser normalizes both43            normalized = parse_model_list(data)44            assert len(normalized) == len(data)45 46    def test_history_renamed_to_v2(self, cloud_key):47        # /history → /api/history_v2 on cloud48        url = resolve_url("https://cloud.comfy.org", "/history/some-fake-id")49        assert "/api/history_v2/some-fake-id" in url50 51    def test_object_info_paid_tier(self, cloud_key):52        # On free tier, /object_info returns 403 with a recognizable message53        url = resolve_url("https://cloud.comfy.org", "/object_info")54        r = http_get(url, headers={"X-API-Key": cloud_key})55        # Should be either 200 (paid) or 403 (free) — not 404 / 50056        assert r.status in {200, 403}57        if r.status == 403:58            # Body should mention the limitation59            assert "free tier" in r.text().lower() or "subscription" in r.text().lower()60 61 62class TestCloudCheckDepsLive:63    def test_check_deps_against_cloud(self, cloud_key, sd15_workflow):64        from check_deps import check_deps65        report = check_deps(sd15_workflow, host="https://cloud.comfy.org", api_key=cloud_key)66        # Either node check passed OR was skipped (free tier)67        assert "missing_models" in report68        assert "is_cloud" in report and report["is_cloud"] is True69 70    def test_flux_workflow_models_resolved_via_aliases(self, cloud_key, flux_workflow):71        """Flux uses unet/clip folders; cloud has them in diffusion_models/text_encoders.72        With folder aliasing, the check should still find them."""73        from check_deps import check_deps74        report = check_deps(flux_workflow, host="https://cloud.comfy.org", api_key=cloud_key)75        # The exact required Flux files (flux1-dev.safetensors, t5xxl_fp16, clip_l, ae)76        # are present on cloud; with folder aliasing, none should be missing.77        # If this fails, either the cloud removed the model or the aliasing logic broke.78        missing_filenames = {m["value"] for m in report["missing_models"]}79        assert "ae.safetensors" not in missing_filenames, \80            "ae.safetensors should be on cloud's vae folder"81        # t5xxl_fp16 / clip_l should be reachable via the clip → text_encoders alias82        # flux1-dev.safetensors likewise via unet → diffusion_models83 84 85class TestHealthCheckLive:86    def test_health_check_passes(self, cloud_key, capsys):87        from health_check import main as health_main88        rc = health_main(["--host", "https://cloud.comfy.org", "--api-key", cloud_key])89        captured = capsys.readouterr()90        # Should produce JSON91        import json92        report = json.loads(captured.out)93        assert report["server"]["reachable"] is True94        assert report["checkpoints"]["queryable"] is True95        assert report["checkpoints"]["count"] > 096