stack/tests/api/test_rustfs.py

"""Tests for RustFSClient and RustFSAdmin."""

from __future__ import annotations

from api.clients.rustfs import RustFSAdmin, RustFSClient


class TestRustFSClientRoutes:
    def test_list_buckets(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.list_buckets()
        req = cap.requests[0]
        assert req.method == "GET"
        assert req.url.path == "/"

    def test_create_bucket(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.create_bucket("my-bucket")
        req = cap.requests[0]
        assert req.method == "PUT"
        assert req.url.path == "/my-bucket"

    def test_put_object(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.put_object("bkt", "key.txt", b"hello")
        req = cap.requests[0]
        assert req.method == "PUT"
        assert req.url.path == "/bkt/key.txt"

    def test_get_object(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.get_object("bkt", "file.csv")
        req = cap.requests[0]
        assert req.method == "GET"
        assert req.url.path == "/bkt/file.csv"

    def test_delete_object(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.delete_object("bkt", "old.txt")
        req = cap.requests[0]
        assert req.method == "DELETE"
        assert req.url.path == "/bkt/old.txt"

    def test_bucket_exists(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        result = c.bucket_exists("my-bucket")
        assert result is True
        req = cap.requests[0]
        assert req.method == "HEAD"
        assert req.url.path == "/my-bucket"

    def test_delete_bucket(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.delete_bucket("old-bucket")
        req = cap.requests[0]
        assert req.method == "DELETE"
        assert req.url.path == "/old-bucket"

    def test_list_objects(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.list_objects("bkt")
        req = cap.requests[0]
        assert req.method == "GET"
        assert req.url.path == "/bkt"

    def test_list_objects_with_prefix(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.list_objects("bkt", prefix="data/")
        req = cap.requests[0]
        assert "prefix=data" in str(req.url)

    def test_head_object(self, capture_transport):
        cap = capture_transport
        c = RustFSClient("ak", "sk", _transport=cap.transport())
        c.head_object("bkt", "key.csv")
        req = cap.requests[0]
        assert req.method == "HEAD"
        assert req.url.path == "/bkt/key.csv"


class TestRustFSAdminRoutes:
    def test_list_users(self, capture_transport):
        cap = capture_transport
        c = RustFSAdmin("ak", "sk", _transport=cap.transport({}))
        c.list_users()
        req = cap.requests[0]
        assert req.method == "GET"
        assert req.url.path == "/minio/v2/iam/users"

    def test_add_user(self, capture_transport):
        cap = capture_transport
        c = RustFSAdmin("ak", "sk", _transport=cap.transport({}))
        c.add_user("new-ak", "new-sk")
        req = cap.requests[0]
        assert req.method == "POST"
        assert req.url.path == "/minio/v2/iam/users"

    def test_server_info(self, capture_transport):
        cap = capture_transport
        c = RustFSAdmin("ak", "sk", _transport=cap.transport({"nodes": 1}))
        result = c.server_info()
        assert result == {"nodes": 1}
        assert req_path(cap) == "/minio/v2/cluster/info"

    def test_remove_user(self, capture_transport):
        cap = capture_transport
        c = RustFSAdmin("ak", "sk", _transport=cap.transport({}))
        c.remove_user("old-ak")
        req = cap.requests[0]
        assert req.method == "DELETE"
        assert req.url.path == "/minio/v2/iam/users/old-ak"

    def test_list_policies(self, capture_transport):
        cap = capture_transport
        c = RustFSAdmin("ak", "sk", _transport=cap.transport({}))
        c.list_policies()
        req = cap.requests[0]
        assert req.method == "GET"
        assert req.url.path == "/minio/v2/iam/policies"

    def test_add_policy(self, capture_transport):
        cap = capture_transport
        c = RustFSAdmin("ak", "sk", _transport=cap.transport({}))
        c.add_policy("readonly", {"Version": "2012-10-17"})
        req = cap.requests[0]
        assert req.method == "POST"
        assert req.url.path == "/minio/v2/iam/policies/readonly"

    def test_set_user_policy(self, capture_transport):
        cap = capture_transport
        c = RustFSAdmin("ak", "sk", _transport=cap.transport())
        c.set_user_policy("user-ak", "readonly")
        req = cap.requests[0]
        assert req.method == "PUT"
        assert req.url.path == "/minio/v2/iam/users/user-ak/policies"


def req_path(cap):
    return cap.requests[0].url.path