Include Atlos tests, metadata fixture.

src/auto_archiver/modules/atlos_db/__init__.py

@@ -1 +1 @@
-from atlos_db import AtlosDb
+from .atlos_db import AtlosDb

tests/conftest.py

@@ -155,7 +155,5 @@ def mock_sleep(mocker):
 def metadata():
     metadata = Metadata()
     metadata.set("_processed_at", "2021-01-01T00:00:00")
-    metadata.set_title("Example Title")
-    metadata.set_content("Example Content")
     metadata.set_url("https://example.com")
     return metadata

tests/databases/test_api_db.py

@@ -19,14 +19,6 @@ def api_db(setup_module):
     return setup_module(AAApiDb, configs)
 
 
-@pytest.fixture
-def metadata():
-    metadata = Metadata()
-    metadata.set("_processed_at", "2021-01-01T00:00:00")
-    metadata.set_url("https://example.com")
-    return metadata
-
-
 def test_fetch_no_cache(api_db, metadata):
     # Test fetch
     api_db.use_api_cache = False

tests/databases/test_atlos_db.py

@@ -0,0 +1,110 @@
+import pytest
+from datetime import datetime
+
+from auto_archiver.core import Metadata
+from auto_archiver.modules.atlos_db import AtlosDb
+
+
+class FakeAPIResponse:
+    """Simulate a response object."""
+
+    def __init__(self, data: dict, raise_error: bool = False) -> None:
+        self._data = data
+        self.raise_error = raise_error
+
+    def raise_for_status(self) -> None:
+        if self.raise_error:
+            raise Exception("HTTP error")
+
+
+@pytest.fixture
+def atlos_db(setup_module) -> AtlosDb:
+    """Fixture for AtlosDb."""
+    configs: dict = {
+        "api_token": "abc123",
+        "atlos_url": "https://platform.atlos.org",
+    }
+    return setup_module("atlos_db", configs)
+
+
+def test_failed_no_atlos_id(atlos_db, metadata, mocker):
+    """Test failed() skips posting when no atlos_id present."""
+    post_mock = mocker.patch("requests.post")
+    atlos_db.failed(metadata, "failure reason")
+    post_mock.assert_not_called()
+
+
+def test_failed_with_atlos_id(atlos_db, metadata, mocker):
+    """Test failed() posts failure when atlos_id is present."""
+    metadata.set("atlos_id", 42)
+    fake_resp = FakeAPIResponse({}, raise_error=False)
+    post_mock = mocker.patch("requests.post", return_value=fake_resp)
+    atlos_db.failed(metadata, "failure reason")
+    expected_url = (
+        f"{atlos_db.atlos_url}/api/v2/source_material/metadata/42/auto_archiver"
+    )
+    expected_headers = {"Authorization": f"Bearer {atlos_db.api_token}"}
+    expected_json = {
+        "metadata": {"processed": True, "status": "error", "error": "failure reason"}
+    }
+    post_mock.assert_called_once_with(
+        expected_url, headers=expected_headers, json=expected_json
+    )
+
+
+def test_failed_http_error(atlos_db, metadata, mocker):
+    """Test failed() raises exception on HTTP error."""
+    metadata.set("atlos_id", 42)
+    fake_resp = FakeAPIResponse({}, raise_error=True)
+    mocker.patch("requests.post", return_value=fake_resp)
+    with pytest.raises(Exception, match="HTTP error"):
+        atlos_db.failed(metadata, "failure reason")
+
+
+def test_fetch_returns_false(atlos_db):
+    """Test fetch() always returns False."""
+    item = Metadata()
+    assert atlos_db.fetch(item) is False
+
+
+def test_done_no_atlos_id(atlos_db, mocker):
+    """Test done() skips posting when no atlos_id present."""
+    item = Metadata().set_url("http://example.com")
+    post_mock = mocker.patch("requests.post")
+    atlos_db.done(item)
+    post_mock.assert_not_called()
+
+
+def test_done_with_atlos_id(atlos_db, metadata, mocker):
+    """Test done() posts success when atlos_id is present."""
+    metadata.set("atlos_id", 99)
+    now = datetime.now()
+    metadata.set("timestamp", now)
+    fake_resp = FakeAPIResponse({}, raise_error=False)
+    post_mock = mocker.patch("requests.post", return_value=fake_resp)
+    atlos_db.done(metadata)
+    expected_url = (
+        f"{atlos_db.atlos_url}/api/v2/source_material/metadata/99/auto_archiver"
+    )
+    expected_headers = {"Authorization": f"Bearer {atlos_db.api_token}"}
+    expected_results = metadata.metadata.copy()
+    expected_results["timestamp"] = now.isoformat()
+    expected_json = {
+        "metadata": {
+            "processed": True,
+            "status": "success",
+            "results": expected_results,
+        }
+    }
+    post_mock.assert_called_once_with(
+        expected_url, headers=expected_headers, json=expected_json
+    )
+
+
+def test_done_http_error(atlos_db, metadata, mocker):
+    """Test done() raises exception on HTTP error."""
+    metadata.set("atlos_id", 123)
+    fake_resp = FakeAPIResponse({}, raise_error=True)
+    mocker.patch("requests.post", return_value=fake_resp)
+    with pytest.raises(Exception, match="HTTP error"):
+        atlos_db.done(metadata)

tests/enrichers/test_meta_enricher.py

@@ -23,14 +23,6 @@ def mock_media(mocker):
     mock.filename = "mock_file.txt"
     return mock
 
-@pytest.fixture
-def metadata():
-    m = Metadata()
-    m.set_url("https://example.com")
-    m.set_title("Test Title")
-    m.set_content("Test Content")
-    return m
-
 
 @pytest.fixture(autouse=True)
 def meta_enricher(setup_module):

tests/feeders/test_atlos_feeder.py

@@ -0,0 +1,108 @@
+import pytest
+from auto_archiver.modules.atlos_feeder import AtlosFeeder
+
+
+class FakeAPIResponse:
+    """Simulate a response object."""
+
+    def __init__(self, data: dict, raise_error: bool = False) -> None:
+        self._data = data
+        self.raise_error = raise_error
+
+    def json(self) -> dict:
+        return self._data
+
+    def raise_for_status(self) -> None:
+        if self.raise_error:
+            raise Exception("HTTP error")
+
+
+@pytest.fixture
+def atlos_feeder(setup_module) -> AtlosFeeder:
+    """Fixture for AtlosFeeder."""
+    configs: dict = {
+        "api_token": "abc123",
+        "atlos_url": "https://platform.atlos.org",
+    }
+    return setup_module("atlos_feeder", configs)
+
+
+@pytest.fixture
+def mock_atlos_api(mocker):
+    """Fixture to mock requests to Atlos API."""
+    def _mock_responses(responses):
+        mocker.patch(
+            "requests.get",
+            side_effect=[FakeAPIResponse(data) for data in responses],
+        )
+    return _mock_responses
+
+
+def test_atlos_feeder_iter_yields_valid_metadata(atlos_feeder, mock_atlos_api):
+    """Test valid items are yielded and invalid ones ignored."""
+    mock_atlos_api([
+        {
+            "next": None,
+            "results": [
+                {"source_url": "http://example.com", "id": 1,
+                 "metadata": {"auto_archiver": {"processed": False}},
+                 "visibility": "visible", "status": "complete"},
+                {"source_url": "", "id": 2,
+                 "metadata": {"auto_archiver": {"processed": False}},
+                 "visibility": "visible", "status": "complete"},
+                {"source_url": "http://example.org", "id": 3,
+                 "metadata": {"auto_archiver": {"processed": True}},
+                 "visibility": "visible", "status": "complete"},
+            ],
+        }
+    ])
+
+    items = list(atlos_feeder)
+    assert len(items) == 1
+    assert items[0].get_url() == "http://example.com"
+    assert items[0].get("atlos_id") == 1
+
+
+def test_atlos_feeder_multiple_pages(atlos_feeder, mock_atlos_api):
+    """Test iteration over multiple pages with valid items."""
+    mock_atlos_api([
+        {
+            "next": "cursor2",
+            "results": [
+                {"source_url": "http://example1.com", "id": 10,
+                 "metadata": {"auto_archiver": {"processed": False}},
+                 "visibility": "visible", "status": "complete"},
+            ],
+        },
+        {
+            "next": None,
+            "results": [
+                {"source_url": "http://example2.com", "id": 20,
+                 "metadata": {"auto_archiver": {"processed": False}},
+                 "visibility": "visible", "status": "complete"},
+            ],
+        },
+    ])
+
+    items = list(atlos_feeder)
+    assert len(items) == 2
+    assert items[0].get_url() == "http://example1.com"
+    assert items[0].get("atlos_id") == 10
+    assert items[1].get_url() == "http://example2.com"
+    assert items[1].get("atlos_id") == 20
+
+
+def test_atlos_feeder_no_results(atlos_feeder, mock_atlos_api):
+    """Test iteration stops when no results are returned."""
+    mock_atlos_api([{"next": None, "results": []}])
+    assert list(atlos_feeder) == []
+
+
+def test_atlos_feeder_http_error(atlos_feeder, mocker):
+    """Test raises an exception on HTTP error."""
+    mocker.patch(
+        "requests.get",
+        return_value=FakeAPIResponse({"next": None, "results": []}, raise_error=True),
+    )
+    with pytest.raises(Exception, match="HTTP error"):
+        list(atlos_feeder)

tests/storages/test_atlos_storage.py

@@ -0,0 +1,142 @@
+import os
+import hashlib
+import pytest
+from auto_archiver.core import Media, Metadata
+from auto_archiver.modules.atlos_storage import AtlosStorage
+
+
+class FakeAPIResponse:
+    """Simulate a response object."""
+
+    def __init__(self, data: dict, raise_error: bool = False) -> None:
+        self._data = data
+        self.raise_error = raise_error
+
+    def json(self) -> dict:
+        return self._data
+
+    def raise_for_status(self) -> None:
+        if self.raise_error:
+            raise Exception("HTTP error")
+
+
+@pytest.fixture
+def atlos_storage(setup_module) -> AtlosStorage:
+    """Fixture for AtlosStorage."""
+    configs: dict = {
+        "api_token": "abc123",
+        "atlos_url": "https://platform.atlos.org",
+    }
+    return setup_module("atlos_storage", configs)
+
+
+@pytest.fixture
+def media(tmp_path) -> Media:
+    """Fixture for Media."""
+    content = b"media content"
+    file_path = tmp_path / "media.txt"
+    file_path.write_bytes(content)
+    media = Media(filename=str(file_path))
+    media.properties = {"something": "Title"}
+    media.key = "key"
+    return media
+
+
+def test_get_cdn_url(atlos_storage: AtlosStorage) -> None:
+    """Test get_cdn_url returns the configured atlos_url."""
+    media = Media(filename="dummy.mp4")
+    url = atlos_storage.get_cdn_url(media)
+    assert url == atlos_storage.atlos_url
+
+
+def test_hash(tmp_path, atlos_storage: AtlosStorage) -> None:
+    """Test _hash() computes the correct SHA-256 hash of a file."""
+    content = b"hello world"
+    file_path = tmp_path / "test.txt"
+    file_path.write_bytes(content)
+    media = Media(filename="dummy.mp4")
+    media.filename = str(file_path)
+    expected_hash = hashlib.sha256(content).hexdigest()
+    assert atlos_storage._hash(media) == expected_hash
+
+
+def test_upload_no_atlos_id(tmp_path, atlos_storage: AtlosStorage, media: Media, mocker) -> None:
+    """Test upload() returns False when metadata lacks atlos_id."""
+    metadata = Metadata()  # atlos_id not set
+    post_mock = mocker.patch("requests.post")
+    result = atlos_storage.upload(media, metadata)
+    assert result is False
+    post_mock.assert_not_called()
+
+
+def test_upload_already_uploaded(atlos_storage: AtlosStorage,
+                                 metadata: Metadata,
+                                 media: Media,
+                                 tmp_path,
+                                 mocker) -> None:
+    """Test upload() returns True if media hash already exists."""
+    content = b"media content"
+    metadata.set("atlos_id", 101)
+    media_hash = hashlib.sha256(content).hexdigest()
+    fake_get = FakeAPIResponse({
+        "result": {"artifacts": [{"file_hash_sha256": media_hash}]}
+    })
+    get_mock = mocker.patch("requests.get", return_value=fake_get)
+    post_mock = mocker.patch("requests.post")
+    result = atlos_storage.upload(media, metadata)
+    assert result is True
+    get_mock.assert_called_once()
+    post_mock.assert_not_called()
+
+
+def test_upload_not_uploaded(tmp_path, atlos_storage: AtlosStorage,
+                             metadata: Metadata,
+                             media: Media,
+                             mocker) -> None:
+    """Test upload() uploads media when not already present."""
+    metadata.set("atlos_id", 202)
+    fake_get = FakeAPIResponse({
+        "result": {"artifacts": [{"file_hash_sha256": "different_hash"}]}
+    })
+    get_mock = mocker.patch("requests.get", return_value=fake_get)
+    fake_post = FakeAPIResponse({}, raise_error=False)
+    post_mock = mocker.patch("requests.post", return_value=fake_post)
+    result = atlos_storage.upload(media, metadata)
+    assert result is True
+    get_mock.assert_called_once()
+    post_mock.assert_called_once()
+    expected_url = f"{atlos_storage.atlos_url}/api/v2/source_material/upload/202"
+    expected_headers = {"Authorization": f"Bearer {atlos_storage.api_token}"}
+    expected_params = {"title": media.properties}
+    call_kwargs = post_mock.call_args.kwargs
+    assert call_kwargs["headers"] == expected_headers
+    assert call_kwargs["params"] == expected_params
+    # Verify the URL passed to requests.post.
+    posted_url = call_kwargs.get("url") or post_mock.call_args.args[0]
+    assert posted_url == expected_url
+    # Verify files parameter contains the correct filename.
+    file_tuple = call_kwargs["files"]["file"]
+    assert file_tuple[0] == os.path.basename(media.filename)
+
+
+def test_upload_post_http_error(tmp_path,
+                                atlos_storage: AtlosStorage,
+                                metadata: Metadata,
+                                media: Media,
+                                mocker) -> None:
+    """Test upload() propagates HTTP error during POST."""
+    metadata.set("atlos_id", 303)
+    fake_get = FakeAPIResponse({
+        "result": {"artifacts": []}
+    })
+    mocker.patch("requests.get", return_value=fake_get)
+    fake_post = FakeAPIResponse({}, raise_error=True)
+    mocker.patch("requests.post", return_value=fake_post)
+    with pytest.raises(Exception, match="HTTP error"):
+        atlos_storage.upload(media, metadata)
+
+
+def test_uploadf_not_implemented(atlos_storage: AtlosStorage) -> None:
+    """Test uploadf() returns None (not implemented)."""
+    result = atlos_storage.uploadf(None, "dummy")
+    assert result is None