create sources in memory

tests/test_web.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import asyncio
+from pathlib import Path
 from typing import Any, cast
 
 from repub.datastar import RefreshBroker, render_sse_event, render_stream
 from repub.web import (
     create_app,
     get_refresh_broker,
+    get_sources_dict,
     render_create_source,
     render_dashboard,
     render_execution_logs,
@@ -38,6 +40,17 @@ def test_root_get_serves_datastar_shim() -> None:
     asyncio.run(run())
 
 
+def test_create_app_bootstraps_default_database_path(
+    monkeypatch, tmp_path: Path
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    app = create_app()
+
+    assert Path(app.config["REPUB_DB_PATH"]) == tmp_path / "republisher.db"
+    assert (tmp_path / "republisher.db").exists()
+
+
 def test_root_get_honors_if_none_match() -> None:
     async def run() -> None:
         client = create_app().test_client()
@@ -161,12 +174,15 @@ def test_render_create_source_shows_dedicated_form_page() -> None:
         assert "Dedicated create page for the source form" in body
         assert "Source and job setup" in body
         assert "data-signals__ifmissing" in body
+        assert "/actions/sources/create" in body
         assert 'data-show="$sourceType === 'feed'"' in body
         assert 'data-show="$sourceType === 'pangea'"' in body
         assert "jobEnabled" in body
         assert "onlyNewest" in body
         assert "includeAuthors" in body
         assert "excludeMedia" in body
+        assert "TEXT_ONLY" in body
+        assert "breakingnews" in body
         assert "Pangea domain" in body
         assert "Feed URL" in body
         assert "Cron schedule" in body
@@ -175,6 +191,87 @@ def test_render_create_source_shows_dedicated_form_page() -> None:
     asyncio.run(run())
 
 
+def test_create_source_action_adds_new_source_to_in_memory_store() -> None:
+    async def run() -> None:
+        app = create_app()
+        client = app.test_client()
+
+        response = await client.post(
+            "/actions/sources/create",
+            headers={"Datastar-Request": "true"},
+            json={
+                "sourceName": "Kenya health desk",
+                "sourceSlug": "kenya-health",
+                "sourceType": "pangea",
+                "pangeaDomain": "example.org",
+                "pangeaCategory": "Health",
+                "contentFormat": "MOBILE_3",
+                "contentType": "breakingnews",
+                "maxArticles": "12",
+                "oldestArticle": "5",
+                "sourceNotes": "Regional health alerts.",
+                "spiderArguments": "language=en",
+                "cronMinute": "0",
+                "cronHour": "*/6",
+                "cronDayOfMonth": "*",
+                "cronDayOfWeek": "*",
+                "cronMonth": "*",
+                "jobEnabled": True,
+                "onlyNewest": True,
+                "includeAuthors": True,
+                "excludeMedia": False,
+            },
+        )
+        body = await response.get_data(as_text=True)
+
+        assert response.status_code == 200
+        assert "window.location = '/sources'" in body
+        assert "kenya-health" in get_sources_dict(app)
+        assert get_sources_dict(app)["kenya-health"]["content_type"] == "breakingnews"
+
+    asyncio.run(run())
+
+
+def test_create_source_action_validates_duplicate_slug_and_pangea_type() -> None:
+    async def run() -> None:
+        app = create_app()
+        client = app.test_client()
+
+        response = await client.post(
+            "/actions/sources/create",
+            headers={"Datastar-Request": "true"},
+            json={
+                "sourceName": "Duplicate guardian",
+                "sourceSlug": "guardian-feed",
+                "sourceType": "pangea",
+                "pangeaDomain": "example.org",
+                "pangeaCategory": "News",
+                "contentFormat": "WEB",
+                "contentType": "not-a-real-type",
+                "maxArticles": "ten",
+                "oldestArticle": "3",
+                "cronMinute": "0",
+                "cronHour": "*",
+                "cronDayOfMonth": "*",
+                "cronDayOfWeek": "*",
+                "cronMonth": "*",
+                "jobEnabled": True,
+            },
+        )
+        body = await response.get_data(as_text=True)
+
+        assert response.status_code == 200
+        assert "Slug must be unique." in body
+        assert "Content format is invalid." in body
+        assert "Content type is invalid." in body
+        assert "Max articles must be an integer." in body
+        assert "Duplicate guardian" not in {
+            str(source["name"]) for source in get_sources_dict(app).values()
+        }
+
+    asyncio.run(run())
+
+
 def test_render_runs_shows_running_upcoming_and_completed_tables() -> None:
     async def run() -> None:
         body = str(await render_runs())