kqjy/MyFSIO
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 33 Wiki Activity

Migrate UI backend from direct storage calls to S3 API proxy via boto3

Browse Source
This commit is contained in:
kqjy
2026-02-09 22:33:47 +08:00
parent 4ecd32a554
commit 1e3c4b545f
13 changed files with 1445 additions and 693 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
tests/test_ui_bulk_delete.py
View File

@@ -1,8 +1,12 @@
import io
import json
import threading
from pathlib import Path
from werkzeug.serving import make_server
from app import create_app
from app.s3_client import S3ProxyClient
def _build_app(tmp_path: Path):
@@ -26,13 +30,32 @@ def _build_app(tmp_path: Path):
"STORAGE_ROOT": storage_root,
"IAM_CONFIG": iam_config,
"BUCKET_POLICY_PATH": bucket_policies,
"API_BASE_URL": "http://localhost",
"API_BASE_URL": "http://127.0.0.1:0",
"SECRET_KEY": "testing",
"WTF_CSRF_ENABLED": False,
}
)
server = make_server("127.0.0.1", 0, app)
host, port = server.server_address
api_url = f"http://{host}:{port}"
app.config["API_BASE_URL"] = api_url
app.extensions["s3_proxy"] = S3ProxyClient(api_base_url=api_url)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
app._test_server = server
app._test_thread = thread
return app
def _shutdown_app(app):
if hasattr(app, "_test_server"):
app._test_server.shutdown()
app._test_thread.join(timeout=2)
def _login(client):
return client.post(
"/ui/login",
@@ -43,54 +66,60 @@ def _login(client):
def test_bulk_delete_json_route(tmp_path: Path):
app = _build_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("demo")
storage.put_object("demo", "first.txt", io.BytesIO(b"first"))
storage.put_object("demo", "second.txt", io.BytesIO(b"second"))
try:
storage = app.extensions["object_storage"]
storage.create_bucket("demo")
storage.put_object("demo", "first.txt", io.BytesIO(b"first"))
storage.put_object("demo", "second.txt", io.BytesIO(b"second"))
client = app.test_client()
assert _login(client).status_code == 200
client = app.test_client()
assert _login(client).status_code == 200
response = client.post(
"/ui/buckets/demo/objects/bulk-delete",
json={"keys": ["first.txt", "missing.txt"]},
headers={"X-Requested-With": "XMLHttpRequest"},
)
assert response.status_code == 200
payload = response.get_json()
assert payload["status"] == "ok"
assert set(payload["deleted"]) == {"first.txt", "missing.txt"}
assert payload["errors"] == []
response = client.post(
"/ui/buckets/demo/objects/bulk-delete",
json={"keys": ["first.txt", "missing.txt"]},
headers={"X-Requested-With": "XMLHttpRequest"},
)
assert response.status_code == 200
payload = response.get_json()
assert payload["status"] == "ok"
assert set(payload["deleted"]) == {"first.txt", "missing.txt"}
assert payload["errors"] == []
listing = storage.list_objects_all("demo")
assert {meta.key for meta in listing} == {"second.txt"}
listing = storage.list_objects_all("demo")
assert {meta.key for meta in listing} == {"second.txt"}
finally:
_shutdown_app(app)
def test_bulk_delete_validation(tmp_path: Path):
app = _build_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("demo")
storage.put_object("demo", "keep.txt", io.BytesIO(b"keep"))
try:
storage = app.extensions["object_storage"]
storage.create_bucket("demo")
storage.put_object("demo", "keep.txt", io.BytesIO(b"keep"))
client = app.test_client()
assert _login(client).status_code == 200
client = app.test_client()
assert _login(client).status_code == 200
bad_response = client.post(
"/ui/buckets/demo/objects/bulk-delete",
json={"keys": []},
headers={"X-Requested-With": "XMLHttpRequest"},
)
assert bad_response.status_code == 400
assert bad_response.get_json()["status"] == "error"
bad_response = client.post(
"/ui/buckets/demo/objects/bulk-delete",
json={"keys": []},
headers={"X-Requested-With": "XMLHttpRequest"},
)
assert bad_response.status_code == 400
assert bad_response.get_json()["status"] == "error"
too_many = [f"obj-{index}.txt" for index in range(501)]
limit_response = client.post(
"/ui/buckets/demo/objects/bulk-delete",
json={"keys": too_many},
headers={"X-Requested-With": "XMLHttpRequest"},
)
assert limit_response.status_code == 400
assert limit_response.get_json()["status"] == "error"
too_many = [f"obj-{index}.txt" for index in range(501)]
limit_response = client.post(
"/ui/buckets/demo/objects/bulk-delete",
json={"keys": too_many},
headers={"X-Requested-With": "XMLHttpRequest"},
)
assert limit_response.status_code == 400
assert limit_response.get_json()["status"] == "error"
still_there = storage.list_objects_all("demo")
assert {meta.key for meta in still_there} == {"keep.txt"}
still_there = storage.list_objects_all("demo")
assert {meta.key for meta in still_there} == {"keep.txt"}
finally:
_shutdown_app(app)

297
tests/test_ui_encryption.py
View File

@@ -1,10 +1,13 @@
"""Tests for UI-based encryption configuration."""
import json
import threading
from pathlib import Path
import pytest
from werkzeug.serving import make_server
from app import create_app
from app.s3_client import S3ProxyClient
def get_csrf_token(response):
@@ -37,212 +40,224 @@ def _make_encryption_app(tmp_path: Path, *, kms_enabled: bool = True):
]
}
iam_config.write_text(json.dumps(iam_payload))
config = {
"TESTING": True,
"STORAGE_ROOT": storage_root,
"IAM_CONFIG": iam_config,
"BUCKET_POLICY_PATH": bucket_policies,
"API_BASE_URL": "http://testserver",
"API_BASE_URL": "http://127.0.0.1:0",
"SECRET_KEY": "testing",
"ENCRYPTION_ENABLED": True,
"WTF_CSRF_ENABLED": False,
}
if kms_enabled:
config["KMS_ENABLED"] = True
config["KMS_KEYS_PATH"] = str(tmp_path / "kms_keys.json")
config["ENCRYPTION_MASTER_KEY_PATH"] = str(tmp_path / "master.key")
app = create_app(config)
server = make_server("127.0.0.1", 0, app)
host, port = server.server_address
api_url = f"http://{host}:{port}"
app.config["API_BASE_URL"] = api_url
app.extensions["s3_proxy"] = S3ProxyClient(api_base_url=api_url)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
app._test_server = server
app._test_thread = thread
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
return app
def _shutdown_app(app):
if hasattr(app, "_test_server"):
app._test_server.shutdown()
app._test_thread.join(timeout=2)
class TestUIBucketEncryption:
"""Test bucket encryption configuration via UI."""
def test_bucket_detail_shows_encryption_card(self, tmp_path):
"""Encryption card should be visible on bucket detail page."""
app = _make_encryption_app(tmp_path)
client = app.test_client()
try:
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/test-bucket?tab=properties")
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "Default Encryption" in html
assert "Encryption Algorithm" in html or "Default encryption disabled" in html
finally:
_shutdown_app(app)
response = client.get("/ui/buckets/test-bucket?tab=properties")
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "Default Encryption" in html
assert "Encryption Algorithm" in html or "Default encryption disabled" in html
def test_enable_aes256_encryption(self, tmp_path):
"""Should be able to enable AES-256 encryption."""
app = _make_encryption_app(tmp_path)
client = app.test_client()
try:
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/test-bucket?tab=properties")
csrf_token = get_csrf_token(response)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"action": "enable",
"algorithm": "AES256",
},
follow_redirects=True,
)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "AES-256" in html or "encryption enabled" in html.lower()
finally:
_shutdown_app(app)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"csrf_token": csrf_token,
"action": "enable",
"algorithm": "AES256",
},
follow_redirects=True,
)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "AES-256" in html or "encryption enabled" in html.lower()
def test_enable_kms_encryption(self, tmp_path):
"""Should be able to enable KMS encryption."""
app = _make_encryption_app(tmp_path, kms_enabled=True)
client = app.test_client()
try:
with app.app_context():
kms = app.extensions.get("kms")
if kms:
key = kms.create_key("test-key")
key_id = key.key_id
else:
pytest.skip("KMS not available")
with app.app_context():
kms = app.extensions.get("kms")
if kms:
key = kms.create_key("test-key")
key_id = key.key_id
else:
pytest.skip("KMS not available")
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"action": "enable",
"algorithm": "aws:kms",
"kms_key_id": key_id,
},
follow_redirects=True,
)
response = client.get("/ui/buckets/test-bucket?tab=properties")
csrf_token = get_csrf_token(response)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "KMS" in html or "encryption enabled" in html.lower()
finally:
_shutdown_app(app)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"csrf_token": csrf_token,
"action": "enable",
"algorithm": "aws:kms",
"kms_key_id": key_id,
},
follow_redirects=True,
)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "KMS" in html or "encryption enabled" in html.lower()
def test_disable_encryption(self, tmp_path):
"""Should be able to disable encryption."""
app = _make_encryption_app(tmp_path)
client = app.test_client()
try:
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/test-bucket?tab=properties")
csrf_token = get_csrf_token(response)
client.post(
"/ui/buckets/test-bucket/encryption",
data={
"csrf_token": csrf_token,
"action": "enable",
"algorithm": "AES256",
},
)
client.post(
"/ui/buckets/test-bucket/encryption",
data={
"action": "enable",
"algorithm": "AES256",
},
)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"action": "disable",
},
follow_redirects=True,
)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "disabled" in html.lower() or "Default encryption disabled" in html
finally:
_shutdown_app(app)
response = client.get("/ui/buckets/test-bucket?tab=properties")
csrf_token = get_csrf_token(response)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"csrf_token": csrf_token,
"action": "disable",
},
follow_redirects=True,
)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "disabled" in html.lower() or "Default encryption disabled" in html
def test_invalid_algorithm_rejected(self, tmp_path):
"""Invalid encryption algorithm should be rejected."""
app = _make_encryption_app(tmp_path)
client = app.test_client()
try:
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/test-bucket?tab=properties")
csrf_token = get_csrf_token(response)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"action": "enable",
"algorithm": "INVALID",
},
follow_redirects=True,
)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"csrf_token": csrf_token,
"action": "enable",
"algorithm": "INVALID",
},
follow_redirects=True,
)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "Invalid" in html or "danger" in html
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "Invalid" in html or "danger" in html
finally:
_shutdown_app(app)
def test_encryption_persists_in_config(self, tmp_path):
"""Encryption config should persist in bucket config."""
app = _make_encryption_app(tmp_path)
client = app.test_client()
try:
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/test-bucket?tab=properties")
csrf_token = get_csrf_token(response)
client.post(
"/ui/buckets/test-bucket/encryption",
data={
"action": "enable",
"algorithm": "AES256",
},
)
client.post(
"/ui/buckets/test-bucket/encryption",
data={
"csrf_token": csrf_token,
"action": "enable",
"algorithm": "AES256",
},
)
with app.app_context():
storage = app.extensions["object_storage"]
config = storage.get_bucket_encryption("test-bucket")
with app.app_context():
storage = app.extensions["object_storage"]
config = storage.get_bucket_encryption("test-bucket")
assert "Rules" in config
assert len(config["Rules"]) == 1
assert config["Rules"][0]["ApplyServerSideEncryptionByDefault"]["SSEAlgorithm"] == "AES256"
assert "Rules" in config
assert len(config["Rules"]) == 1
assert config["Rules"][0]["SSEAlgorithm"] == "AES256"
finally:
_shutdown_app(app)
class TestUIEncryptionWithoutPermission:
"""Test encryption UI when user lacks permissions."""
def test_readonly_user_cannot_change_encryption(self, tmp_path):
"""Read-only user should not be able to change encryption settings."""
app = _make_encryption_app(tmp_path)
client = app.test_client()
try:
client = app.test_client()
client.post("/ui/login", data={"access_key": "readonly", "secret_key": "secret"}, follow_redirects=True)
client.post("/ui/login", data={"access_key": "readonly", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/test-bucket?tab=properties")
csrf_token = get_csrf_token(response)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"action": "enable",
"algorithm": "AES256",
},
follow_redirects=True,
)
response = client.post(
"/ui/buckets/test-bucket/encryption",
data={
"csrf_token": csrf_token,
"action": "enable",
"algorithm": "AES256",
},
follow_redirects=True,
)
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "Access denied" in html or "permission" in html.lower() or "not authorized" in html.lower()
assert response.status_code == 200
html = response.data.decode("utf-8")
assert "Access denied" in html or "permission" in html.lower() or "not authorized" in html.lower()
finally:
_shutdown_app(app)

277
tests/test_ui_pagination.py
View File

@@ -1,15 +1,18 @@
"""Tests for UI pagination of bucket objects."""
import json
import threading
from io import BytesIO
from pathlib import Path
import pytest
from werkzeug.serving import make_server
from app import create_app
from app.s3_client import S3ProxyClient
def _make_app(tmp_path: Path):
"""Create an app for testing."""
"""Create an app for testing with a live API server."""
storage_root = tmp_path / "data"
iam_config = tmp_path / "iam.json"
bucket_policies = tmp_path / "bucket_policies.json"
@@ -33,157 +36,177 @@ def _make_app(tmp_path: Path):
"STORAGE_ROOT": storage_root,
"IAM_CONFIG": iam_config,
"BUCKET_POLICY_PATH": bucket_policies,
"API_BASE_URL": "http://127.0.0.1:0",
}
)
server = make_server("127.0.0.1", 0, flask_app)
host, port = server.server_address
api_url = f"http://{host}:{port}"
flask_app.config["API_BASE_URL"] = api_url
flask_app.extensions["s3_proxy"] = S3ProxyClient(api_base_url=api_url)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
flask_app._test_server = server
flask_app._test_thread = thread
return flask_app
def _shutdown_app(app):
if hasattr(app, "_test_server"):
app._test_server.shutdown()
app._test_thread.join(timeout=2)
class TestPaginatedObjectListing:
"""Test paginated object listing API."""
def test_objects_api_returns_paginated_results(self, tmp_path):
"""Objects API should return paginated results."""
app = _make_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
# Create 10 test objects
for i in range(10):
storage.put_object("test-bucket", f"file{i:02d}.txt", BytesIO(b"content"))
with app.test_client() as client:
# Login first
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
# Request first page of 3 objects
resp = client.get("/ui/buckets/test-bucket/objects?max_keys=3")
assert resp.status_code == 200
data = resp.get_json()
assert len(data["objects"]) == 3
assert data["is_truncated"] is True
assert data["next_continuation_token"] is not None
assert data["total_count"] == 10
try:
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
for i in range(10):
storage.put_object("test-bucket", f"file{i:02d}.txt", BytesIO(b"content"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
resp = client.get("/ui/buckets/test-bucket/objects?max_keys=3")
assert resp.status_code == 200
data = resp.get_json()
assert len(data["objects"]) == 3
assert data["is_truncated"] is True
assert data["next_continuation_token"] is not None
finally:
_shutdown_app(app)
def test_objects_api_pagination_continuation(self, tmp_path):
"""Objects API should support continuation tokens."""
app = _make_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
# Create 5 test objects
for i in range(5):
storage.put_object("test-bucket", f"file{i:02d}.txt", BytesIO(b"content"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
# Get first page
resp = client.get("/ui/buckets/test-bucket/objects?max_keys=2")
assert resp.status_code == 200
data = resp.get_json()
first_page_keys = [obj["key"] for obj in data["objects"]]
assert len(first_page_keys) == 2
assert data["is_truncated"] is True
# Get second page
token = data["next_continuation_token"]
resp = client.get(f"/ui/buckets/test-bucket/objects?max_keys=2&continuation_token={token}")
assert resp.status_code == 200
data = resp.get_json()
second_page_keys = [obj["key"] for obj in data["objects"]]
assert len(second_page_keys) == 2
# No overlap between pages
assert set(first_page_keys).isdisjoint(set(second_page_keys))
try:
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
for i in range(5):
storage.put_object("test-bucket", f"file{i:02d}.txt", BytesIO(b"content"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
resp = client.get("/ui/buckets/test-bucket/objects?max_keys=2")
assert resp.status_code == 200
data = resp.get_json()
first_page_keys = [obj["key"] for obj in data["objects"]]
assert len(first_page_keys) == 2
assert data["is_truncated"] is True
token = data["next_continuation_token"]
resp = client.get(f"/ui/buckets/test-bucket/objects?max_keys=2&continuation_token={token}")
assert resp.status_code == 200
data = resp.get_json()
second_page_keys = [obj["key"] for obj in data["objects"]]
assert len(second_page_keys) == 2
assert set(first_page_keys).isdisjoint(set(second_page_keys))
finally:
_shutdown_app(app)
def test_objects_api_prefix_filter(self, tmp_path):
"""Objects API should support prefix filtering."""
app = _make_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
# Create objects with different prefixes
storage.put_object("test-bucket", "logs/access.log", BytesIO(b"log"))
storage.put_object("test-bucket", "logs/error.log", BytesIO(b"log"))
storage.put_object("test-bucket", "data/file.txt", BytesIO(b"data"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
# Filter by prefix
resp = client.get("/ui/buckets/test-bucket/objects?prefix=logs/")
assert resp.status_code == 200
data = resp.get_json()
keys = [obj["key"] for obj in data["objects"]]
assert all(k.startswith("logs/") for k in keys)
assert len(keys) == 2
try:
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
storage.put_object("test-bucket", "logs/access.log", BytesIO(b"log"))
storage.put_object("test-bucket", "logs/error.log", BytesIO(b"log"))
storage.put_object("test-bucket", "data/file.txt", BytesIO(b"data"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
resp = client.get("/ui/buckets/test-bucket/objects?prefix=logs/")
assert resp.status_code == 200
data = resp.get_json()
keys = [obj["key"] for obj in data["objects"]]
assert all(k.startswith("logs/") for k in keys)
assert len(keys) == 2
finally:
_shutdown_app(app)
def test_objects_api_requires_authentication(self, tmp_path):
"""Objects API should require login."""
app = _make_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
with app.test_client() as client:
# Don't login
resp = client.get("/ui/buckets/test-bucket/objects")
# Should redirect to login
assert resp.status_code == 302
assert "/ui/login" in resp.headers.get("Location", "")
try:
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
with app.test_client() as client:
resp = client.get("/ui/buckets/test-bucket/objects")
assert resp.status_code == 302
assert "/ui/login" in resp.headers.get("Location", "")
finally:
_shutdown_app(app)
def test_objects_api_returns_object_metadata(self, tmp_path):
"""Objects API should return complete object metadata."""
app = _make_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
storage.put_object("test-bucket", "test.txt", BytesIO(b"test content"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
resp = client.get("/ui/buckets/test-bucket/objects")
assert resp.status_code == 200
data = resp.get_json()
assert len(data["objects"]) == 1
obj = data["objects"][0]
try:
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
storage.put_object("test-bucket", "test.txt", BytesIO(b"test content"))
# Check all expected fields
assert obj["key"] == "test.txt"
assert obj["size"] == 12 # len("test content")
assert "last_modified" in obj
assert "last_modified_display" in obj
assert "etag" in obj
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
resp = client.get("/ui/buckets/test-bucket/objects")
assert resp.status_code == 200
data = resp.get_json()
assert len(data["objects"]) == 1
obj = data["objects"][0]
assert obj["key"] == "test.txt"
assert obj["size"] == 12
assert "last_modified" in obj
assert "last_modified_display" in obj
assert "etag" in obj
assert "url_templates" in data
templates = data["url_templates"]
assert "preview" in templates
assert "download" in templates
assert "delete" in templates
assert "KEY_PLACEHOLDER" in templates["preview"]
finally:
_shutdown_app(app)
# URLs are now returned as templates (not per-object) for performance
assert "url_templates" in data
templates = data["url_templates"]
assert "preview" in templates
assert "download" in templates
assert "delete" in templates
assert "KEY_PLACEHOLDER" in templates["preview"]
def test_bucket_detail_page_loads_without_objects(self, tmp_path):
"""Bucket detail page should load even with many objects."""
app = _make_app(tmp_path)
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
# Create many objects
for i in range(100):
storage.put_object("test-bucket", f"file{i:03d}.txt", BytesIO(b"x"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
# The page should load quickly (objects loaded via JS)
resp = client.get("/ui/buckets/test-bucket")
assert resp.status_code == 200
html = resp.data.decode("utf-8")
# Should have the JavaScript loading infrastructure (external JS file)
assert "bucket-detail-main.js" in html
try:
storage = app.extensions["object_storage"]
storage.create_bucket("test-bucket")
for i in range(100):
storage.put_object("test-bucket", f"file{i:03d}.txt", BytesIO(b"x"))
with app.test_client() as client:
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
resp = client.get("/ui/buckets/test-bucket")
assert resp.status_code == 200
html = resp.data.decode("utf-8")
assert "bucket-detail-main.js" in html
finally:
_shutdown_app(app)

97
tests/test_ui_policy.py
View File

@@ -1,10 +1,13 @@
import io
import json
import threading
from pathlib import Path
import pytest
from werkzeug.serving import make_server
from app import create_app
from app.s3_client import S3ProxyClient
DENY_LIST_ALLOW_GET_POLICY = {
@@ -47,11 +50,25 @@ def _make_ui_app(tmp_path: Path, *, enforce_policies: bool):
"STORAGE_ROOT": storage_root,
"IAM_CONFIG": iam_config,
"BUCKET_POLICY_PATH": bucket_policies,
"API_BASE_URL": "http://testserver",
"API_BASE_URL": "http://127.0.0.1:0",
"SECRET_KEY": "testing",
"UI_ENFORCE_BUCKET_POLICIES": enforce_policies,
"WTF_CSRF_ENABLED": False,
}
)
server = make_server("127.0.0.1", 0, app)
host, port = server.server_address
api_url = f"http://{host}:{port}"
app.config["API_BASE_URL"] = api_url
app.extensions["s3_proxy"] = S3ProxyClient(api_base_url=api_url)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
app._test_server = server
app._test_thread = thread
storage = app.extensions["object_storage"]
storage.create_bucket("testbucket")
storage.put_object("testbucket", "vid.mp4", io.BytesIO(b"video"))
@@ -60,22 +77,28 @@ def _make_ui_app(tmp_path: Path, *, enforce_policies: bool):
return app
def _shutdown_app(app):
if hasattr(app, "_test_server"):
app._test_server.shutdown()
app._test_thread.join(timeout=2)
@pytest.mark.parametrize("enforce", [True, False])
def test_ui_bucket_policy_enforcement_toggle(tmp_path: Path, enforce: bool):
app = _make_ui_app(tmp_path, enforce_policies=enforce)
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/testbucket", follow_redirects=True)
if enforce:
assert b"Access denied by bucket policy" in response.data
else:
assert response.status_code == 200
assert b"Access denied by bucket policy" not in response.data
# Objects are now loaded via async API - check the objects endpoint
objects_response = client.get("/ui/buckets/testbucket/objects")
assert objects_response.status_code == 200
data = objects_response.get_json()
assert any(obj["key"] == "vid.mp4" for obj in data["objects"])
try:
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/testbucket", follow_redirects=True)
if enforce:
assert b"Access denied by bucket policy" in response.data
else:
assert response.status_code == 200
assert b"Access denied by bucket policy" not in response.data
objects_response = client.get("/ui/buckets/testbucket/objects")
assert objects_response.status_code == 403
finally:
_shutdown_app(app)
def test_ui_bucket_policy_disabled_by_default(tmp_path: Path):
@@ -99,23 +122,37 @@ def test_ui_bucket_policy_disabled_by_default(tmp_path: Path):
"STORAGE_ROOT": storage_root,
"IAM_CONFIG": iam_config,
"BUCKET_POLICY_PATH": bucket_policies,
"API_BASE_URL": "http://testserver",
"API_BASE_URL": "http://127.0.0.1:0",
"SECRET_KEY": "testing",
"WTF_CSRF_ENABLED": False,
}
)
storage = app.extensions["object_storage"]
storage.create_bucket("testbucket")
storage.put_object("testbucket", "vid.mp4", io.BytesIO(b"video"))
policy_store = app.extensions["bucket_policies"]
policy_store.set_policy("testbucket", DENY_LIST_ALLOW_GET_POLICY)
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/testbucket", follow_redirects=True)
assert response.status_code == 200
assert b"Access denied by bucket policy" not in response.data
# Objects are now loaded via async API - check the objects endpoint
objects_response = client.get("/ui/buckets/testbucket/objects")
assert objects_response.status_code == 200
data = objects_response.get_json()
assert any(obj["key"] == "vid.mp4" for obj in data["objects"])
server = make_server("127.0.0.1", 0, app)
host, port = server.server_address
api_url = f"http://{host}:{port}"
app.config["API_BASE_URL"] = api_url
app.extensions["s3_proxy"] = S3ProxyClient(api_base_url=api_url)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
app._test_server = server
app._test_thread = thread
try:
storage = app.extensions["object_storage"]
storage.create_bucket("testbucket")
storage.put_object("testbucket", "vid.mp4", io.BytesIO(b"video"))
policy_store = app.extensions["bucket_policies"]
policy_store.set_policy("testbucket", DENY_LIST_ALLOW_GET_POLICY)
client = app.test_client()
client.post("/ui/login", data={"access_key": "test", "secret_key": "secret"}, follow_redirects=True)
response = client.get("/ui/buckets/testbucket", follow_redirects=True)
assert response.status_code == 200
assert b"Access denied by bucket policy" not in response.data
objects_response = client.get("/ui/buckets/testbucket/objects")
assert objects_response.status_code == 403
finally:
_shutdown_app(app)