File size: 2,834 Bytes
62bb9d8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
"""Simplified tests for WebSocket feature flags functionality."""
from comfy_api import feature_flags


class TestWebSocketFeatureFlags:
    """Test suite for WebSocket feature flags integration."""

    def test_server_feature_flags_response(self):
        """Test server feature flags are properly formatted."""
        features = feature_flags.get_server_features()

        # Check expected server features
        assert "supports_preview_metadata" in features
        assert features["supports_preview_metadata"] is True
        assert "max_upload_size" in features
        assert isinstance(features["max_upload_size"], (int, float))

    def test_progress_py_checks_feature_flags(self):
        """Test that progress.py checks feature flags before sending metadata."""
        # This simulates the check in progress.py
        client_id = "test_client"
        sockets_metadata = {"test_client": {"feature_flags": {}}}

        # The actual check would be in progress.py
        supports_metadata = feature_flags.supports_feature(
            sockets_metadata, client_id, "supports_preview_metadata"
        )

        assert supports_metadata is False

    def test_multiple_clients_different_features(self):
        """Test handling multiple clients with different feature support."""
        sockets_metadata = {
            "modern_client": {
                "feature_flags": {"supports_preview_metadata": True}
            },
            "legacy_client": {
                "feature_flags": {}
            }
        }

        # Check modern client
        assert feature_flags.supports_feature(
            sockets_metadata, "modern_client", "supports_preview_metadata"
        ) is True

        # Check legacy client
        assert feature_flags.supports_feature(
            sockets_metadata, "legacy_client", "supports_preview_metadata"
        ) is False

    def test_feature_negotiation_message_format(self):
        """Test the format of feature negotiation messages."""
        # Client message format
        client_message = {
            "type": "feature_flags",
            "data": {
                "supports_preview_metadata": True,
                "api_version": "1.0.0"
            }
        }

        # Verify structure
        assert client_message["type"] == "feature_flags"
        assert "supports_preview_metadata" in client_message["data"]

        # Server response format (what would be sent)
        server_features = feature_flags.get_server_features()
        server_message = {
            "type": "feature_flags",
            "data": server_features
        }

        # Verify structure
        assert server_message["type"] == "feature_flags"
        assert "supports_preview_metadata" in server_message["data"]
        assert server_message["data"]["supports_preview_metadata"] is True