diff --git a/.github/workflows/lint_test.yaml b/.github/workflows/lint_test.yaml new file mode 100644 index 0000000..f8c78e5 --- /dev/null +++ b/.github/workflows/lint_test.yaml @@ -0,0 +1,35 @@ +name: Lint archivist + +on: + push: + branches: + - main + pull_request: + branches: + - main + +jobs: + lint: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Install dependencies + run: uv sync --python 3.11 --group dev + + - name: Run lint + run: | + uv run ruff check + + - name: Run lint + run: | + uv run ruff format --check --diff + + - name: Run isort + run: | + uv run isort . --check --diff diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml new file mode 100644 index 0000000..2e0e5f7 --- /dev/null +++ b/.github/workflows/release.yaml @@ -0,0 +1,24 @@ +name: Release archivist + +on: + release: + types: [published] + +jobs: + deploy: + runs-on: ubuntu-latest + environment: release + permissions: + id-token: write + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Build package + run: uv build --python 3.11 + + - name: Publish package distributions to PyPI + uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml new file mode 100644 index 0000000..eed6642 --- /dev/null +++ b/.github/workflows/test.yaml @@ -0,0 +1,36 @@ +name: Test archivist + +on: + push: + branches: + - main + pull_request: + branches: + - main + +jobs: + test: + runs-on: ubuntu-latest + + strategy: + matrix: + python-version: ["3.11", "3.12", "3.13", "3.14"] + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Install dependencies + run: uv sync --python ${{ matrix.python-version }} --group dev + + - name: Run tests + run: | + uv run pytest + + - name: Upload coverage report + if: matrix.python-version == '3.11' + run: uv run coveralls + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/README.md b/README.md index ba4c4f7..502f8b2 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,5 @@ # Archivist +[![Coverage Status](https://coveralls.io/repos/github/simonsobs/archivist/badge.svg?branch=main)](https://coveralls.io/github/simonsobs/archivist?branch=main) + Tools for offline data storage compatible with the Librarian. diff --git a/pyproject.toml b/pyproject.toml index b662569..3db50fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,10 +7,11 @@ name = "archivist" dynamic = ["version"] description = "Archival data storage for the Librarian" readme = "README.md" -requires-python = ">=3.9" +requires-python = ">=3.11" +license = "MIT" +license-files = ["LICENSE"] classifiers = [ "Programming Language :: Python :: 3", - "License :: OSI Approved :: MIT License", "Operating System :: OS Independent", ] dependencies = [ @@ -36,6 +37,8 @@ write_to = "src/archivist/_version.py" dev = [ "coverage>=7.10.7", "coveralls>=4.0.1", + "httpx>=0.28.1", + "hypothesis>=6.141.1", "isort>=6.1.0", "pre-commit>=4.3.0", "pytest>=8.4.2", @@ -113,3 +116,10 @@ mark-parentheses = false profile = "black" line_length = 120 skip = ["_version.py"] + +[tool.pytest.ini_options] +addopts = "--random-order --cov=archivist --cov-report=term-missing --cov-report=xml" + +[tool.coverage.run] +source = ["archivist"] +branch = true diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..5643039 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,137 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. +"""Shared pytest fixtures for the archivist test suite.""" + +from datetime import datetime, timezone + +import pytest + +import archivist.database as database +import archivist.queue as queue_module +import archivist.settings as settings_module +from archivist.settings import Settings +from archivist.storage.storage_disk import StorageDisk + + +@pytest.fixture(autouse=True) +def _reset_module_singletons(): + """ + Several modules cache lazily-created global state (the DB engine, + session maker, loaded settings, the in-memory status queue, and the + shared StorageDisk thread pool). Reset all of it before and after + every test so tests don't leak state into one another. + """ + + def _reset(): + database._engine = None + database._SessionMaker = None + settings_module._settings = None + queue_module._status_queue = None + StorageDisk._executor = None + + _reset() + yield + _reset() + + +@pytest.fixture +def archive_root(tmp_path): + path = tmp_path / "archive_root" + path.mkdir() + return path + + +@pytest.fixture +def local_root(tmp_path): + path = tmp_path / "local_root" + path.mkdir() + return path + + +@pytest.fixture +def settings(tmp_path, archive_root, local_root) -> Settings: + """A Settings instance backed by a throwaway on-disk sqlite database.""" + + db_path = tmp_path / "archivist_test.db" + return Settings( + database_driver="sqlite", + database=str(db_path), + archive_type="posix", + archive_root=str(archive_root), + local_root=str(local_root), + ) + + +@pytest.fixture +def use_settings(monkeypatch, settings, tmp_path): + """ + Make `get_settings()` return `settings`, regardless of whether the + caller does `import archivist.settings` or + `from archivist.settings import get_settings` (the latter binds its + own reference, so patching the module attribute alone would miss it). + We instead drive this through the same env-var + cache mechanism + `get_settings()` itself uses. + """ + + config_path = tmp_path / "test_config.json" + config_path.write_text(settings.model_dump_json()) + + monkeypatch.setenv("ARCHIVIST_CONFIG_PATH", str(config_path)) + monkeypatch.setattr(settings_module, "_settings", None) + + return settings_module.get_settings() + + +@pytest.fixture +def db_session(use_settings): + """A real database session against a freshly created schema.""" + + database.create_all() + session = database.get_session() + try: + yield session + finally: + session.close() + + +def make_manifest_entry(**overrides): + """Build a dict-form manifest entry, suitable for `ManifestEntry(**entry)`.""" + + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + entry = dict( + name="file.txt", + create_time=now, + size=1024, + checksum="0" * 64, + uploader="test-uploader", + source="/source", + instance_path="/source/file.txt", + instance_create_time=now, + instance_available=True, + outgoing_transfer_id=0, + ) + entry.update(overrides) + return entry + + +def make_manifest_entry_json(**overrides): + """Like `make_manifest_entry`, but with datetimes as ISO strings for use as raw HTTP JSON bodies.""" + + entry = make_manifest_entry(**overrides) + for key in ("create_time", "instance_create_time"): + if isinstance(entry[key], datetime): + entry[key] = entry[key].isoformat() + return entry + + +@pytest.fixture +def manifest_entry_data(): + return make_manifest_entry() + + +@pytest.fixture +def manifest_request_data(manifest_entry_data): + return { + "librarian_name": "test-librarian", + "store_files": [manifest_entry_data], + } diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..dc4a95c --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,145 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +""" +Tests for the FastAPI routes in archivist.api. + +These build a bare FastAPI app that includes the same routers as the real +server (`archivist.server.main`), but skip the lifespan handler -- which +starts busy-looping background worker threads -- since it isn't relevant +to exercising the HTTP endpoints themselves. +""" + +import unittest + +import pytest +from conftest import make_manifest_entry_json +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from archivist.api import health_router +from archivist.api import router as api_router +from archivist.database import yield_session +from archivist.orm.archivequeue import ArchiveQueue +from archivist.settings import get_settings + + +def _build_app(settings, session_factory): + app = FastAPI() + app.include_router(api_router) + app.include_router(health_router) + + app.dependency_overrides[get_settings] = lambda: settings + app.dependency_overrides[yield_session] = session_factory + + return app + + +@pytest.mark.usefixtures("db_session") +class TestApiBase(unittest.TestCase): + @pytest.fixture(autouse=True) + def _inject(self, db_session, use_settings): + self.session = db_session + self.settings = use_settings + + def _session_factory(): + yield self.session + + self.app = _build_app(self.settings, _session_factory) + self.client = TestClient(self.app) + + +class TestHealthEndpoint(TestApiBase): + def test_health_returns_ok(self): + response = self.client.get("/health") + + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertEqual(body["status"], "ok") + self.assertEqual(body["name"], self.settings.displayed_site_name) + + +class TestArchiveEndpoint(TestApiBase): + def test_archive_valid_manifest_returns_manifest_id(self): + payload = { + "librarian_name": "test-librarian", + "store_files": [make_manifest_entry_json()], + } + + response = self.client.post("/api/v1/archive", json=payload) + + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertIn("manifest_id", body) + + def test_archive_persists_queue_item(self): + payload = { + "librarian_name": "test-librarian", + "store_files": [make_manifest_entry_json()], + } + + response = self.client.post("/api/v1/archive", json=payload) + manifest_id = response.json()["manifest_id"] + + item = self.session.query(ArchiveQueue).filter_by(manifest_id=manifest_id).one() + self.assertFalse(item.consumed) + self.assertFalse(item.completed) + + def test_archive_over_size_limit_returns_400(self): + self.settings.maximal_size_bytes = 100 + + payload = { + "librarian_name": "test-librarian", + "store_files": [make_manifest_entry_json(size=1_000_000)], + } + + response = self.client.post("/api/v1/archive", json=payload) + + self.assertEqual(response.status_code, 400) + self.assertIn("error", response.json()) + + def test_archive_empty_store_files_succeeds(self): + payload = {"librarian_name": "test-librarian", "store_files": []} + + response = self.client.post("/api/v1/archive", json=payload) + + self.assertEqual(response.status_code, 200) + + def test_archive_missing_librarian_name_returns_422(self): + payload = {"store_files": [make_manifest_entry_json()]} + + response = self.client.post("/api/v1/archive", json=payload) + + self.assertEqual(response.status_code, 422) + + +class TestExtractEndpoint(TestApiBase): + """ + `/api/v1/extract` is currently an unimplemented stub (`archivist.api.extract.extract` + just does `pass`, i.e. returns `None`). Since the route declares + `response_model=ManifestResponse | ManifestFailedResponse`, returning `None` + fails FastAPI's response validation. These tests document that current, + not-yet-implemented behavior rather than asserting a "correct" outcome. + """ + + def test_extract_stub_raises_response_validation_error(self): + no_raise_client = TestClient(self.app, raise_server_exceptions=False) + + payload = { + "librarian_name": "test-librarian", + "store_files": [make_manifest_entry_json()], + } + response = no_raise_client.post("/api/v1/extract", json=payload) + + self.assertEqual(response.status_code, 500) + + def test_extract_missing_librarian_name_returns_422(self): + payload = {"store_files": [make_manifest_entry_json()]} + + response = self.client.post("/api/v1/extract", json=payload) + + self.assertEqual(response.status_code, 422) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_archive.py b/tests/test_archive.py new file mode 100644 index 0000000..282b083 --- /dev/null +++ b/tests/test_archive.py @@ -0,0 +1,94 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.core.archive.Archive.""" + +import unittest +from pathlib import Path + +from conftest import make_manifest_entry + +from archivist.core.archive import Archive + + +class TestArchiveProperties(unittest.TestCase): + def test_defaults_are_none(self): + archive = Archive() + self.assertIsNone(archive.manifest) + self.assertIsNone(archive.manifest_id) + self.assertIsNone(archive.local_root) + self.assertIsNone(archive.archive_root) + self.assertIsNone(archive.checksum) + + def test_constructor_sets_properties(self): + manifest = {"store_files": []} + archive = Archive( + manifest=manifest, + manifest_id="abc", + local_root="/local", + archive_root="/archive", + type="posix", + ) + self.assertEqual(archive.manifest, manifest) + self.assertEqual(archive.manifest_id, "abc") + self.assertEqual(archive.local_root, "/local") + self.assertEqual(archive.archive_root, "/archive") + + def test_repr_includes_key_fields(self): + archive = Archive(manifest_id="abc", type="posix") + text = repr(archive) + self.assertIn("abc", text) + self.assertIn("posix", text) + + +class TestCreateArchivePosix(unittest.TestCase): + def test_builds_src_dst_pairs_relative_to_local_root(self): + manifest = { + "store_files": [ + make_manifest_entry(instance_path="/local/sub/file.txt"), + ] + } + archive = Archive( + manifest=manifest, + local_root="/local", + archive_root="/archive", + type="posix", + ) + + pairs = archive.create_archive() + + self.assertEqual(len(pairs), 1) + src, dst = pairs[0] + self.assertEqual(src, Path("/local/sub/file.txt")) + self.assertEqual(dst, Path("/archive/sub/file.txt")) + + def test_multiple_entries_preserve_order(self): + manifest = { + "store_files": [ + make_manifest_entry(instance_path="/local/a.txt"), + make_manifest_entry(instance_path="/local/b.txt"), + ] + } + archive = Archive(manifest=manifest, local_root="/local", archive_root="/archive", type="posix") + + pairs = archive.create_archive() + + self.assertEqual([str(src) for src, _ in pairs], ["/local/a.txt", "/local/b.txt"]) + + def test_empty_store_files_returns_empty_list(self): + archive = Archive(manifest={"store_files": []}, local_root="/local", archive_root="/archive", type="posix") + self.assertEqual(archive.create_archive(), []) + + +class TestCreateArchiveOtherTypes(unittest.TestCase): + def test_hpss_returns_none(self): + archive = Archive(manifest={"store_files": []}, type="hpss") + self.assertIsNone(archive.create_archive()) + + def test_unknown_type_returns_none(self): + archive = Archive(manifest={"store_files": []}, type="something-else") + self.assertIsNone(archive.create_archive()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_archivequeue.py b/tests/test_archivequeue.py new file mode 100644 index 0000000..630bf16 --- /dev/null +++ b/tests/test_archivequeue.py @@ -0,0 +1,119 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.orm.archivequeue.ArchiveQueue, against a real sqlite DB.""" + +import unittest + +import pytest + +from archivist.orm.archivequeue import ArchiveQueue + + +@pytest.mark.usefixtures("db_session") +class TestArchiveQueueBase(unittest.TestCase): + """Base TestCase that pulls the pytest `db_session` fixture into `self.session`.""" + + @pytest.fixture(autouse=True) + def _inject_session(self, db_session): + self.session = db_session + + +class TestNewItem(TestArchiveQueueBase): + def test_new_item_defaults(self): + item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=["/a"], root="/root") + self.assertEqual(item.manifest_id, "m1") + self.assertEqual(item.paths, ["/a"]) + self.assertEqual(item.root, "/root") + self.assertEqual(item.retries, 0) + self.assertFalse(item.consumed) + self.assertFalse(item.completed) + self.assertFalse(item.failed) + + def test_new_item_persists(self): + item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=["/a"], root="/root") + self.session.add(item) + self.session.commit() + + fetched = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() + self.assertEqual(fetched.root, "/root") + + def test_manifest_id_must_be_unique(self): + from sqlalchemy.exc import IntegrityError + + self.session.add(ArchiveQueue.new_item(manifest_id="dup", manifest="{}", paths=[], root="/root")) + self.session.commit() + + self.session.add(ArchiveQueue.new_item(manifest_id="dup", manifest="{}", paths=[], root="/root")) + with self.assertRaises(IntegrityError): + self.session.commit() + + +class TestDequeue(TestArchiveQueueBase): + def test_dequeue_returns_none_when_empty(self): + self.assertIsNone(ArchiveQueue.dequeue(self.session)) + + def test_dequeue_marks_item_consumed(self): + item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") + self.session.add(item) + self.session.commit() + + dequeued = ArchiveQueue.dequeue(self.session) + + self.assertEqual(dequeued.manifest_id, "m1") + self.assertTrue(dequeued.consumed) + self.assertIsNotNone(dequeued.consumed_time) + + def test_dequeue_returns_oldest_first(self): + import time + + first = ArchiveQueue.new_item(manifest_id="first", manifest="{}", paths=[], root="/root") + self.session.add(first) + self.session.commit() + + time.sleep(0.01) + + second = ArchiveQueue.new_item(manifest_id="second", manifest="{}", paths=[], root="/root") + self.session.add(second) + self.session.commit() + + dequeued = ArchiveQueue.dequeue(self.session) + + self.assertEqual(dequeued.manifest_id, "first") + + def test_dequeue_skips_already_consumed_items(self): + item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") + self.session.add(item) + self.session.commit() + + ArchiveQueue.dequeue(self.session) + + self.assertIsNone(ArchiveQueue.dequeue(self.session)) + + +class TestCompleteAndFail(TestArchiveQueueBase): + def test_complete_sets_flags(self): + item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") + self.session.add(item) + self.session.commit() + + item.complete(self.session) + + self.assertTrue(item.completed) + self.assertFalse(item.failed) + self.assertIsNotNone(item.completed_time) + + def test_fail_sets_flags(self): + item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") + self.session.add(item) + self.session.commit() + + item.fail(self.session) + + self.assertTrue(item.completed) + self.assertTrue(item.failed) + self.assertIsNotNone(item.completed_time) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_classes.py b/tests/test_classes.py index 2db294e..1fdc8ac 100644 --- a/tests/test_classes.py +++ b/tests/test_classes.py @@ -1,16 +1,31 @@ # Copyright (c) 2025-2026 Simons Observatory. # Full license can be found in the top level "LICENSE" file. -"""Test package imports.""" +"""Test top-level package imports and construction of the public classes.""" + +import unittest import archivist from archivist import Archive, RegistryLibrarian, RegistrySqlite, StorageDisk, StorageHPSS -from archivist import _version as pkg_version -def test_construction(): - archive = Archive(paths=[]) - reg_lib = RegistryLibrarian() - reg_sql = RegistrySqlite() - store_disk = StorageDisk(".") - store_hpss = StorageHPSS() +class TestPackageImports(unittest.TestCase): + def test_version_is_exposed(self): + self.assertIsInstance(archivist.__version__, str) + + def test_construction(self): + archive = Archive() + reg_lib = RegistryLibrarian() + reg_sql = RegistrySqlite() + store_disk = StorageDisk() + store_hpss = StorageHPSS() + + self.assertIsInstance(archive, Archive) + self.assertIsInstance(reg_lib, RegistryLibrarian) + self.assertIsInstance(reg_sql, RegistrySqlite) + self.assertIsInstance(store_disk, StorageDisk) + self.assertIsInstance(store_hpss, StorageHPSS) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_database.py b/tests/test_database.py new file mode 100644 index 0000000..6c4af90 --- /dev/null +++ b/tests/test_database.py @@ -0,0 +1,77 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.database engine/session helpers.""" + +import unittest +import unittest.mock + +import pytest +from sqlalchemy.orm import Session + +import archivist.database as database + + +@pytest.mark.usefixtures("use_settings") +class TestDatabaseBase(unittest.TestCase): + @pytest.fixture(autouse=True) + def _inject(self, use_settings): + self.settings = use_settings + + +class TestGetEngine(TestDatabaseBase): + def test_engine_is_created_lazily_and_cached(self): + self.assertIsNone(database._engine) + + engine = database.get_engine() + + self.assertIsNotNone(database._engine) + self.assertIs(engine, database.get_engine()) + + +class TestGetSessionmaker(TestDatabaseBase): + def test_sessionmaker_is_created_lazily_and_cached(self): + self.assertIsNone(database._SessionMaker) + + maker = database.get_sessionmaker() + + self.assertIsNotNone(database._SessionMaker) + self.assertIs(maker, database.get_sessionmaker()) + + +class TestGetSession(TestDatabaseBase): + def test_returns_a_session_the_caller_must_close(self): + database.create_all() + session = database.get_session() + try: + self.assertIsInstance(session, Session) + finally: + session.close() + + +class TestYieldSession(TestDatabaseBase): + def test_yields_a_session_and_closes_it_afterwards(self): + database.create_all() + generator = database.yield_session() + + session = next(generator) + self.assertIsInstance(session, Session) + + # Draining the generator triggers the `finally: session.close()`. + with self.assertRaises(StopIteration): + next(generator) + + def test_closes_session_even_if_consumer_raises(self): + database.create_all() + generator = database.yield_session() + session = next(generator) + + with unittest.mock.patch.object(session, "close") as mock_close: + with self.assertRaises(RuntimeError): + generator.throw(RuntimeError("boom")) + + mock_close.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_main_cli.py b/tests/test_main_cli.py new file mode 100644 index 0000000..c9ab97b --- /dev/null +++ b/tests/test_main_cli.py @@ -0,0 +1,205 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for the archivist.__main__ click CLI.""" + +import os +import unittest +import unittest.mock + +import pytest +import requests +from click.testing import CliRunner + +import archivist.__main__ as cli_module +from archivist.__main__ import main + + +@pytest.fixture(autouse=True) +def _restore_config_path_env(): + """ + The CLI's `main()` callback mutates `os.environ` directly (not via + click/pydantic-settings helpers), so make sure it doesn't leak the path + to a tmp_path config file that gets torn down after this test into + later, unrelated tests. + """ + + original = os.environ.get("ARCHIVIST_CONFIG_PATH") + yield + if original is None: + os.environ.pop("ARCHIVIST_CONFIG_PATH", None) + else: + os.environ["ARCHIVIST_CONFIG_PATH"] = original + + +@pytest.fixture +def cli_config_path(settings, tmp_path): + config_path = tmp_path / "cli_config.json" + config_path.write_text(settings.model_dump_json()) + return str(config_path) + + +@pytest.fixture +def runner(): + return CliRunner() + + +class TestMainRequiresConfig(unittest.TestCase): + def test_missing_config_exits_1(self): + runner = CliRunner() + result = runner.invoke(main, ["archive", "--source-path", "."]) + + self.assertEqual(result.exit_code, 1) + self.assertIn("requires a configuration file", result.output) + + def test_nonexistent_config_is_a_usage_error(self): + runner = CliRunner() + result = runner.invoke(main, ["-c", "/no/such/config.json", "archive", "--source-path", "."]) + + self.assertEqual(result.exit_code, 2) + + +class TestArchiveCommand: + def test_archive_single_file_success(self, runner, cli_config_path, tmp_path): + source = tmp_path / "file.txt" + source.write_text("hello world") + + fake_response = unittest.mock.Mock(status_code=200) + fake_response.json.return_value = {"manifest_id": "abc-123"} + + with unittest.mock.patch.object(cli_module.requests, "post", return_value=fake_response) as mock_post: + result = runner.invoke(main, ["-c", cli_config_path, "archive", "--source-path", str(source)]) + + assert result.exit_code == 0 + assert "Archive succeeded, manifest_id=abc-123" in result.output + + mock_post.assert_called_once() + _, kwargs = mock_post.call_args + body = kwargs["json"] + assert body["librarian_name"] == "librarian" + assert len(body["store_files"]) == 1 + assert body["store_files"][0]["name"] == "file.txt" + + def test_archive_computes_correct_checksum(self, runner, cli_config_path, tmp_path): + import hashlib + + source = tmp_path / "file.txt" + source.write_bytes(b"some content to checksum") + + fake_response = unittest.mock.Mock(status_code=200) + fake_response.json.return_value = {"manifest_id": "abc-123"} + + with unittest.mock.patch.object(cli_module.requests, "post", return_value=fake_response) as mock_post: + runner.invoke(main, ["-c", cli_config_path, "archive", "--source-path", str(source)]) + + _, kwargs = mock_post.call_args + expected = hashlib.sha256(b"some content to checksum").hexdigest() + assert kwargs["json"]["store_files"][0]["checksum"] == expected + + def test_archive_directory_walks_all_files(self, runner, cli_config_path, tmp_path): + source_dir = tmp_path / "adir" + (source_dir / "sub").mkdir(parents=True) + (source_dir / "a.txt").write_text("a") + (source_dir / "sub" / "b.txt").write_text("b") + + fake_response = unittest.mock.Mock(status_code=200) + fake_response.json.return_value = {"manifest_id": "abc-123"} + + with unittest.mock.patch.object(cli_module.requests, "post", return_value=fake_response) as mock_post: + result = runner.invoke(main, ["-c", cli_config_path, "archive", "--source-path", str(source_dir)]) + + assert result.exit_code == 0 + _, kwargs = mock_post.call_args + names = sorted(entry["name"] for entry in kwargs["json"]["store_files"]) + assert names == ["a.txt", "b.txt"] + + def test_archive_uses_custom_librarian_name(self, runner, cli_config_path, tmp_path): + source = tmp_path / "file.txt" + source.write_text("hello") + + fake_response = unittest.mock.Mock(status_code=200) + fake_response.json.return_value = {"manifest_id": "abc-123"} + + with unittest.mock.patch.object(cli_module.requests, "post", return_value=fake_response) as mock_post: + runner.invoke( + main, + [ + "-c", + cli_config_path, + "archive", + "--source-path", + str(source), + "--librarian-name", + "custom-librarian", + ], + ) + + _, kwargs = mock_post.call_args + assert kwargs["json"]["librarian_name"] == "custom-librarian" + assert kwargs["json"]["store_files"][0]["uploader"] == "custom-librarian" + + def test_archive_server_error_prints_error_and_exits_cleanly(self, runner, cli_config_path, tmp_path): + source = tmp_path / "file.txt" + source.write_text("hello") + + fake_response = unittest.mock.Mock(status_code=400) + fake_response.json.return_value = {"error": "too big"} + + with unittest.mock.patch.object(cli_module.requests, "post", return_value=fake_response): + result = runner.invoke(main, ["-c", cli_config_path, "archive", "--source-path", str(source)]) + + assert result.exit_code == 0 + assert "Archive failed: too big" in result.output + + def test_archive_connection_error_propagates(self, runner, cli_config_path, tmp_path): + source = tmp_path / "file.txt" + source.write_text("hello") + + with unittest.mock.patch.object( + cli_module.requests, "post", side_effect=requests.exceptions.ConnectionError("no server") + ): + result = runner.invoke(main, ["-c", cli_config_path, "archive", "--source-path", str(source)]) + + assert result.exit_code != 0 + assert isinstance(result.exception, requests.exceptions.ConnectionError) + + def test_archive_missing_source_path_is_usage_error(self, runner, cli_config_path): + result = runner.invoke(main, ["-c", cli_config_path, "archive", "--source-path", "/no/such/file"]) + + assert result.exit_code == 2 + + +class TestExtractCommand: + def test_extract_command_currently_raises_type_error(self, runner, cli_config_path): + """ + `extract()`'s signature is `(ctx, archive, dest_path)`, but the command + declares `@click.argument("cmd")`, which click passes as a `cmd` kwarg + the function doesn't accept. This documents the current, broken + behavior rather than asserting the (not-yet-implemented) command works. + """ + result = runner.invoke( + main, + ["-c", cli_config_path, "extract", "cmd", "--archive", "foo", "--dest-path", "/tmp/out"], + ) + + assert result.exit_code != 0 + assert isinstance(result.exception, TypeError) + + +class TestStartServerCommand: + def test_start_server_invokes_uvicorn_with_settings(self, runner, cli_config_path, settings): + with unittest.mock.patch("uvicorn.run") as mock_run: + result = runner.invoke(main, ["-c", cli_config_path, "start-server"]) + + assert result.exit_code == 0 + mock_run.assert_called_once_with( + "archivist.server:main", + host=settings.host, + port=settings.port, + log_level=settings.log_level.lower(), + factory=True, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..803305e --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,67 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.core.models pydantic models.""" + +import unittest + +from conftest import make_manifest_entry +from pydantic import ValidationError + +from archivist.core.models import ( + ManifestEntry, + ManifestFailedResponse, + ManifestRequest, + ManifestResponse, +) + + +class TestManifestEntry(unittest.TestCase): + def test_valid_entry_constructs(self): + entry = ManifestEntry(**make_manifest_entry()) + self.assertEqual(entry.name, "file.txt") + self.assertEqual(entry.size, 1024) + self.assertTrue(entry.instance_available) + + def test_missing_required_field_raises(self): + data = make_manifest_entry() + del data["checksum"] + with self.assertRaises(ValidationError): + ManifestEntry(**data) + + def test_wrong_type_raises(self): + data = make_manifest_entry(size="not-a-number") + with self.assertRaises(ValidationError): + ManifestEntry(**data) + + +class TestManifestRequest(unittest.TestCase): + def test_valid_request_constructs(self): + request = ManifestRequest( + librarian_name="lib", + store_files=[make_manifest_entry(), make_manifest_entry(name="other.txt")], + ) + self.assertEqual(request.librarian_name, "lib") + self.assertEqual(len(request.store_files), 2) + + def test_empty_store_files_is_allowed(self): + request = ManifestRequest(librarian_name="lib", store_files=[]) + self.assertEqual(request.store_files, []) + + def test_missing_librarian_name_raises(self): + with self.assertRaises(ValidationError): + ManifestRequest(store_files=[]) + + +class TestManifestResponses(unittest.TestCase): + def test_manifest_response(self): + response = ManifestResponse(manifest_id="abc-123") + self.assertEqual(response.manifest_id, "abc-123") + + def test_manifest_failed_response(self): + response = ManifestFailedResponse(error="boom") + self.assertEqual(response.error, "boom") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_queue.py b/tests/test_queue.py new file mode 100644 index 0000000..696441f --- /dev/null +++ b/tests/test_queue.py @@ -0,0 +1,53 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.queue.Queue and get_status_queue.""" + +import queue as stdlib_queue +import unittest + +from archivist.queue import Queue, get_status_queue + + +class TestQueue(unittest.TestCase): + def test_new_queue_is_empty(self): + q = Queue() + self.assertEqual(q.size, 0) + + def test_enqueue_increases_size(self): + q = Queue() + q.enqueue("item") + self.assertEqual(q.size, 1) + + def test_dequeue_returns_enqueued_item_fifo(self): + q = Queue() + q.enqueue("first") + q.enqueue("second") + + self.assertEqual(q.dequeue(), "first") + self.assertEqual(q.dequeue(), "second") + + def test_dequeue_nonblocking_raises_when_empty(self): + q = Queue() + with self.assertRaises(stdlib_queue.Empty): + q.dequeue(block=False) + + def test_task_done_does_not_raise_after_dequeue(self): + q = Queue() + q.enqueue("item") + q.dequeue() + q.task_done() + + +class TestGetStatusQueue(unittest.TestCase): + def test_returns_singleton(self): + first = get_status_queue() + second = get_status_queue() + self.assertIs(first, second) + + def test_returns_a_queue_instance(self): + self.assertIsInstance(get_status_queue(), Queue) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..c862a09 --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,36 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.core.registry classes.""" + +import unittest + +from archivist.core.registry import Registry, RegistryLibrarian, RegistrySqlite + + +class TestRegistryBase(unittest.TestCase): + def test_register_raises_not_implemented(self): + registry = Registry() + with self.assertRaises(NotImplementedError): + registry.register(archive=None, storage_info=None) + + def test_remove_raises_not_implemented(self): + registry = Registry() + with self.assertRaises(NotImplementedError): + registry.remove(archive_name="foo") + + +class TestRegistrySubclasses(unittest.TestCase): + def test_registry_sqlite_register_and_remove_are_noops(self): + registry = RegistrySqlite() + self.assertIsNone(registry.register(archive=None, storage_info=None)) + self.assertIsNone(registry.remove(archive_name="foo")) + + def test_registry_librarian_register_and_remove_are_noops(self): + registry = RegistryLibrarian() + self.assertIsNone(registry.register(archive=None, storage_info=None)) + self.assertIsNone(registry.remove(archive_name="foo")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..ff76a7a --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,198 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.server (FastAPI app factory, lifespan, worker loops).""" + +import asyncio +import threading +import unittest +import unittest.mock + +import pytest + +import archivist.server as server_module +from archivist.settings import Settings + + +@pytest.fixture(autouse=True) +def _reset_stop_event(): + """`_archive_stop_event` is module-level shared state; keep it clean per test.""" + + server_module._archive_stop_event.clear() + yield + server_module._archive_stop_event.clear() + + +class TestMain(unittest.TestCase): + def test_main_builds_app_with_settings_metadata(self): + settings = Settings( + displayed_site_name="My Archivist", + displayed_site_description="A description", + debug=False, + ) + with unittest.mock.patch.object(server_module, "server_settings", settings): + app = server_module.main() + + self.assertEqual(app.title, "My Archivist") + self.assertEqual(app.description, "A description") + self.assertIsNone(app.openapi_url) + + def test_main_enables_openapi_when_debug(self): + settings = Settings(debug=True) + with unittest.mock.patch.object(server_module, "server_settings", settings): + app = server_module.main() + + self.assertEqual(app.openapi_url, "/api/v2/openapi.json") + + def test_main_registers_health_and_archive_routes(self): + settings = Settings() + with unittest.mock.patch.object(server_module, "server_settings", settings): + app = server_module.main() + + paths = set(app.openapi()["paths"].keys()) + self.assertIn("/health", paths) + self.assertIn("/api/v1/archive", paths) + + def test_main_wires_up_the_shared_lifespan_handler(self): + """ + `main()` doesn't set `app.router.lifespan_context` to + `slack_post_at_startup_shutdown` directly -- FastAPI wraps it in an + internal `merged_lifespan` closure. Drive it through a real + (mocked-out) startup/shutdown cycle instead of asserting identity. + """ + settings = Settings() + + async def _run(): + with unittest.mock.patch.object(server_module, "server_settings", settings): + app = server_module.main() + + with ( + unittest.mock.patch.object(server_module, "_archive_worker_loop", lambda: None), + unittest.mock.patch.object(server_module, "_status_worker_loop", lambda: None), + unittest.mock.patch("archivist.database.create_all") as mock_create_all, + ): + async with app.router.lifespan_context(app): + pass + + mock_create_all.assert_called_once() + + asyncio.run(_run()) + + +class TestArchiveWorkerLoop(unittest.TestCase): + def test_calls_start_archive_until_stopped(self): + settings = Settings(name="test-server") + call_count = 0 + + def _fake_start_archive(librarian_name): + nonlocal call_count + call_count += 1 + if call_count >= 3: + server_module._archive_stop_event.set() + + with ( + unittest.mock.patch.object(server_module, "server_settings", settings), + unittest.mock.patch("archivist.tasks.archive.start_archive", side_effect=_fake_start_archive), + ): + thread = threading.Thread(target=server_module._archive_worker_loop) + thread.start() + thread.join(timeout=5) + + self.assertFalse(thread.is_alive()) + self.assertGreaterEqual(call_count, 3) + + def test_survives_exceptions_and_keeps_looping(self): + call_count = 0 + + def _fake_start_archive(librarian_name): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError("boom") + server_module._archive_stop_event.set() + + with unittest.mock.patch("archivist.tasks.archive.start_archive", side_effect=_fake_start_archive): + thread = threading.Thread(target=server_module._archive_worker_loop) + thread.start() + thread.join(timeout=5) + + self.assertFalse(thread.is_alive()) + self.assertGreaterEqual(call_count, 2) + + def test_noop_when_stop_event_already_set(self): + server_module._archive_stop_event.set() + + with unittest.mock.patch("archivist.tasks.archive.start_archive") as mock_start: + server_module._archive_worker_loop() + + mock_start.assert_not_called() + + +class TestStatusWorkerLoop(unittest.TestCase): + def test_calls_process_status_queue_until_stopped(self): + call_count = 0 + + def _fake_process(): + nonlocal call_count + call_count += 1 + if call_count >= 3: + server_module._archive_stop_event.set() + + with unittest.mock.patch("archivist.tasks.archive.process_status_queue", side_effect=_fake_process): + thread = threading.Thread(target=server_module._status_worker_loop) + thread.start() + thread.join(timeout=5) + + self.assertFalse(thread.is_alive()) + self.assertGreaterEqual(call_count, 3) + + def test_survives_exceptions_and_keeps_looping(self): + call_count = 0 + + def _fake_process(): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError("boom") + server_module._archive_stop_event.set() + + with unittest.mock.patch("archivist.tasks.archive.process_status_queue", side_effect=_fake_process): + thread = threading.Thread(target=server_module._status_worker_loop) + thread.start() + thread.join(timeout=5) + + self.assertFalse(thread.is_alive()) + self.assertGreaterEqual(call_count, 2) + + +class TestLifespan(unittest.TestCase): + def test_lifespan_creates_schema_starts_workers_and_stops_on_exit(self): + submitted = [] + + def _fake_archive_loop(): + submitted.append("archive") + + def _fake_status_loop(): + submitted.append("status") + + async def _run(): + app = unittest.mock.Mock() + async with server_module.slack_post_at_startup_shutdown(app): + # Give the submitted worker threads a beat to run. + await asyncio.sleep(0.1) + + with ( + unittest.mock.patch.object(server_module, "_archive_worker_loop", _fake_archive_loop), + unittest.mock.patch.object(server_module, "_status_worker_loop", _fake_status_loop), + unittest.mock.patch("archivist.database.create_all") as mock_create_all, + ): + asyncio.run(_run()) + + mock_create_all.assert_called_once() + self.assertIn("archive", submitted) + self.assertIn("status", submitted) + self.assertTrue(server_module._archive_stop_event.is_set()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..aa1dd0e --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,201 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.settings.Settings and get_settings().""" + +import json +import unittest +import unittest.mock + +import archivist.settings as settings_module +from archivist.settings import Settings, get_settings + + +class TestSettingsDefaults(unittest.TestCase): + def test_defaults(self): + settings = Settings() + self.assertEqual(settings.name, "archivist_server") + self.assertFalse(settings.debug) + self.assertEqual(settings.database_driver, "sqlite") + self.assertEqual(settings.archive_type, "posix") + self.assertEqual(settings.host, "0.0.0.0") + self.assertEqual(settings.port, 8080) + + +class TestSqlalchemyDatabaseUri(unittest.TestCase): + def test_sqlite_uri(self): + settings = Settings(database_driver="sqlite", database="/tmp/foo.db") + uri = settings.sqlalchemy_database_uri + self.assertEqual(uri.drivername, "sqlite") + self.assertEqual(uri.database, "/tmp/foo.db") + + def test_postgres_uri_includes_credentials(self): + settings = Settings( + database_driver="postgresql", + database_user="user", + database_password="pass", + database_host="localhost", + database_port=5432, + database="archivist", + ) + uri = settings.sqlalchemy_database_uri + self.assertEqual(uri.drivername, "postgresql") + self.assertEqual(uri.username, "user") + self.assertEqual(uri.password, "pass") + self.assertEqual(uri.host, "localhost") + self.assertEqual(uri.port, 5432) + self.assertEqual(uri.database, "archivist") + + +class TestFromFile(unittest.TestCase): + def test_from_file_loads_values(self): + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + config_path = Path(tmp_dir) / "config.json" + config_path.write_text(json.dumps({"name": "my-archivist", "port": 9999})) + + settings = Settings.from_file(config_path) + + self.assertEqual(settings.name, "my-archivist") + self.assertEqual(settings.port, 9999) + + def test_from_file_missing_raises(self): + with self.assertRaises(FileNotFoundError): + Settings.from_file("/nonexistent/path/config.json") + + +class TestSecretFiles(unittest.TestCase): + def test_globus_client_secret_read_from_file(self): + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + secret_path = Path(tmp_dir) / "secret.txt" + secret_path.write_text("super-secret\n") + + settings = Settings(globus_client_secret_file=secret_path) + + self.assertEqual(settings.globus_client_secret, "super-secret") + + def test_database_password_read_from_file(self): + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + secret_path = Path(tmp_dir) / "dbpass.txt" + secret_path.write_text("db-secret\n") + + settings = Settings(database_password_file=secret_path) + + self.assertEqual(settings.database_password, "db-secret") + + +class TestGetSettings(unittest.TestCase): + def setUp(self): + self._original = settings_module._settings + settings_module._settings = None + + def tearDown(self): + settings_module._settings = self._original + + def test_get_settings_from_env_config_path(self): + import tempfile + from pathlib import Path + + from pytest import MonkeyPatch + + mp = MonkeyPatch() + try: + with tempfile.TemporaryDirectory() as tmp_dir: + config_path = Path(tmp_dir) / "config.json" + config_path.write_text(json.dumps({"name": "from-env-config"})) + + mp.setenv("ARCHIVIST_CONFIG_PATH", str(config_path)) + + settings = get_settings() + + self.assertEqual(settings.name, "from-env-config") + finally: + mp.undo() + + def test_get_settings_falls_back_to_defaults(self): + from pytest import MonkeyPatch + + mp = MonkeyPatch() + try: + mp.delenv("ARCHIVIST_CONFIG_PATH", raising=False) + + settings = get_settings() + + self.assertEqual(settings.name, "archivist_server") + finally: + mp.undo() + + def test_server_settings_attribute_lazily_loads(self): + from pytest import MonkeyPatch + + mp = MonkeyPatch() + try: + mp.delenv("ARCHIVIST_CONFIG_PATH", raising=False) + + self.assertIsInstance(settings_module.server_settings, Settings) + finally: + mp.undo() + + def test_unknown_attribute_raises(self): + with self.assertRaises(AttributeError): + settings_module.not_a_real_attribute + + def test_server_settings_returns_already_loaded_settings(self): + sentinel = Settings(name="already-loaded") + settings_module._settings = sentinel + + self.assertIs(settings_module.server_settings, sentinel) + + def test_get_settings_from_file_with_invalid_content_raises(self): + import tempfile + from pathlib import Path + + from pydantic import ValidationError + from pytest import MonkeyPatch + + mp = MonkeyPatch() + try: + with tempfile.TemporaryDirectory() as tmp_dir: + config_path = Path(tmp_dir) / "config.json" + config_path.write_text(json.dumps({"port": "not-a-port"})) + + mp.setenv("ARCHIVIST_CONFIG_PATH", str(config_path)) + + with self.assertRaises(ValidationError): + get_settings() + finally: + mp.undo() + + def test_get_settings_default_construction_error_is_reraised(self): + from pydantic import BaseModel, ValidationError + from pytest import MonkeyPatch + + class _RequiresField(BaseModel): + required: int + + try: + _RequiresField() + except ValidationError as captured: + validation_error = captured + + mp = MonkeyPatch() + try: + mp.delenv("ARCHIVIST_CONFIG_PATH", raising=False) + mp.setattr(settings_module, "Settings", unittest.mock.Mock(side_effect=validation_error)) + + with self.assertRaises(ValidationError): + get_settings() + finally: + mp.undo() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_storage_base.py b/tests/test_storage_base.py new file mode 100644 index 0000000..8a66016 --- /dev/null +++ b/tests/test_storage_base.py @@ -0,0 +1,69 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.storage.base.Storage.""" + +import unittest + +from archivist.storage.base import Storage + + +class TestStorageBase(unittest.TestCase): + def test_store_raises_not_implemented(self): + storage = Storage() + with self.assertRaises(NotImplementedError): + storage.store(archive=None) + + def test_extract_raises_not_implemented(self): + storage = Storage() + with self.assertRaises(NotImplementedError): + storage.extract(archive_name="a", storage_info={}, outdir="/tmp/out") + + def test_settings_defaults_to_none(self): + storage = Storage() + self.assertIsNone(storage._settings) + + def test_settings_stored_when_provided(self): + sentinel = object() + storage = Storage(settings=sentinel) + self.assertIs(storage._settings, sentinel) + + +class _RecordingStorage(Storage): + """Subclass that records the arguments _extract/_store receive.""" + + def __init__(self, settings=None): + super().__init__(settings=settings) + self.store_calls = [] + self.extract_calls = [] + + def _store(self, archive): + self.store_calls.append(archive) + return {"stored": True} + + def _extract(self, archive_name, storage_info, outdir, paths=None): + self.extract_calls.append((archive_name, storage_info, outdir, paths)) + + +class TestStorageDelegation(unittest.TestCase): + def test_store_delegates_to_store_impl_and_returns_its_value(self): + storage = _RecordingStorage() + result = storage.store(archive="my-archive") + + self.assertEqual(storage.store_calls, ["my-archive"]) + self.assertEqual(result, {"stored": True}) + + def test_extract_ignores_caller_supplied_paths(self): + """ + `Storage.extract()` currently hardcodes `paths=None` in its call to + `_extract()` regardless of what the caller passes in -- this test + documents that existing (surprising) behavior. + """ + storage = _RecordingStorage() + storage.extract(archive_name="a", storage_info={"k": "v"}, outdir="/tmp/out", paths=["only", "these"]) + + self.assertEqual(storage.extract_calls, [("a", {"k": "v"}, "/tmp/out", None)]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_storage_disk.py b/tests/test_storage_disk.py new file mode 100644 index 0000000..8dc0a2f --- /dev/null +++ b/tests/test_storage_disk.py @@ -0,0 +1,126 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.storage.storage_disk.StorageDisk.""" + +import unittest + +import pytest +from conftest import make_manifest_entry + +from archivist.core.archive import Archive +from archivist.storage.storage_disk import StorageDisk + + +@pytest.mark.usefixtures("use_settings") +class TestStorageDiskBase(unittest.TestCase): + @pytest.fixture(autouse=True) + def _inject(self, use_settings, local_root, archive_root): + self.settings = use_settings + self.local_root = local_root + self.archive_root = archive_root + + +class TestStoreSingleFile(TestStorageDiskBase): + def test_store_copies_file_to_archive_root(self): + source = self.local_root / "file.txt" + source.write_text("hello") + + manifest = {"store_files": [make_manifest_entry(instance_path=str(source))]} + archive = Archive( + manifest=manifest, + local_root=str(self.local_root), + archive_root=str(self.archive_root), + type="posix", + ) + + storage = StorageDisk(settings=self.settings) + task = storage.store(archive) + task.future.result(timeout=5) + + dest = self.archive_root / "file.txt" + self.assertTrue(dest.exists()) + self.assertEqual(dest.read_text(), "hello") + + def test_store_copies_nested_file_preserving_relative_path(self): + nested_dir = self.local_root / "sub" / "dir" + nested_dir.mkdir(parents=True) + source = nested_dir / "nested.txt" + source.write_text("nested-content") + + manifest = {"store_files": [make_manifest_entry(instance_path=str(source))]} + archive = Archive( + manifest=manifest, + local_root=str(self.local_root), + archive_root=str(self.archive_root), + type="posix", + ) + + storage = StorageDisk(settings=self.settings) + task = storage.store(archive) + task.future.result(timeout=5) + + dest = self.archive_root / "sub" / "dir" / "nested.txt" + self.assertTrue(dest.exists()) + self.assertEqual(dest.read_text(), "nested-content") + + def test_store_returns_self_with_archive_and_future(self): + source = self.local_root / "file.txt" + source.write_text("hello") + + manifest = {"store_files": [make_manifest_entry(instance_path=str(source))]} + archive = Archive( + manifest=manifest, + local_root=str(self.local_root), + archive_root=str(self.archive_root), + type="posix", + ) + + storage = StorageDisk(settings=self.settings) + result = storage.store(archive) + result.future.result(timeout=5) + + self.assertIs(result, storage) + self.assertIs(result._archive, archive) + self.assertIsNotNone(result.future) + + +class TestStoreDirectory(TestStorageDiskBase): + def test_store_copies_directory_tree(self): + src_dir = self.local_root / "adir" + src_dir.mkdir() + (src_dir / "inner.txt").write_text("inner-content") + + manifest = {"store_files": [make_manifest_entry(instance_path=str(src_dir))]} + archive = Archive( + manifest=manifest, + local_root=str(self.local_root), + archive_root=str(self.archive_root), + type="posix", + ) + + storage = StorageDisk(settings=self.settings) + task = storage.store(archive) + task.future.result(timeout=5) + + dest_file = self.archive_root / "adir" / "inner.txt" + self.assertTrue(dest_file.exists()) + self.assertEqual(dest_file.read_text(), "inner-content") + + +class TestSharedExecutor(TestStorageDiskBase): + def test_executor_is_shared_across_instances(self): + first = StorageDisk(settings=self.settings) + second = StorageDisk(settings=self.settings) + self.assertIs(first._executor, second._executor) + self.assertIs(StorageDisk._executor, first._executor) + + +class TestExtractStub(TestStorageDiskBase): + def test_extract_is_currently_a_noop_stub(self): + storage = StorageDisk(settings=self.settings) + self.assertIsNone(storage.extract(archive_name="a", storage_info={}, outdir=str(self.archive_root))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_storage_hpss.py b/tests/test_storage_hpss.py new file mode 100644 index 0000000..a44ff67 --- /dev/null +++ b/tests/test_storage_hpss.py @@ -0,0 +1,26 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.storage.storage_hpss.StorageHPSS (currently a stub backend).""" + +import unittest + +from archivist.storage.storage_hpss import StorageHPSS + + +class TestStorageHPSS(unittest.TestCase): + def test_construction_takes_no_arguments(self): + storage = StorageHPSS() + self.assertIsInstance(storage, StorageHPSS) + + def test_store_is_a_noop_stub(self): + storage = StorageHPSS() + self.assertIsNone(storage.store(archive=None)) + + def test_extract_is_a_noop_stub(self): + storage = StorageHPSS() + self.assertIsNone(storage.extract(archive_name="a", storage_info={}, outdir="/tmp/out")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tasks_archive.py b/tests/test_tasks_archive.py new file mode 100644 index 0000000..1df5ca5 --- /dev/null +++ b/tests/test_tasks_archive.py @@ -0,0 +1,177 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Integration tests for archivist.tasks.archive (start_archive, process_status_queue).""" + +import queue as stdlib_queue +import time +import unittest +import unittest.mock + +import pytest +from conftest import make_manifest_entry + +from archivist.core.models import ManifestRequest +from archivist.orm.archivequeue import ArchiveQueue +from archivist.queue import get_status_queue +from archivist.tasks.archive import process_status_queue, start_archive + + +@pytest.mark.usefixtures("db_session") +class TestTasksArchiveBase(unittest.TestCase): + @pytest.fixture(autouse=True) + def _inject(self, db_session, use_settings, local_root, archive_root): + self.session = db_session + self.settings = use_settings + self.local_root = local_root + self.archive_root = archive_root + + def _enqueue_manifest_for(self, source_path, manifest_id="m1"): + request = ManifestRequest( + librarian_name="test-librarian", + store_files=[make_manifest_entry(instance_path=str(source_path))], + ) + item = ArchiveQueue.new_item( + manifest_id=manifest_id, + manifest=request.model_dump_json(), + paths=[str(source_path)], + root=str(self.archive_root), + ) + self.session.add(item) + self.session.commit() + return item + + +class TestStartArchive(TestTasksArchiveBase): + def test_start_archive_noop_when_queue_empty(self): + start_archive(librarian_name="test-librarian") + self.assertEqual(get_status_queue().size, 0) + + def test_start_archive_dequeues_and_enqueues_status(self): + source = self.local_root / "file.txt" + source.write_text("hello") + self._enqueue_manifest_for(source) + + start_archive(librarian_name="test-librarian") + + self.assertEqual(get_status_queue().size, 1) + + item = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() + self.assertTrue(item.consumed) + + def test_start_archive_actually_copies_file(self): + source = self.local_root / "file.txt" + source.write_text("archived-content") + self._enqueue_manifest_for(source) + + start_archive(librarian_name="test-librarian") + + task = get_status_queue().dequeue(block=False) + task.future.result(timeout=5) + + dest = self.archive_root / "file.txt" + self.assertTrue(dest.exists()) + self.assertEqual(dest.read_text(), "archived-content") + + +class TestProcessStatusQueue(TestTasksArchiveBase): + def test_process_status_queue_noop_when_empty(self): + process_status_queue() + + def test_process_status_queue_completes_finished_task(self): + source = self.local_root / "file.txt" + source.write_text("hello") + self._enqueue_manifest_for(source) + + start_archive(librarian_name="test-librarian") + + deadline = time.time() + 5 + while time.time() < deadline: + process_status_queue() + self.session.expire_all() + item = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() + if item.completed: + break + time.sleep(0.05) + + self.assertTrue(item.completed) + self.assertFalse(item.failed) + + def test_process_status_queue_requeues_unfinished_task(self): + class _FakeFuture: + def done(self): + return False + + class _FakeStorageTask: + def __init__(self): + self.future = _FakeFuture() + self._archive = type("A", (), {"manifest_id": "does-not-matter"})() + + status_queue = get_status_queue() + status_queue.enqueue(_FakeStorageTask()) + + process_status_queue() + + self.assertEqual(status_queue.size, 1) + + def test_process_status_queue_swallows_empty(self): + with self.assertRaises(stdlib_queue.Empty): + get_status_queue().dequeue(block=False) + process_status_queue() + + def test_process_status_queue_returns_when_no_matching_db_row(self): + class _FakeFuture: + def done(self): + return True + + class _FakeStorageTask: + def __init__(self): + self.future = _FakeFuture() + self._archive = type("A", (), {"manifest_id": "no-such-manifest"})() + + status_queue = get_status_queue() + status_queue.enqueue(_FakeStorageTask()) + + # Should not raise, even though no ArchiveQueue row matches. + process_status_queue() + + self.assertEqual(status_queue.size, 0) + + def test_process_status_queue_marks_item_failed_when_complete_raises(self): + source = self.local_root / "file.txt" + source.write_text("hello") + self._enqueue_manifest_for(source) + + start_archive(librarian_name="test-librarian") + + deadline = time.time() + 5 + task = get_status_queue().dequeue(block=False) + while time.time() < deadline and not task.future.done(): + time.sleep(0.05) + get_status_queue().enqueue(task) + + with unittest.mock.patch.object(ArchiveQueue, "complete", side_effect=RuntimeError("boom")): + process_status_queue() + + self.session.expire_all() + item = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() + self.assertTrue(item.completed) + self.assertTrue(item.failed) + + def test_process_status_queue_sleeps_and_swallows_unexpected_exception(self): + class _BoomStorageTask: + @property + def future(self): + raise RuntimeError("boom") + + status_queue = get_status_queue() + status_queue.enqueue(_BoomStorageTask()) + + with unittest.mock.patch("archivist.tasks.archive.sleep") as mock_sleep: + process_status_queue() + + mock_sleep.assert_called_once_with(1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..f947c51 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,72 @@ +# Copyright (c) 2025-2026 Simons Observatory. +# Full license can be found in the top level "LICENSE" file. + +"""Tests for archivist.utils._checksum.""" + +import hashlib +import os +import tempfile +import unittest + +from hypothesis import given, settings +from hypothesis import strategies as st + +from archivist.utils import _checksum + + +def _write_tmp_file(data: bytes) -> str: + fd, path = tempfile.mkstemp() + with os.fdopen(fd, "wb") as handle: + handle.write(data) + return path + + +class TestChecksum(unittest.TestCase): + def setUp(self): + self._paths = [] + + def tearDown(self): + for path in self._paths: + os.remove(path) + + def _tmp_file(self, data: bytes) -> str: + path = _write_tmp_file(data) + self._paths.append(path) + return path + + def test_empty_file(self): + path = self._tmp_file(b"") + self.assertEqual(_checksum(path), hashlib.sha256(b"").hexdigest()) + + def test_known_value(self): + path = self._tmp_file(b"hello world") + self.assertEqual(_checksum(path), hashlib.sha256(b"hello world").hexdigest()) + + def test_larger_than_chunk_size(self): + data = b"x" * (1024 * 1024 + 17) + path = self._tmp_file(data) + self.assertEqual(_checksum(path), hashlib.sha256(data).hexdigest()) + + +class TestChecksumProperty(unittest.TestCase): + @given(st.binary(max_size=1 << 16)) + @settings(deadline=None) + def test_matches_hashlib_for_arbitrary_bytes(self, data: bytes): + path = _write_tmp_file(data) + try: + self.assertEqual(_checksum(path), hashlib.sha256(data).hexdigest()) + finally: + os.remove(path) + + @given(st.binary(min_size=1, max_size=4096)) + @settings(deadline=None) + def test_is_deterministic(self, data: bytes): + path = _write_tmp_file(data) + try: + self.assertEqual(_checksum(path), _checksum(path)) + finally: + os.remove(path) + + +if __name__ == "__main__": + unittest.main()