weedforge 0.1.0

Rust-first, Python-friendly SDK for SeaweedFS
Documentation 
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

"""Python tests for weedforge.

These tests require a running SeaweedFS instance.
Run with: pytest tests/python_test.py -v
"""

import os
import pytest

# Skip all tests if weedforge is not installed
pytest.importorskip("weedforge")

from weedforge import WeedClient, FileId


class TestFileId:
    """Tests for FileId class."""

    def test_parse_valid(self):
        """Test parsing a valid file ID."""
        fid = FileId.parse("3,0000016300007037")
        assert fid.volume_id == 3
        assert isinstance(fid.file_key, int)
        assert isinstance(fid.cookie, int)

    def test_parse_invalid(self):
        """Test parsing an invalid file ID."""
        with pytest.raises(RuntimeError):
            FileId.parse("invalid")

    def test_render(self):
        """Test rendering a file ID."""
        fid = FileId.parse("3,0000016300007037")
        rendered = fid.render()
        assert rendered.startswith("3,")

    def test_str(self):
        """Test string representation."""
        fid = FileId.parse("3,0000016300007037")
        assert str(fid).startswith("3,")

    def test_repr(self):
        """Test repr."""
        fid = FileId.parse("3,0000016300007037")
        assert "FileId" in repr(fid)
        assert "volume_id=" in repr(fid)


class TestWeedClient:
    """Tests for WeedClient class."""

    def test_create_client(self):
        """Test creating a client."""
        client = WeedClient(master_urls=["http://localhost:9333"])
        assert client is not None

    def test_create_client_with_strategy(self):
        """Test creating a client with different strategies."""
        for strategy in ["round_robin", "failover", "random"]:
            client = WeedClient(
                master_urls=["http://localhost:9333"],
                strategy=strategy,
            )
            assert client is not None

    def test_create_client_invalid_strategy(self):
        """Test creating a client with invalid strategy."""
        with pytest.raises(RuntimeError):
            WeedClient(
                master_urls=["http://localhost:9333"],
                strategy="invalid",
            )

    def test_parse_file_id(self):
        """Test static parse_file_id method."""
        fid = WeedClient.parse_file_id("3,0000016300007037")
        assert fid.volume_id == 3


@pytest.mark.skipif(
    not os.environ.get("SEAWEEDFS_MASTER"),
    reason="SEAWEEDFS_MASTER not set",
)
class TestIntegration:
    """Integration tests requiring a running SeaweedFS instance."""

    @pytest.fixture
    def client(self):
        """Create a test client."""
        master_url = os.environ.get("SEAWEEDFS_MASTER", "http://localhost:9333")
        return WeedClient(master_urls=[master_url])

    def test_write_read_delete(self, client):
        """Test write, read, and delete operations."""
        # Write
        data = b"Hello, SeaweedFS!"
        fid = client.write(data, filename="test.txt")
        assert fid is not None
        assert fid.volume_id > 0

        # Read
        downloaded = client.read(fid)
        assert downloaded == data

        # Read with string fid
        downloaded2 = client.read(str(fid))
        assert downloaded2 == data

        # Public URL
        url = client.public_url(fid)
        assert url.startswith("http")

        # Delete
        client.delete(fid)

    def test_upload_bytes_alias(self, client):
        """Test upload_bytes alias."""
        data = b"Test data"
        fid = client.upload_bytes(data, filename="alias_test.txt")
        assert fid is not None

        # Cleanup
        client.delete(fid)

    def test_public_url_resized(self, client):
        """Test public URL with resize parameters."""
        data = b"fake image data"
        fid = client.write(data, filename="image.jpg")

        url = client.public_url_resized(fid, width=200, height=200)
        assert "width=200" in url
        assert "height=200" in url

        # Cleanup
        client.delete(fid)


if __name__ == "__main__":
    pytest.main([__file__, "-v"])