forked from LiveCarta/ContentAutomation
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from content_automation.adapters.publish_store import MongoPublishedContentStore
|
|
from content_automation.factories import build_published_content_store
|
|
from content_automation.models import PublishedContentRecord
|
|
from content_automation.settings import AppSettings
|
|
|
|
|
|
class FakeCollection:
|
|
def __init__(self) -> None:
|
|
self.documents: list[dict[str, object]] = []
|
|
|
|
def insert_one(self, document: dict[str, object]) -> None:
|
|
self.documents.append(document)
|
|
|
|
|
|
class FakeDatabase:
|
|
def __init__(self, collection: FakeCollection) -> None:
|
|
self._collection = collection
|
|
|
|
def __getitem__(self, collection_name: str) -> FakeCollection:
|
|
return self._collection
|
|
|
|
|
|
class FakeMongoClient:
|
|
def __init__(self, connection_uri: str) -> None:
|
|
self.connection_uri = connection_uri
|
|
self.collection = FakeCollection()
|
|
|
|
def __getitem__(self, database_name: str) -> FakeDatabase:
|
|
return FakeDatabase(self.collection)
|
|
|
|
|
|
def test_mongo_published_content_store_saves_record(monkeypatch) -> None:
|
|
fake_client = FakeMongoClient("mongodb://example:27017")
|
|
|
|
monkeypatch.setattr(
|
|
"content_automation.adapters.publish_store.mongodb.MongoClient",
|
|
lambda connection_uri: fake_client,
|
|
)
|
|
|
|
store = MongoPublishedContentStore(
|
|
connection_uri="mongodb://example:27017",
|
|
database_name="content_automation",
|
|
collection_name="published_content",
|
|
)
|
|
|
|
store.save(
|
|
PublishedContentRecord(
|
|
relative_path="media/video.mp4",
|
|
media_url="https://cdn.example.com/media/video.mp4",
|
|
caption="caption",
|
|
platform_responses={
|
|
"instagram": {"id": "ig-1"},
|
|
"youtube": "yt-1",
|
|
},
|
|
)
|
|
)
|
|
|
|
assert len(fake_client.collection.documents) == 1
|
|
assert fake_client.collection.documents[0]["relative_path"] == "media/video.mp4"
|
|
assert fake_client.collection.documents[0]["platform_responses"] == {
|
|
"instagram": {"id": "ig-1"},
|
|
"youtube": "yt-1",
|
|
}
|
|
|
|
|
|
def test_build_published_content_store_returns_none_when_disabled() -> None:
|
|
settings = AppSettings.model_validate({"mongodb": {"enabled": False}})
|
|
|
|
store = build_published_content_store(settings)
|
|
|
|
assert store is None
|
|
|
|
|
|
def test_build_published_content_store_requires_connection_uri() -> None:
|
|
settings = AppSettings.model_validate({"mongodb": {"enabled": True}})
|
|
|
|
with pytest.raises(ValueError):
|
|
build_published_content_store(settings)
|
|
|
|
|
|
def test_build_published_content_store_returns_mongo_store(monkeypatch) -> None:
|
|
class FakeMongoStore:
|
|
def __init__(
|
|
self,
|
|
connection_uri: str,
|
|
database_name: str,
|
|
collection_name: str,
|
|
) -> None:
|
|
self.connection_uri = connection_uri
|
|
self.database_name = database_name
|
|
self.collection_name = collection_name
|
|
|
|
monkeypatch.setattr("content_automation.factories.MongoPublishedContentStore", FakeMongoStore)
|
|
|
|
settings = AppSettings.model_validate(
|
|
{
|
|
"mongodb": {
|
|
"enabled": True,
|
|
"connection_uri": "mongodb://localhost:27017",
|
|
"database_name": "content_automation",
|
|
"collection_name": "published_content",
|
|
}
|
|
}
|
|
)
|
|
|
|
store = build_published_content_store(settings)
|
|
|
|
assert isinstance(store, FakeMongoStore)
|
|
assert store.connection_uri == "mongodb://localhost:27017"
|
|
assert store.database_name == "content_automation"
|
|
assert store.collection_name == "published_content"
|