diff --git a/fastapi_startkit/tests/storage/test_local_driver.py b/fastapi_startkit/tests/storage/test_local_driver.py new file mode 100644 index 00000000..84a2aa06 --- /dev/null +++ b/fastapi_startkit/tests/storage/test_local_driver.py @@ -0,0 +1,217 @@ +"""Tests for the LocalDriver, FakeDriver, and FileStream (task #14).""" + +import os +from unittest.mock import MagicMock + +import pytest + +from fastapi_startkit.storage.drivers.fake import FakeDriver +from fastapi_startkit.storage.drivers.local import LocalDriver +from fastapi_startkit.storage.filestream import FileStream + + +# --------------------------------------------------------------------------- +# LocalDriver — put / get / delete / exists / list / move / copy +# --------------------------------------------------------------------------- + + +class TestLocalDriverPutGet: + @pytest.fixture + def driver(self, tmp_path): + app = MagicMock() + app.base_path = str(tmp_path) + d = LocalDriver(app) + d.set_options({"root": str(tmp_path / "storage")}) + return d + + def test_put_creates_file(self, driver, tmp_path): + driver.put("hello.txt", "hello world") + path = tmp_path / "storage" / "hello.txt" + assert path.exists() + assert path.read_text() == "hello world" + + def test_put_binary_content(self, driver, tmp_path): + driver.put("data.bin", b"\x00\x01\x02") + path = tmp_path / "storage" / "data.bin" + assert path.read_bytes() == b"\x00\x01\x02" + + def test_get_returns_content(self, driver, tmp_path): + (tmp_path / "storage").mkdir(parents=True, exist_ok=True) + (tmp_path / "storage" / "read.txt").write_text("read me") + assert driver.get("read.txt") == "read me" + + def test_get_returns_none_for_missing_file(self, driver): + result = driver.get("nonexistent.txt") + assert result is None + + def test_exists_true_for_present_file(self, driver, tmp_path): + (tmp_path / "storage").mkdir(parents=True, exist_ok=True) + (tmp_path / "storage" / "present.txt").write_text("here") + assert driver.exists("present.txt") is True + + def test_exists_false_for_missing_file(self, driver): + assert driver.exists("missing.txt") is False + + def test_missing_is_inverse_of_exists(self, driver, tmp_path): + (tmp_path / "storage").mkdir(parents=True, exist_ok=True) + (tmp_path / "storage" / "file.txt").write_text("x") + assert driver.missing("file.txt") is False + assert driver.missing("no_file.txt") is True + + def test_delete_removes_file(self, driver, tmp_path): + (tmp_path / "storage").mkdir(parents=True, exist_ok=True) + target = tmp_path / "storage" / "delete_me.txt" + target.write_text("bye") + driver.delete("delete_me.txt") + assert not target.exists() + + def test_append_adds_to_existing_content(self, driver, tmp_path): + (tmp_path / "storage").mkdir(parents=True, exist_ok=True) + (tmp_path / "storage" / "log.txt").write_text("line1\n") + driver.append("log.txt", "line2\n") + assert (tmp_path / "storage" / "log.txt").read_text() == "line1\nline2\n" + + def test_prepend_adds_before_existing_content(self, driver, tmp_path): + (tmp_path / "storage").mkdir(parents=True, exist_ok=True) + (tmp_path / "storage" / "pre.txt").write_text("world") + driver.prepend("pre.txt", "hello ") + assert (tmp_path / "storage" / "pre.txt").read_text() == "hello world" + + def test_move_relocates_file(self, driver, tmp_path): + (tmp_path / "storage").mkdir(parents=True, exist_ok=True) + src = tmp_path / "storage" / "src.txt" + src.write_text("move me") + driver.move("src.txt", "dst.txt") + assert not src.exists() + assert (tmp_path / "storage" / "dst.txt").read_text() == "move me" + + def test_copy_duplicates_file(self, driver, tmp_path): + storage = tmp_path / "storage" + storage.mkdir(parents=True, exist_ok=True) + src = storage / "orig.txt" + src.write_text("copy me") + driver.copy(str(src), str(storage / "copy.txt")) + assert src.exists() + assert (storage / "copy.txt").read_text() == "copy me" + + def test_get_files_lists_files_in_directory(self, driver, tmp_path): + storage = tmp_path / "storage" + storage.mkdir(parents=True, exist_ok=True) + (storage / "a.txt").write_text("a") + (storage / "b.txt").write_text("b") + files = driver.get_files("") + names = [f.name() for f in files] + assert set(names) == {"a.txt", "b.txt"} + + +# --------------------------------------------------------------------------- +# FakeDriver — in-memory / temp directory behaviour +# --------------------------------------------------------------------------- + + +class TestFakeDriver: + @pytest.fixture + def fake(self): + app = MagicMock() + app.base_path = "/fake" + return FakeDriver(app, disk_name="test") + + def test_fake_driver_uses_temp_directory(self, fake): + assert os.path.exists(fake._root) + assert "storage_fake_test_" in fake._root + + def test_fake_driver_put_and_assert_exists(self, fake): + fake.put("hello.txt", "hi") + fake.assert_exists("hello.txt") + + def test_fake_driver_assert_exists_with_content(self, fake): + fake.put("content.txt", "expected") + fake.assert_exists("content.txt", content="expected") + + def test_fake_driver_assert_missing(self, fake): + fake.assert_missing("gone.txt") + + def test_fake_driver_assert_missing_after_delete(self, fake): + fake.put("temp.txt", "x") + fake.delete("temp.txt") + fake.assert_missing("temp.txt") + + def test_fake_driver_assert_count(self, fake): + fake.put("one.txt", "1") + fake.put("two.txt", "2") + fake.assert_count(2) + + def test_fake_driver_assert_directory_empty(self, fake): + fake.assert_directory_empty() + + def test_fake_driver_set_options_is_noop(self, fake): + original_root = fake._root + fake.set_options({"root": "/different/path"}) + assert fake.options["root"] == original_root + + def test_fake_driver_cleanup_removes_directory(self, fake): + root = fake._root + assert os.path.exists(root) + fake.cleanup() + assert not os.path.exists(root) + + def test_fake_driver_context_manager(self): + app = MagicMock() + app.base_path = "/fake" + with FakeDriver(app, disk_name="ctx") as fake: + fake.put("file.txt", "data") + fake.assert_exists("file.txt") + root = fake._root + assert not os.path.exists(root) + + def test_fake_driver_isolated_between_instances(self): + app = MagicMock() + app.base_path = "/fake" + fake1 = FakeDriver(app, disk_name="iso") + fake2 = FakeDriver(app, disk_name="iso") + fake1.put("shared.txt", "in fake1") + fake2.assert_missing("shared.txt") + fake1.cleanup() + fake2.cleanup() + + +# --------------------------------------------------------------------------- +# FileStream +# --------------------------------------------------------------------------- + + +class TestFileStream: + def test_filestream_path(self, tmp_path): + f = tmp_path / "sample.txt" + f.write_text("hello") + with open(f) as fh: + stream = FileStream(fh) + assert stream.path() == str(f) + + def test_filestream_extension(self, tmp_path): + f = tmp_path / "image.png" + f.write_text("data") + with open(f) as fh: + stream = FileStream(fh) + assert stream.extension() == ".png" + + def test_filestream_name_from_path(self, tmp_path): + f = tmp_path / "document.pdf" + f.write_text("pdf") + with open(f) as fh: + stream = FileStream(fh) + assert stream.name() == "document.pdf" + + def test_filestream_name_from_explicit_name(self, tmp_path): + f = tmp_path / "actual.txt" + f.write_text("text") + with open(f) as fh: + stream = FileStream(fh, name="alias.txt") + assert stream.name() == "alias.txt" + + def test_filestream_extension_from_explicit_name(self, tmp_path): + f = tmp_path / "file.txt" + f.write_text("text") + with open(f) as fh: + stream = FileStream(fh, name="renamed.csv") + assert stream.extension() == ".csv" diff --git a/fastapi_startkit/tests/storage/test_s3_driver.py b/fastapi_startkit/tests/storage/test_s3_driver.py new file mode 100644 index 00000000..90c10ed4 --- /dev/null +++ b/fastapi_startkit/tests/storage/test_s3_driver.py @@ -0,0 +1,280 @@ +"""Tests for the S3Driver with boto3/botocore mocked (task #14).""" + +from unittest.mock import MagicMock, patch + +import pytest + +from fastapi_startkit.storage.drivers.s3 import S3Driver + + +@pytest.fixture +def driver(): + app = MagicMock() + d = S3Driver(app) + d.set_options( + { + "bucket": "test-bucket", + "key": "AWS_KEY", + "secret": "AWS_SECRET", + "region": "us-east-1", + } + ) + return d + + +@pytest.fixture +def mock_resource(driver): + """Patch get_resource so S3 calls never hit AWS.""" + mock_res = MagicMock() + with patch.object(driver, "get_resource", return_value=mock_res): + yield mock_res + + +@pytest.fixture +def mock_client(driver): + mock_cl = MagicMock() + with patch.object(driver, "get_client", return_value=mock_cl): + yield mock_cl + + +# --------------------------------------------------------------------------- +# put +# --------------------------------------------------------------------------- + + +class TestS3DriverPut: + def test_put_calls_put_object_with_correct_key(self, driver, mock_resource): + driver.put("uploads/file.txt", "content") + mock_resource.Bucket.assert_called_with("test-bucket") + mock_resource.Bucket().put_object.assert_called_with(Key="uploads/file.txt", Body="content") + + def test_put_returns_content(self, driver, mock_resource): + result = driver.put("file.txt", "data") + assert result == "data" + + +# --------------------------------------------------------------------------- +# get +# --------------------------------------------------------------------------- + + +class TestS3DriverGet: + def test_get_returns_decoded_content(self, driver, mock_resource): + body = MagicMock() + body.read.return_value = b"file content" + mock_resource.Bucket().Object().get.return_value = {"Body": body} + + result = driver.get("file.txt") + assert result == "file content" + + def test_get_returns_none_on_client_error(self, driver, mock_resource): + class FakeClientError(Exception): + pass + + mock_resource.Bucket().Object().get.side_effect = FakeClientError("not found") + + with patch.object(driver, "missing_file_exceptions", return_value=(FakeClientError,)): + result = driver.get("missing.txt") + assert result is None + + +# --------------------------------------------------------------------------- +# exists +# --------------------------------------------------------------------------- + + +class TestS3DriverExists: + def test_exists_returns_true_when_object_loadable(self, driver, mock_resource): + mock_resource.Bucket().Object().load.return_value = None + with patch.object(driver, "missing_file_exceptions", return_value=(Exception,)): + result = driver.exists("file.txt") + assert result is True + + def test_exists_returns_false_on_client_error(self, driver, mock_resource): + class FakeClientError(Exception): + pass + + mock_resource.Bucket().Object().load.side_effect = FakeClientError("not found") + with patch.object(driver, "missing_file_exceptions", return_value=(FakeClientError,)): + result = driver.exists("missing.txt") + assert result is False + + def test_missing_is_inverse_of_exists(self, driver, mock_resource): + mock_resource.Bucket().Object().load.return_value = None + with patch.object(driver, "missing_file_exceptions", return_value=(Exception,)): + assert driver.missing("file.txt") is False + + +# --------------------------------------------------------------------------- +# delete +# --------------------------------------------------------------------------- + + +class TestS3DriverDelete: + def test_delete_calls_object_delete(self, driver, mock_resource): + driver.delete("remove.txt") + mock_resource.Object.assert_called_with("test-bucket", "remove.txt") + mock_resource.Object().delete.assert_called_once() + + +# --------------------------------------------------------------------------- +# copy / move +# --------------------------------------------------------------------------- + + +class TestS3DriverCopyMove: + def test_copy_calls_meta_client_copy(self, driver, mock_resource): + driver.copy("src.txt", "dst.txt") + expected_source = {"Bucket": "test-bucket", "Key": "src.txt"} + mock_resource.meta.client.copy.assert_called_once_with(expected_source, "test-bucket", "dst.txt") + + def test_move_copies_then_deletes_source(self, driver, mock_resource): + with patch.object(driver, "copy") as mock_copy, patch.object(driver, "delete") as mock_delete: + driver.move("src.txt", "dst.txt") + mock_copy.assert_called_once_with("src.txt", "dst.txt") + mock_delete.assert_called_once_with("src.txt") + + +# --------------------------------------------------------------------------- +# download (presigned URL) +# --------------------------------------------------------------------------- + + +class TestS3DriverDownload: + def test_download_returns_redirect_response(self, driver, mock_client): + mock_client.generate_presigned_url.return_value = "https://s3.amazonaws.com/signed-url" + + from fastapi.responses import RedirectResponse + + response = driver.download("file.txt") + assert isinstance(response, RedirectResponse) + + def test_download_generates_presigned_url_with_correct_params(self, driver, mock_client): + mock_client.generate_presigned_url.return_value = "https://presigned" + + driver.download("folder/file.txt") + mock_client.generate_presigned_url.assert_called_once_with( + "get_object", + Params={"Bucket": "test-bucket", "Key": "folder/file.txt"}, + ExpiresIn=3600, + ) + + +# --------------------------------------------------------------------------- +# url helper +# --------------------------------------------------------------------------- + + +class TestS3DriverUrl: + def test_url_returns_correct_format(self, driver): + driver.set_options({**driver.options, "url": "https://cdn.example.com"}) + assert driver.url("images/photo.jpg") == "https://cdn.example.com/images/photo.jpg" + + +# --------------------------------------------------------------------------- +# get_bucket +# --------------------------------------------------------------------------- + + +class TestS3DriverGetBucket: + def test_get_bucket_returns_configured_bucket(self, driver): + assert driver.get_bucket() == "test-bucket" + + def test_get_bucket_returns_none_when_not_configured(self): + app = MagicMock() + d = S3Driver(app) + d.set_options({}) + assert d.get_bucket() is None + + +# --------------------------------------------------------------------------- +# Edge cases: unexpected exceptions must propagate (not be swallowed) +# --------------------------------------------------------------------------- + + +class TestS3DriverExceptionPropagation: + def test_get_raises_unexpected_exception(self, driver, mock_resource): + class DatabaseError(Exception): + pass + + mock_resource.Bucket().Object().get.side_effect = DatabaseError("unexpected") + + with patch.object(driver, "missing_file_exceptions", return_value=(FileNotFoundError,)): + with pytest.raises(DatabaseError): + driver.get("file.txt") + + def test_exists_raises_unexpected_exception(self, driver, mock_resource): + class AWSDown(Exception): + pass + + mock_resource.Bucket().Object().load.side_effect = AWSDown("service down") + + with patch.object(driver, "missing_file_exceptions", return_value=(FileNotFoundError,)): + with pytest.raises(AWSDown): + driver.exists("file.txt") + + +# --------------------------------------------------------------------------- +# URL normalization +# --------------------------------------------------------------------------- + + +class TestS3DriverUrlNormalization: + def test_url_with_trailing_slash(self, driver): + driver.set_options({**driver.options, "url": "https://cdn.example.com/"}) + assert driver.url("file.txt") == "https://cdn.example.com//file.txt" + + def test_url_without_trailing_slash(self, driver): + driver.set_options({**driver.options, "url": "https://cdn.example.com"}) + assert driver.url("file.txt") == "https://cdn.example.com/file.txt" + + +# --------------------------------------------------------------------------- +# Empty and binary content +# --------------------------------------------------------------------------- + + +class TestS3DriverContentTypes: + def test_put_empty_content(self, driver, mock_resource): + driver.put("empty.txt", "") + mock_resource.Bucket().put_object.assert_called_with(Key="empty.txt", Body="") + + def test_put_binary_content(self, driver, mock_resource): + content = b"\x00\x01\x02" + driver.put("binary.bin", content) + mock_resource.Bucket().put_object.assert_called_with(Key="binary.bin", Body=content) + + +# --------------------------------------------------------------------------- +# Move atomicity: copy failure must not delete source +# --------------------------------------------------------------------------- + + +class TestS3DriverMoveAtomicity: + def test_move_does_not_delete_source_when_copy_fails(self, driver): + with ( + patch.object(driver, "copy", side_effect=Exception("copy failed")), + patch.object(driver, "delete") as mock_delete, + ): + with pytest.raises(Exception, match="copy failed"): + driver.move("src.txt", "dst.txt") + mock_delete.assert_not_called() + + +# --------------------------------------------------------------------------- +# Connection caching +# --------------------------------------------------------------------------- + + +class TestS3DriverConnectionCaching: + def test_get_connection_returns_same_session(self): + mock_boto3 = MagicMock() + app = MagicMock() + d = S3Driver(app) + d.set_options({"key": "k", "secret": "s", "region": "us-east-1"}) + + with patch.dict("sys.modules", {"boto3": mock_boto3}): + c1 = d.get_connection() + c2 = d.get_connection() + assert c1 is c2 + mock_boto3.Session.assert_called_once()