nas-tools/dashboard/backend/tests/integration/test_files.py

"""
Integration tests for file operations.
"""
import pytest
import os


class TestFileBrowse:
    """Test file browsing."""

    def test_browse_root(self, test_app, admin_headers, temp_volume_root):
        """Test browsing root directory."""
        response = test_app.get(
            "/api/files/browse",
            headers=admin_headers,
            params={"path": "/"}
        )

        assert response.status_code == 200
        data = response.json()
        assert "entries" in data or isinstance(data, list)

    def test_browse_without_auth(self, test_app):
        """Test browsing without authentication."""
        response = test_app.get("/api/files/browse", params={"path": "/"})

        assert response.status_code == 401

    def test_browse_blocked_directory(self, test_app, admin_headers):
        """Test browsing blocked directory."""
        response = test_app.get(
            "/api/files/browse",
            headers=admin_headers,
            params={"path": "/@appstore"}
        )

        # Should be forbidden
        assert response.status_code == 403

    def test_browse_path_traversal_attempt(self, test_app, admin_headers):
        """Test path traversal protection."""
        response = test_app.get(
            "/api/files/browse",
            headers=admin_headers,
            params={"path": "/../../../etc"}
        )

        # Should be forbidden
        assert response.status_code == 403


class TestFileUpload:
    """Test file upload."""

    def test_upload_file_success(self, test_app, admin_headers, temp_volume_root):
        """Test uploading a file."""
        test_content = b"test file content"
        files = {"file": ("test.txt", test_content, "text/plain")}

        response = test_app.post(
            "/api/files/upload",
            headers=admin_headers,
            files=files,
            params={"path": "/"}
        )

        assert response.status_code in [200, 201]

    def test_upload_without_auth(self, test_app):
        """Test upload without authentication."""
        files = {"file": ("test.txt", b"content", "text/plain")}

        response = test_app.post(
            "/api/files/upload",
            files=files,
            params={"path": "/"}
        )

        assert response.status_code == 401

    def test_upload_as_member_forbidden(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
        """Test that non-admin cannot upload."""
        from auth_service import create_access_token
        from datetime import timedelta

        member_token = create_access_token(
            data={"sub": "memberuser", "role": "member"},
            expires_delta=timedelta(minutes=30)
        )
        headers = {"Authorization": f"Bearer {member_token}"}
        files = {"file": ("test.txt", b"content", "text/plain")}

        response = test_app.post(
            "/api/files/upload",
            headers=headers,
            files=files,
            params={"path": "/"}
        )

        # Upload should be admin-only
        assert response.status_code == 403


class TestFileDelete:
    """Test file deletion."""

    def test_delete_file_success(self, test_app, admin_headers, temp_volume_root):
        """Test deleting a file."""
        # Create a test file first
        test_file = os.path.join(temp_volume_root, "test_delete.txt")
        with open(test_file, "w") as f:
            f.write("test content")

        response = test_app.delete(
            "/api/files/delete",
            headers=admin_headers,
            params={"path": "/test_delete.txt"}
        )

        assert response.status_code in [200, 204]

    def test_delete_without_auth(self, test_app):
        """Test delete without authentication."""
        response = test_app.delete(
            "/api/files/delete",
            params={"path": "/test.txt"}
        )

        assert response.status_code == 401

    def test_delete_as_member_forbidden(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
        """Test that non-admin cannot delete."""
        from auth_service import create_access_token
        from datetime import timedelta

        member_token = create_access_token(
            data={"sub": "memberuser", "role": "member"},
            expires_delta=timedelta(minutes=30)
        )
        headers = {"Authorization": f"Bearer {member_token}"}

        response = test_app.delete(
            "/api/files/delete",
            headers=headers,
            params={"path": "/test.txt"}
        )

        # Delete should be admin-only
        assert response.status_code == 403


class TestFileDownload:
    """Test file download."""

    def test_download_file_success(self, test_app, admin_headers, temp_volume_root):
        """Test downloading a file."""
        # Create a test file
        test_file = os.path.join(temp_volume_root, "test_download.txt")
        test_content = "test download content"
        with open(test_file, "w") as f:
            f.write(test_content)

        response = test_app.get(
            "/api/files/download",
            headers=admin_headers,
            params={"path": "/test_download.txt"}
        )

        assert response.status_code == 200
        assert test_content in response.text or response.content == test_content.encode()

    def test_download_without_auth(self, test_app):
        """Test download without authentication."""
        response = test_app.get(
            "/api/files/download",
            params={"path": "/test.txt"}
        )

        assert response.status_code == 401

    def test_download_nonexistent_file(self, test_app, admin_headers):
        """Test downloading nonexistent file."""
        response = test_app.get(
            "/api/files/download",
            headers=admin_headers,
            params={"path": "/nonexistent.txt"}
        )

        assert response.status_code == 404