bellingcat/auto-archiver
1
0
Fork 0
mirror of https://github.com/bellingcat/auto-archiver.git synced 2026-06-08 03:18:28 +03:00
Code Issues Packages Projects Releases Wiki Activity

Update Atlos tests

Browse Source
This commit is contained in:
erinhmclark
2025-03-05 21:24:38 +00:00
parent 0f911543cd
commit b9c2f98f46
14 changed files with 66 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
tests/databases/test_atlos_db.py
View File

@@ -2,7 +2,7 @@ import pytest
from datetime import datetime
from auto_archiver.core import Metadata
from auto_archiver.modules.atlos_db import AtlosDb
from auto_archiver.modules.atlos_feeder_db_storage import AtlosFeederDbStorage as AtlosDb
class FakeAPIResponse:
@@ -12,19 +12,28 @@ class FakeAPIResponse:
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_db(setup_module) -> AtlosDb:
def atlos_db(setup_module, mocker) -> AtlosDb:
"""Fixture for AtlosDb."""
configs: dict = {
"api_token": "abc123",
"atlos_url": "https://platform.atlos.org",
}
return setup_module("atlos_db", configs)
mocker.patch("requests.Session")
atlos_feeder = setup_module("atlos_feeder_db_storage", configs)
fake_session = mocker.MagicMock()
# Configure the default response to have no results so that __iter__ terminates
fake_session.get.return_value = FakeAPIResponse({"next": None, "results": []})
atlos_feeder.session = fake_session
return atlos_feeder
def test_failed_no_atlos_id(atlos_db, metadata, mocker):
@@ -38,25 +47,20 @@ 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)
post_mock = mocker.patch.object(atlos_db, "_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_endpoint = f"/api/v2/source_material/metadata/42/auto_archiver"
expected_json = {
"metadata": {"processed": True, "status": "error", "error": "failure reason"}
}
post_mock.assert_called_once_with(
expected_url, headers=expected_headers, json=expected_json
)
post_mock.assert_called_once_with(expected_endpoint, 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)
# Patch _post to raise an exception instead of returning a fake response.
mocker.patch.object(atlos_db, "_post", side_effect=Exception("HTTP error"))
with pytest.raises(Exception, match="HTTP error"):
atlos_db.failed(metadata, "failure reason")
@@ -81,12 +85,9 @@ def test_done_with_atlos_id(atlos_db, metadata, mocker):
now = datetime.now()
metadata.set("timestamp", now)
fake_resp = FakeAPIResponse({}, raise_error=False)
post_mock = mocker.patch("requests.post", return_value=fake_resp)
post_mock = mocker.patch.object(atlos_db, "_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_endpoint = f"/api/v2/source_material/metadata/99/auto_archiver"
expected_results = metadata.metadata.copy()
expected_results["timestamp"] = now.isoformat()
expected_json = {
@@ -96,15 +97,13 @@ def test_done_with_atlos_id(atlos_db, metadata, mocker):
"results": expected_results,
}
}
post_mock.assert_called_once_with(
expected_url, headers=expected_headers, json=expected_json
)
post_mock.assert_called_once_with(expected_endpoint, json=expected_json)
def test_done_http_error(atlos_db, metadata, mocker):
"""Test done() raises exception on HTTP error."""
"""Test done() raises an exception on HTTP error."""
metadata.set("atlos_id", 123)
fake_resp = FakeAPIResponse({}, raise_error=True)
mocker.patch("requests.post", return_value=fake_resp)
# Patch _post to raise an exception.
mocker.patch.object(atlos_db, "_post", side_effect=Exception("HTTP error"))
with pytest.raises(Exception, match="HTTP error"):
atlos_db.done(metadata)

27
tests/feeders/test_atlos_feeder.py
View File

@@ -1,5 +1,5 @@
import pytest
from auto_archiver.modules.atlos_feeder import AtlosFeeder
from auto_archiver.modules.atlos_feeder_db_storage import AtlosFeederDbStorage as AtlosFeeder
class FakeAPIResponse:
@@ -18,23 +18,26 @@ class FakeAPIResponse:
@pytest.fixture
def atlos_feeder(setup_module) -> AtlosFeeder:
def atlos_feeder(setup_module, mocker) -> AtlosFeeder:
"""Fixture for AtlosFeeder."""
configs: dict = {
"api_token": "abc123",
"atlos_url": "https://platform.atlos.org",
}
return setup_module("atlos_feeder", configs)
mocker.patch("requests.Session")
atlos_feeder = setup_module("atlos_feeder_db_storage", configs)
fake_session = mocker.MagicMock()
# Configure the default response to have no results so that __iter__ terminates
fake_session.get.return_value = FakeAPIResponse({"next": None, "results": []})
atlos_feeder.session = fake_session
return atlos_feeder
@pytest.fixture
def mock_atlos_api(mocker):
"""Fixture to mock requests to Atlos API."""
def mock_atlos_api(atlos_feeder):
"""Fixture to update the atlos_feeder.session.get side_effect."""
def _mock_responses(responses):
mocker.patch(
"requests.get",
side_effect=[FakeAPIResponse(data) for data in responses],
)
atlos_feeder.session.get.side_effect = [FakeAPIResponse(data) for data in responses]
return _mock_responses
@@ -100,9 +103,7 @@ def test_atlos_feeder_no_results(atlos_feeder, mock_atlos_api):
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),
)
fake_response = FakeAPIResponse({"next": None, "results": []}, raise_error=True)
atlos_feeder.session.get.side_effect = [fake_response]
with pytest.raises(Exception, match="HTTP error"):
list(atlos_feeder)

84
tests/storages/test_atlos_storage.py
View File

@@ -2,7 +2,7 @@ import os
import hashlib
import pytest
from auto_archiver.core import Media, Metadata
from auto_archiver.modules.atlos_storage import AtlosStorage
from auto_archiver.modules.atlos_feeder_db_storage import AtlosFeederDbStorage as AtlosStorage
class FakeAPIResponse:
@@ -21,13 +21,19 @@ class FakeAPIResponse:
@pytest.fixture
def atlos_storage(setup_module) -> AtlosStorage:
def atlos_storage(setup_module, mocker) -> AtlosStorage:
"""Fixture for AtlosStorage."""
configs: dict = {
"api_token": "abc123",
"atlos_url": "https://platform.atlos.org",
}
return setup_module("atlos_storage", configs)
mocker.patch("requests.Session")
atlos_feeder = setup_module("atlos_feeder_db_storage", configs)
mock_session = mocker.MagicMock()
# Configure the default response to have no results so that __iter__ terminates
mock_session.get.return_value = FakeAPIResponse({"next": None, "results": []})
atlos_feeder.session = mock_session
return atlos_feeder
@pytest.fixture
@@ -49,17 +55,6 @@ def test_get_cdn_url(atlos_storage: AtlosStorage) -> None:
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
@@ -69,74 +64,49 @@ def test_upload_no_atlos_id(tmp_path, atlos_storage: AtlosStorage, media: Media,
post_mock.assert_not_called()
def test_upload_already_uploaded(atlos_storage: AtlosStorage,
metadata: Metadata,
media: Media,
tmp_path,
mocker) -> None:
def test_upload_already_uploaded(atlos_storage: AtlosStorage, metadata: Metadata, media: Media, 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")
fake_get_response = {"result": {"artifacts": [{"file_hash_sha256": media_hash}]}}
get_mock = mocker.patch.object(atlos_storage, "_get", return_value=fake_get_response)
post_mock = mocker.patch.object(atlos_storage, "_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:
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)
fake_get_response = {"result": {"artifacts": [{"file_hash_sha256": "different_hash"}]}}
get_mock = mocker.patch.object(atlos_storage, "_get", return_value=fake_get_response)
fake_post_response = {"result": "uploaded"}
post_mock = mocker.patch.object(atlos_storage, "_post", return_value=fake_post_response)
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_endpoint = f"/api/v2/source_material/upload/202"
call_args = post_mock.call_args[0]
assert call_args[0] == expected_endpoint
call_kwargs = post_mock.call_args[1]
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:
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)
fake_get_response = {"result": {"artifacts": []}}
mocker.patch.object(atlos_storage, "_get", return_value=fake_get_response)
mocker.patch.object(atlos_storage, "_post", side_effect=Exception("HTTP error"))
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