thejayman77
575f562ad5
Reshape the long single-page dashboard into a sectioned console (one route, ?section= tabs, sticky subnav) focused on "what needs my attention" first. * Overview: an "Attention Needed" strip (soft amber/blue, never alarming red) derived from the same data — sources resting/flagged, image coverage <70%, thin brief, recent feedback — plus at-a-glance pulse cards. * Content: corpus health + image/summary coverage (with_image, summaries_with_ image, brief image coverage, 24h image misses) + top opened / topics / tags. * Sources: filterable table (All/Healthy/Resting/Flagged) — served, last success, next poll, failure streak, status — instead of a card pile. * Audience: visitors, retention, accounts, funnel, sharing, daily trend. * Feedback: inbox with category filter, newest first, quick mailto reply. Backend: content_stats gains added_7d + image-coverage fields; source_health gains review_flag; admin_stats adds attention[] + feedback_7d. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
75 lines
3.5 KiB
Python
75 lines
3.5 KiB
Python
from goodnews.db import connect, init_db
|
|
from goodnews import queries, localtime
|
|
|
|
|
|
def test_site_tz_from_env_and_fallback(monkeypatch):
|
|
monkeypatch.setenv("GOODNEWS_TZ", "America/New_York")
|
|
assert localtime.site_tz().key == "America/New_York"
|
|
assert localtime.local_now().utcoffset() is not None # tz-aware
|
|
# A bogus zone must not crash — fall back to UTC.
|
|
monkeypatch.setenv("GOODNEWS_TZ", "Not/AZone")
|
|
assert localtime.site_tz().key == "UTC"
|
|
|
|
|
|
def _seed(c):
|
|
c.execute("INSERT INTO sources (id,name,feed_url,active,default_category) VALUES (1,'Good','http://s/1',1,'science')")
|
|
c.execute("INSERT INTO sources (id,name,feed_url,active,consecutive_failures,poll_interval_minutes) "
|
|
"VALUES (2,'Flaky','http://s/2',1,5,60)")
|
|
# source 1: two accepted (one duplicate → not served), one rejected
|
|
for aid, dup in [(1, None), (2, None), (3, 1)]:
|
|
c.execute("INSERT INTO articles (id,source_id,canonical_url,title,url_hash,duplicate_of,published_at) "
|
|
"VALUES (?,1,?,?,?,?,datetime('now'))", (aid, f"u{aid}", f"T{aid}", f"h{aid}", dup))
|
|
c.execute("INSERT INTO article_scores (article_id,accepted) VALUES (1,1)")
|
|
c.execute("INSERT INTO article_scores (article_id,accepted) VALUES (2,0)")
|
|
c.execute("INSERT INTO article_scores (article_id,accepted) VALUES (3,1)") # duplicate → excluded
|
|
c.commit()
|
|
|
|
|
|
def test_content_stats_counts_served_excluding_duplicates(tmp_path):
|
|
c = connect(str(tmp_path / "t.db")); init_db(c); _seed(c)
|
|
cs = queries.content_stats(c)
|
|
assert cs["served"] == 1 # only article 1 (2 is rejected, 3 is a duplicate)
|
|
assert cs["total"] == 3
|
|
assert cs["rejected"] == 1
|
|
assert cs["active_sources"] == 2
|
|
|
|
|
|
def test_source_health_orders_failing_first_and_computes_next_due(tmp_path):
|
|
c = connect(str(tmp_path / "t.db")); init_db(c); _seed(c)
|
|
c.execute("INSERT INTO ingest_runs (source_id, finished_at, status) "
|
|
"VALUES (2, datetime('now','-30 minutes'), 'failed')")
|
|
c.commit()
|
|
sh = queries.source_health(c)
|
|
assert [s["name"] for s in sh][0] == "Flaky" # failing source floats to top
|
|
flaky = next(s for s in sh if s["name"] == "Flaky")
|
|
assert flaky["failures"] == 5
|
|
# next_due = last_attempt + 60*(1+5)=360 min; attempt was 30m ago → still resting
|
|
assert flaky["next_due_at"] is not None
|
|
good = next(s for s in sh if s["name"] == "Good")
|
|
assert good["served"] == 1 and good["next_due_at"] is None # never attempted → due now
|
|
|
|
|
|
def test_attention_flags_resting_flagged_and_brief():
|
|
from goodnews import queries
|
|
content = {"served": 100, "with_image": 90, "latest_brief_size": 5}
|
|
sources = [
|
|
{"failures": 3, "review_flag": 0},
|
|
{"failures": 0, "review_flag": 1},
|
|
{"failures": 0, "review_flag": 0},
|
|
]
|
|
items = queries._attention(content, sources, feedback_7d=2)
|
|
texts = " | ".join(i["text"] for i in items)
|
|
assert "1 source backing off" in texts # one resting
|
|
assert "1 source flagged" in texts # one flagged
|
|
assert "brief has only 5" in texts # brief < 7
|
|
assert "2 feedback" in texts # recent feedback
|
|
# 90/100 = 90% image coverage → no coverage warning
|
|
assert "Image coverage" not in texts
|
|
|
|
|
|
def test_attention_clear_when_healthy():
|
|
from goodnews import queries
|
|
content = {"served": 100, "with_image": 95, "latest_brief_size": 7}
|
|
sources = [{"failures": 0, "review_flag": 0}]
|
|
assert queries._attention(content, sources, feedback_7d=0) == []
|