Add publisher dashboard routes
All checks were successful
buildbot/nix-eval Build done.
buildbot/nix-build Build done.
buildbot/nix-effects Build done.

This commit is contained in:
Abel Luck 2026-06-02 10:18:59 +02:00
parent 96551c2788
commit e4a5246ab3
31 changed files with 1603 additions and 516 deletions

View file

@ -2,8 +2,9 @@ from __future__ import annotations
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import cast
from repub.jobs import load_runs_view
from repub.jobs import load_dashboard_view, load_runs_view
from repub.model import (
Job,
JobExecution,
@ -232,6 +233,90 @@ def test_load_runs_view_projects_queued_executions_in_fifo_order(
assert view["queued"][1]["move_down_disabled"] is True
def test_load_runs_view_projects_admin_prefixed_action_and_log_paths(
tmp_path: Path,
) -> None:
initialize_database(tmp_path / "jobs-admin-path-prefix.db")
source = create_source(
name="Admin prefixed source",
slug="admin-prefixed-source",
source_type="feed",
notes="",
spider_arguments="",
enabled=True,
cron_minute="*/5",
cron_hour="*",
cron_day_of_month="*",
cron_day_of_week="*",
cron_month="*",
feed_url="https://example.com/admin-prefixed.xml",
)
with database.writer():
job = Job.get(Job.source == source)
running = JobExecution.create(
job=job,
running_status=JobExecutionStatus.RUNNING,
started_at=datetime(2026, 3, 30, 12, 0, tzinfo=UTC),
)
view = load_runs_view(
log_dir=tmp_path / "out" / "logs",
now=datetime(2026, 3, 30, 12, 30, tzinfo=UTC),
path_prefix="/admin",
)
assert (
view["running"][0]["log_href"]
== f"/admin/job/{job.id}/execution/{int(running.get_id())}/logs"
)
assert (
view["running"][0]["cancel_post_path"]
== f"/admin/actions/executions/{int(running.get_id())}/cancel"
)
assert view["upcoming"][0]["run_post_path"] == (
f"/admin/actions/jobs/{job.id}/run-now"
)
assert view["upcoming"][0]["toggle_post_path"] == (
f"/admin/actions/jobs/{job.id}/toggle-enabled"
)
assert view["upcoming"][0]["delete_post_path"] == (
f"/admin/actions/jobs/{job.id}/delete"
)
def test_load_dashboard_view_projects_publisher_run_action_but_root_feed_links(
tmp_path: Path,
) -> None:
initialize_database(tmp_path / "jobs-publisher-dashboard-prefix.db")
source = create_source(
name="Publisher prefixed source",
slug="publisher-prefixed-source",
source_type="feed",
notes="",
spider_arguments="",
enabled=True,
cron_minute="*/5",
cron_hour="*",
cron_day_of_month="*",
cron_day_of_week="*",
cron_month="*",
feed_url="https://example.com/publisher-prefixed.xml",
)
with database.reader():
job = Job.get(Job.source == source)
view = load_dashboard_view(
log_dir=tmp_path / "out" / "logs",
now=datetime(2026, 3, 30, 12, 30, tzinfo=UTC),
path_prefix="/publisher",
)
source_feeds = cast(tuple[dict[str, object], ...], view["source_feeds"])
source_feed = source_feeds[0]
assert source_feed["feed_href"] == "/feeds/publisher-prefixed-source/feed.rss"
assert source_feed["run_post_path"] == (f"/publisher/actions/jobs/{job.id}/run-now")
def test_load_runs_view_keeps_queued_jobs_in_scheduled_jobs(
tmp_path: Path,
) -> None: