tests/test_docvault.py

"""HF-backed API tests for DocVault."""

from __future__ import annotations

import io
import os
import unittest
from unittest.mock import patch

os.environ.setdefault("STORAGE_MODE", "HF")
os.environ.setdefault("HF_TOKEN", "hf_test_token")
os.environ.setdefault("HF_REPO_ID", "test-user/docvault-storage")
os.environ.setdefault("HF_REPO_TYPE", "dataset")
os.environ.setdefault("SECRET_KEY", "test-secret")

from server.app import create_app
from server.storage import factory


class FakeCommit:
    def __init__(self, commit_id: str, title: str):
        self.commit_id = commit_id
        self.title = title
        self.message = title
        self.created_at = __import__("datetime").datetime.now(__import__("datetime").timezone.utc)
        self.authors = ["test"]


class FakeHfApi:
    def __init__(self, token=None):
        self.token = token
        self.files = {}
        self.commits = []
        self.repo_created = False

    def repo_info(self, repo_id, repo_type):
        if not self.repo_created:
            raise RuntimeError("repo missing")
        return {"repo_id": repo_id, "repo_type": repo_type}

    def create_repo(self, repo_id, repo_type, private=True, exist_ok=True):
        self.repo_created = True
        return {"repo_id": repo_id, "repo_type": repo_type}

    def list_repo_files(self, repo_id, repo_type):
        return sorted(self.files.keys())

    def create_commit(self, repo_id, repo_type, operations, commit_message):
        for operation in operations:
            operation_type = operation.__class__.__name__
            if operation_type == "CommitOperationAdd":
                self.files[operation.path_in_repo] = bytes(operation.path_or_fileobj)
            elif operation_type == "CommitOperationDelete":
                self.files.pop(operation.path_in_repo, None)
            elif operation_type == "CommitOperationCopy":
                self.files[operation.path_in_repo] = self.files[operation.src_path_in_repo]
        self.commits.append(FakeCommit(f"commit-{len(self.commits) + 1}", commit_message))
        return {"commit_message": commit_message}

    def delete_file(self, path_in_repo, repo_id, repo_type, commit_message):
        self.files.pop(path_in_repo, None)
        self.commits.append(FakeCommit(f"commit-{len(self.commits) + 1}", commit_message))

    def list_repo_commits(self, repo_id, repo_type):
        return list(reversed(self.commits))


class DocVaultApiTestCase(unittest.TestCase):
    def setUp(self):
        self.api = FakeHfApi()
        self.hf_patcher = patch("server.storage.hf.HfApi", return_value=self.api)
        self.hf_patcher.start()
        factory._storage_instance = None

        self.app = create_app()
        self.client = self.app.test_client()
        self.headers = {"X-User-ID": "test_user"}

    def tearDown(self):
        factory._storage_instance = None
        self.hf_patcher.stop()

    def test_health_check(self):
        response = self.client.get("/api/health")
        self.assertEqual(response.status_code, 200)
        payload = response.get_json()
        self.assertEqual(payload, {
            "storage": "HF",
            "repo": "test-user/docvault-storage",
            "status": "ok",
        })

    def test_create_folder(self):
        response = self.client.post(
            "/api/create-folder",
            json={"folder_path": "Docs"},
            headers=self.headers,
        )
        self.assertEqual(response.status_code, 201)
        payload = response.get_json()
        self.assertTrue(payload["success"])
        self.assertEqual(payload["folder"]["path"], "Docs")
        self.assertIn("test_user/Docs/.gitkeep", self.api.files)

    def test_upload_file(self):
        response = self.client.post(
            "/api/upload",
            data={"folder_path": "Docs", "file": (io.BytesIO(b"hello"), "file.txt")},
            headers=self.headers,
        )
        self.assertEqual(response.status_code, 201)
        payload = response.get_json()
        self.assertTrue(payload["success"])
        self.assertEqual(payload["file"]["path"], "Docs/file.txt")
        self.assertIn("test_user/Docs/file.txt", self.api.files)

    def test_duplicate_upload_is_renamed(self):
        first = self.client.post(
            "/api/upload",
            data={"folder_path": "Docs", "file": (io.BytesIO(b"a"), "file.txt")},
            headers=self.headers,
        )
        second = self.client.post(
            "/api/upload",
            data={"folder_path": "Docs", "file": (io.BytesIO(b"b"), "file.txt")},
            headers=self.headers,
        )
        self.assertEqual(first.status_code, 201)
        self.assertEqual(second.status_code, 201)
        self.assertEqual(second.get_json()["file"]["name"], "file_1.txt")
        self.assertIn("test_user/Docs/file_1.txt", self.api.files)

    def test_rename_folder(self):
        self.client.post("/api/create-folder", json={"folder_path": "Old"}, headers=self.headers)
        self.client.post(
            "/api/upload",
            data={"folder_path": "Old", "file": (io.BytesIO(b"content"), "note.txt")},
            headers=self.headers,
        )

        response = self.client.post(
            "/api/rename",
            json={"item_path": "Old", "new_name": "New"},
            headers=self.headers,
        )
        self.assertEqual(response.status_code, 200)
        payload = response.get_json()
        self.assertTrue(payload["success"])
        self.assertIn("test_user/New/note.txt", self.api.files)
        self.assertNotIn("test_user/Old/note.txt", self.api.files)

    def test_delete_file(self):
        self.client.post(
            "/api/upload",
            data={"folder_path": "", "file": (io.BytesIO(b"content"), "note.txt")},
            headers=self.headers,
        )

        response = self.client.post(
            "/api/delete",
            json={"path": "note.txt", "type": "file"},
            headers=self.headers,
        )
        self.assertEqual(response.status_code, 200)
        self.assertTrue(response.get_json()["success"])
        self.assertNotIn("test_user/note.txt", self.api.files)

    def test_delete_folder(self):
        self.client.post("/api/create-folder", json={"folder_path": "Trash"}, headers=self.headers)
        self.client.post(
            "/api/upload",
            data={"folder_path": "Trash", "file": (io.BytesIO(b"content"), "note.txt")},
            headers=self.headers,
        )

        response = self.client.post(
            "/api/delete",
            json={"path": "Trash", "type": "folder"},
            headers=self.headers,
        )
        self.assertEqual(response.status_code, 200)
        self.assertTrue(response.get_json()["success"])
        self.assertFalse(any(path.startswith("test_user/Trash/") for path in self.api.files))


if __name__ == "__main__":
    unittest.main()