Compare commits

...

4 Commits

Author SHA1 Message Date
0aa9b1f561 Updte
Some checks failed
DevPlace CI / test (push) Has been cancelled
2026-06-05 20:34:03 +02:00
a98ded68e4 Update 2026-06-05 20:33:35 +02:00
8d4a82965d iUpdate 2026-06-05 20:05:07 +02:00
9e836ea152 Update 2026-06-05 19:32:46 +02:00
29 changed files with 1317 additions and 42 deletions

View File

@ -21,9 +21,26 @@ jobs:
pip install -e ".[dev]"
python -m playwright install chromium --with-deps
- name: Run integration tests
- name: Run integration tests with coverage
env:
COVERAGE_PROCESS_START: ${{ github.workspace }}/.coveragerc
PLAYWRIGHT_HEADLESS: "1"
run: |
python -m pytest tests/ -v --tb=line -x
python -m coverage run -m pytest tests/ -p no:xdist --tb=line
- name: Build coverage report
if: always()
run: |
python -m coverage combine
python -m coverage report
python -m coverage html
- name: Publish coverage HTML
if: always()
uses: actions/upload-artifact@v4
with:
name: coverage-html
path: htmlcov/
- name: Upload test screenshots
if: failure()

5
.gitignore vendored
View File

@ -15,3 +15,8 @@ devplacepy/static/uploads/*.jpg
devplacepy/static/uploads/*.jpeg
devplacepy/static/uploads/*.gif
devplacepy/static/uploads/*.webp
# coverage
.coverage
.coverage.*
htmlcov/

View File

@ -7,7 +7,7 @@ LOCUST_SPAWN_RATE ?= 5
LOCUST_RUN_TIME ?= 120s
DEVPLACE_RATE_LIMIT ?= 1000000
.PHONY: install dev clean test test-headed locust locust-headless
.PHONY: install dev clean test test-headed coverage coverage-headed coverage-html locust locust-headless
install:
pip install -e .
@ -24,6 +24,24 @@ test:
test-headed:
PLAYWRIGHT_HEADLESS=0 python -m pytest tests/ -v --tb=line -x
coverage:
rm -f .coverage .coverage.*
COVERAGE_PROCESS_START=$(CURDIR)/.coveragerc PLAYWRIGHT_HEADLESS=1 \
python -m coverage run -m pytest tests/ -p no:xdist --tb=line
python -m coverage combine
python -m coverage report
coverage-headed:
rm -f .coverage .coverage.*
COVERAGE_PROCESS_START=$(CURDIR)/.coveragerc PLAYWRIGHT_HEADLESS=0 \
python -m coverage run -m pytest tests/ -p no:xdist --tb=line
python -m coverage combine
python -m coverage report
coverage-html: coverage
python -m coverage html
@echo "Report written to htmlcov/index.html"
locust:
export DEVPLACE_DATABASE_URL="sqlite:///$(LOCUST_DB)"; \
export DEVPLACE_RATE_LIMIT=$(DEVPLACE_RATE_LIMIT); \

View File

@ -22,6 +22,23 @@ def is_owner(item: dict | None, user: dict | None) -> bool:
return bool(item and user and item["user_uid"] == user["uid"])
def canonical_redirect(area: str, item: dict, requested: str) -> RedirectResponse | None:
canonical = item.get("slug") or item["uid"]
if requested == canonical:
return None
return RedirectResponse(url=f"/{area}/{canonical}", status_code=301)
def first_image_url(item: dict, attachments: list | None) -> str | None:
inline = item.get("image")
if inline:
return f"/static/uploads/{inline}"
for attachment in attachments or []:
if attachment.get("is_image"):
return attachment["url"]
return None
def create_content_item(table_name: str, target_type: str, user: dict, fields: dict, slug_source: str, xp: int, badge: str, mention_text: str, attachment_uids: list | None) -> tuple[str, str]:
uid = generate_uid()
slug = make_combined_slug(slug_source, uid)
@ -65,6 +82,7 @@ def edit_content_item(table_name: str, user: dict, slug: str, update_fields: dic
item = resolve_by_slug(table, slug)
if not is_owner(item, user):
return RedirectResponse(url=redirect_fail, status_code=302)
update_fields = {**update_fields, "updated_at": datetime.now(timezone.utc).isoformat()}
table.update({"uid": item["uid"], **update_fields}, ["uid"])
logger.info(f"{table_name} {item['uid']} edited by {user['username']}")
return RedirectResponse(url=f"/{table_name}/{item['slug'] or item['uid']}", status_code=302)

View File

@ -6,7 +6,7 @@ from devplacepy.attachments import get_attachments_batch
from devplacepy.content import enrich_items
from devplacepy.templating import templates
from devplacepy.utils import get_current_user
from devplacepy.seo import list_page_seo
from devplacepy.seo import list_page_seo, next_page_url
logger = logging.getLogger(__name__)
router = APIRouter()
@ -59,6 +59,7 @@ async def feed_page(request: Request, tab: str = "all", topic: str = None, befor
title="Feed",
description="Discover the latest developer discussions, projects, and community activity on DevPlace.",
breadcrumbs=[{"name": "Home", "url": "/feed"}, {"name": "Feed", "url": "/feed"}],
next_url=next_page_url(request, next_cursor),
)
return templates.TemplateResponse(request, "feed.html", {
**seo_ctx,

View File

@ -4,10 +4,10 @@ from fastapi import APIRouter, Request, Form
from devplacepy.models import GistForm, GistEditForm
from fastapi.responses import HTMLResponse, RedirectResponse
from devplacepy.database import get_table, get_users_by_uids, get_gist_languages, paginate
from devplacepy.content import load_detail, edit_content_item, delete_content_item, enrich_items, create_content_item, detail_context
from devplacepy.content import load_detail, edit_content_item, delete_content_item, enrich_items, create_content_item, detail_context, canonical_redirect, first_image_url
from devplacepy.templating import templates
from devplacepy.utils import get_current_user, require_user, not_found, XP_GIST
from devplacepy.seo import base_seo_context, site_url, website_schema, software_source_code_schema, list_page_seo
from devplacepy.seo import base_seo_context, site_url, website_schema, software_source_code_schema, list_page_seo, next_page_url
logger = logging.getLogger(__name__)
router = APIRouter()
@ -53,6 +53,7 @@ async def gists_page(request: Request, language: str = None, user_uid: str = Non
{"name": "Home", "url": "/feed"},
{"name": "Gists", "url": "/gists"},
],
next_url=next_page_url(request, next_cursor),
)
return templates.TemplateResponse(request, "gists.html", {
**seo_ctx,
@ -74,6 +75,9 @@ async def gist_detail(request: Request, gist_slug: str):
if not detail:
raise not_found("Gist not found")
gist = detail["item"]
redirect = canonical_redirect("gists", gist, gist_slug)
if redirect:
return redirect
base = site_url(request)
seo_ctx = base_seo_context(
@ -81,6 +85,7 @@ async def gist_detail(request: Request, gist_slug: str):
title=gist.get("title", "Gist"),
description=gist.get("description", "")[:160],
og_type="article",
og_image=first_image_url(gist, detail["attachments"]),
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Gists", "url": "/gists"},

View File

@ -5,7 +5,8 @@ from fastapi.responses import HTMLResponse
from devplacepy.database import get_table, db, load_comments, resolve_by_slug, get_news_images_by_uids, paginate
from devplacepy.templating import templates
from devplacepy.utils import get_current_user, time_ago, not_found
from devplacepy.seo import base_seo_context, website_schema, site_url, news_article_schema, list_page_seo
from devplacepy.content import canonical_redirect
from devplacepy.seo import base_seo_context, website_schema, site_url, news_article_schema, list_page_seo, next_page_url
logger = logging.getLogger(__name__)
router = APIRouter()
@ -45,6 +46,7 @@ async def news_page(request: Request, before: str = None):
title="Developer News",
description="Curated developer news and industry signals. Stay ahead with hand-picked articles.",
breadcrumbs=[{"name": "Home", "url": "/feed"}, {"name": "News", "url": "/news"}],
next_url=next_page_url(request, next_cursor),
)
return templates.TemplateResponse(request, "news.html", {
@ -63,6 +65,9 @@ async def news_detail_page(request: Request, news_slug: str):
article = resolve_by_slug(news_table, news_slug)
if not article:
raise not_found("News article not found")
redirect = canonical_redirect("news", article, news_slug)
if redirect:
return redirect
image_url = ""
if "news_images" in db.tables:

View File

@ -6,7 +6,7 @@ from devplacepy.constants import TOPICS
from devplacepy.database import db
from devplacepy.templating import templates
from devplacepy.utils import get_current_user, require_user, time_ago, not_found, XP_POST
from devplacepy.content import load_detail, edit_content_item, delete_content_item, create_content_item, detail_context
from devplacepy.content import load_detail, edit_content_item, delete_content_item, create_content_item, detail_context, canonical_redirect, first_image_url
from devplacepy.seo import base_seo_context, site_url, website_schema, discussion_forum_posting, truncate
from devplacepy.attachments import save_inline_image
from devplacepy.models import PostForm, PostEditForm
@ -49,6 +49,9 @@ async def view_post(request: Request, post_slug: str):
if not detail:
raise not_found("Post not found")
post = detail["item"]
redirect = canonical_redirect("posts", post, post_slug)
if redirect:
return redirect
author = detail["author"]
top_level = detail["comments"]
@ -69,6 +72,7 @@ async def view_post(request: Request, post_slug: str):
{"name": post.get("title") or "Post", "url": f"/posts/{post['slug'] or post['uid']}"},
],
og_type="article",
og_image=first_image_url(post, detail["attachments"]),
schemas=[
website_schema(base),
discussion_forum_posting(post, author, comment_count, detail["star_count"], base),

View File

@ -6,7 +6,8 @@ from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from devplacepy.database import get_table, db, get_user_stars, get_user_rank, get_comment_counts_by_post_uids
from devplacepy.content import enrich_items
from devplacepy.templating import templates
from devplacepy.utils import get_current_user, require_user, require_user_api, time_ago, clear_user_cache
from devplacepy.utils import get_current_user, require_user, require_user_api, time_ago, clear_user_cache, not_found
from devplacepy.avatar import avatar_url
from devplacepy.seo import base_seo_context, site_url, website_schema, profile_page_schema
logger = logging.getLogger(__name__)
@ -35,7 +36,7 @@ async def profile_page(request: Request, username: str, tab: str = "posts"):
users = get_table("users")
profile_user = users.find_one(username=username)
if not profile_user:
return RedirectResponse(url="/feed", status_code=302)
raise not_found("Profile not found")
profile_user["stars"] = get_user_stars(profile_user["uid"])
rank = get_user_rank(profile_user["uid"])
@ -82,6 +83,7 @@ async def profile_page(request: Request, username: str, tab: str = "posts"):
description=desc,
robots=robots,
og_type="profile",
og_image=avatar_url("multiavatar", profile_user["username"], 256),
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": profile_user['username'], "url": f"/profile/{profile_user['username']}"},

View File

@ -5,10 +5,10 @@ from fastapi import APIRouter, Request, Form
from devplacepy.models import ProjectForm
from fastapi.responses import HTMLResponse, RedirectResponse
from devplacepy.database import get_table, get_users_by_uids, get_site_stats, get_user_votes, paginate
from devplacepy.content import load_detail, delete_content_item, create_content_item, detail_context
from devplacepy.content import load_detail, delete_content_item, create_content_item, detail_context, canonical_redirect, first_image_url
from devplacepy.templating import templates
from devplacepy.utils import get_current_user, require_user, not_found, XP_PROJECT
from devplacepy.seo import base_seo_context, site_url, website_schema, software_application_schema, list_page_seo
from devplacepy.seo import base_seo_context, site_url, website_schema, software_application_schema, list_page_seo, next_page_url
logger = logging.getLogger(__name__)
router = APIRouter()
@ -66,6 +66,7 @@ async def projects_page(
{"name": "Home", "url": "/feed"},
{"name": "Projects", "url": "/projects"},
],
next_url=next_page_url(request, next_cursor),
)
return templates.TemplateResponse(request, "projects.html", {
**seo_ctx,
@ -88,12 +89,16 @@ async def project_detail(request: Request, project_slug: str):
if not detail:
raise not_found("Project not found")
project = detail["item"]
redirect = canonical_redirect("projects", project, project_slug)
if redirect:
return redirect
base = site_url(request)
seo_ctx = base_seo_context(
request,
title=project.get("title", "Project"),
description=project.get("description", "")[:160],
og_image=first_image_url(project, detail["attachments"]),
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Projects", "url": "/projects"},

View File

@ -1,5 +1,7 @@
import json
import logging
import time
from urllib.parse import urlencode
from xml.etree.ElementTree import Element, tostring
from xml.dom import minidom
from devplacepy.config import SITE_URL
@ -9,6 +11,9 @@ logger = logging.getLogger(__name__)
SITE_NAME = "DevPlace"
SITEMAP_URL_LIMIT = 5000
SITEMAP_TTL = 3600
_sitemap_cache = {}
def truncate(text, max_len=160):
@ -70,6 +75,7 @@ def discussion_forum_posting(post, author, comment_count, star_count, base_url):
"url": f"{base_url}/profile/{author['username']}" if author else ""
},
"datePublished": post.get("created_at", ""),
"dateModified": post.get("updated_at") or post.get("created_at", ""),
"interactionStatistic": [
{"@type": "InteractionCounter", "interactionType": "https://schema.org/LikeAction", "userInteractionCount": star_count},
{"@type": "InteractionCounter", "interactionType": "https://schema.org/CommentAction", "userInteractionCount": comment_count}
@ -106,6 +112,7 @@ def software_application_schema(project, base_url):
"name": project.get("author_name", "Unknown")
},
"datePublished": project.get("created_at", ""),
"dateModified": project.get("updated_at") or project.get("created_at", ""),
"offers": {
"@type": "Offer",
"price": "0",
@ -131,6 +138,7 @@ def news_article_schema(article, base_url, image_url=""):
"description": truncate(strip_html(article.get("description", "") or ""), 200),
"url": url,
"datePublished": article.get("synced_at", "") or article.get("created_at", ""),
"dateModified": article.get("synced_at", "") or article.get("created_at", ""),
"mainEntityOfPage": {"@type": "WebPage", "@id": url},
"author": {"@type": "Organization", "name": article.get("source_name") or SITE_NAME},
"publisher": organization_schema(base_url),
@ -148,6 +156,7 @@ def software_source_code_schema(gist, base_url):
"url": f"{base_url}/gists/{gist.get('slug') or gist['uid']}",
"programmingLanguage": gist.get("language", "") or "text",
"dateCreated": gist.get("created_at", ""),
"dateModified": gist.get("updated_at") or gist.get("created_at", ""),
}
@ -183,7 +192,13 @@ def combine(schemas):
DEFAULT_OG_IMAGE = "/static/og-default.png"
def base_seo_context(request, title="", description="", robots="index,follow", og_type="website", og_image=None, breadcrumbs=None, schemas=None):
def absolute_url(base, path):
if not path:
return ""
return path if path.startswith("http") else f"{base}{path}"
def base_seo_context(request, title="", description="", robots="index,follow", og_type="website", og_image=None, breadcrumbs=None, schemas=None, prev_url=None, next_url=None):
base = site_url(request)
page_title = f"{title} - {SITE_NAME}" if title else SITE_NAME
canonical = f"{base}{request.url.path}"
@ -191,7 +206,10 @@ def base_seo_context(request, title="", description="", robots="index,follow", o
if page and page not in ("", "1"):
canonical = f"{canonical}?page={page}"
clean_description = truncate(strip_html(description), 160)
og_img = og_image or f"{base}{DEFAULT_OG_IMAGE}"
og_img = absolute_url(base, og_image) or f"{base}{DEFAULT_OG_IMAGE}"
page_schemas = list(schemas or [])
if breadcrumbs:
page_schemas.append(breadcrumb_schema(breadcrumbs, base))
return {
"page_title": page_title,
"meta_description": clean_description,
@ -202,11 +220,21 @@ def base_seo_context(request, title="", description="", robots="index,follow", o
"og_image": og_img,
"og_type": og_type,
"breadcrumbs": breadcrumbs or [],
"page_schema": combine(schemas or []),
"page_schema": combine(page_schemas),
"prev_url": absolute_url(base, prev_url),
"next_url": absolute_url(base, next_url),
}
def list_page_seo(request, title="", description="", breadcrumbs=None):
def next_page_url(request, next_cursor):
if not next_cursor:
return None
params = dict(request.query_params)
params["before"] = next_cursor
return f"{request.url.path}?{urlencode(params)}"
def list_page_seo(request, title="", description="", breadcrumbs=None, prev_url=None, next_url=None):
base = site_url(request)
return base_seo_context(
request,
@ -214,10 +242,28 @@ def list_page_seo(request, title="", description="", breadcrumbs=None):
description=description,
breadcrumbs=breadcrumbs,
schemas=[website_schema(base)],
prev_url=prev_url,
next_url=next_url,
)
def make_sitemap(base_url):
cached = _sitemap_cache.get(base_url)
if cached and time.time() - cached[0] < SITEMAP_TTL:
return cached[1]
xml = _build_sitemap(base_url)
_sitemap_cache[base_url] = (time.time(), xml)
return xml
def _collect(table, limit, label, **query):
rows = list(table.find(_limit=limit, **query))
if len(rows) >= limit:
logger.warning("sitemap: %s truncated at %d entries", label, limit)
return rows
def _build_sitemap(base_url):
from devplacepy.database import get_table, db
def url_element(loc, lastmod=None, changefreq=None, priority=None):
@ -250,7 +296,7 @@ def make_sitemap(base_url):
urlset.append(url_element(f"{base_url}/leaderboard", changefreq="daily", priority="0.7"))
if "posts" in db.tables:
posts = list(get_table("posts").find(order_by=["-created_at"], _limit=500))
posts = _collect(get_table("posts"), SITEMAP_URL_LIMIT, "posts", order_by=["-created_at"])
for p in posts:
urlset.append(url_element(
f"{base_url}/posts/{p.get('slug') or p['uid']}",
@ -260,7 +306,7 @@ def make_sitemap(base_url):
))
if "projects" in db.tables:
projects = list(get_table("projects").find(order_by=["-created_at"], _limit=1000))
projects = _collect(get_table("projects"), SITEMAP_URL_LIMIT, "projects", order_by=["-created_at"])
for p in projects:
urlset.append(url_element(
f"{base_url}/projects/{p.get('slug') or p['uid']}",
@ -270,7 +316,7 @@ def make_sitemap(base_url):
))
if "gists" in db.tables:
gists = list(get_table("gists").find(order_by=["-created_at"], _limit=500))
gists = _collect(get_table("gists"), SITEMAP_URL_LIMIT, "gists", order_by=["-created_at"])
for g in gists:
urlset.append(url_element(
f"{base_url}/gists/{g.get('slug') or g['uid']}",
@ -280,7 +326,7 @@ def make_sitemap(base_url):
))
if "news" in db.tables:
articles = list(get_table("news").find(status="published", order_by=["-synced_at"], _limit=500))
articles = _collect(get_table("news"), SITEMAP_URL_LIMIT, "news", status="published", order_by=["-synced_at"])
for a in articles:
urlset.append(url_element(
f"{base_url}/news/{a.get('slug') or a['uid']}",
@ -294,7 +340,7 @@ def make_sitemap(base_url):
if "posts" in db.tables:
for row in db.query("SELECT user_uid, COUNT(*) AS c FROM posts GROUP BY user_uid"):
post_counts[row["user_uid"]] = row["c"]
users = list(get_table("users").find(order_by=["-created_at"], _limit=200))
users = _collect(get_table("users"), SITEMAP_URL_LIMIT, "users", order_by=["-created_at"])
for u in users:
if post_counts.get(u["uid"], 0) < 2:
continue

View File

@ -207,8 +207,42 @@ class NewsService(BaseService):
content = strip_html(article.get("content", "") or "")[:1500]
prompt = (
"Rate this article's relevance to software developers on a scale of 1-10. "
"Return only a single integer between 1 and 10, nothing else.\n\n"
"You are an expert content strategist and editor for DevPlace, a "
"server-rendered social network for software developers. Decide how "
"strongly a given article should be published, expressed as a single "
"grade.\n\n"
"Site context:\n"
"- Audience: software developers, hobbyists and professionals, "
"interested in software development, AI agents, open-source tools, "
"systems programming, and self-hosting.\n"
"- Mission: deliver deep technical insight, critical analysis, and "
"practical value; build authority and reader trust; avoid low-effort, "
"rehashed, or clickbait content.\n"
"- Tone: technical, practical, nuanced, skeptical of hype; no "
"sensationalism or unverified claims.\n\n"
"Evaluate the article against these weighted criteria and weigh them "
"internally:\n"
"1. Relevance & alignment to the developer niche and audience needs "
"(20%).\n"
"2. Quality & accuracy: depth, originality, factual correctness, "
"sourcing, technical correctness, clear writing (25%).\n"
"3. Engagement & readability: strong title/intro/flow, useful "
"examples, appropriate length (15%).\n"
"4. Originality & uniqueness: new perspective, data, or analysis; not "
"duplicated generic web content (15%).\n"
"5. SEO & discoverability: keyword opportunity without stuffing, "
"shareable (10%).\n"
"6. Ethics, legality & risk: no misinformation, plagiarism, or "
"brand-safety issues; controversial topics handled with nuance and "
"evidence (10%).\n"
"7. Strategic fit for the site (5%).\n\n"
"Convert your overall judgment to a single integer grade from 1 to "
"10:\n"
"- 9-10: excellent, publish as-is.\n"
"- 7-8: good, worth publishing.\n"
"- 5-6: marginal, weak fit or quality.\n"
"- 1-4: reject, low quality or off-topic.\n\n"
"Return ONLY a single integer from 1 to 10, nothing else.\n\n"
f"Title: {title}\n"
f"Description: {description}\n"
f"Content: {content}"

View File

@ -7,6 +7,8 @@
<title>{% if page_title %}{{ page_title }}{% else %}DevPlace - The Developer Social Network{% endif %}</title>
<meta name="description" content="{% if meta_description %}{{ meta_description }}{% else %}Track industry shifts. Discover bold releases. Share what you're building in an open, uncensored environment.{% endif %}">
<link rel="canonical" href="{{ canonical_url or request.url }}">
{% if prev_url %}<link rel="prev" href="{{ prev_url }}">{% endif %}
{% if next_url %}<link rel="next" href="{{ next_url }}">{% endif %}
<meta name="robots" content="{{ meta_robots or 'index,follow' }}">
<meta property="og:title" content="{{ og_title or page_title or 'DevPlace' }}">

View File

@ -83,7 +83,7 @@
<article class="news-card">
{% if item.image_url %}
<div class="news-card-image">
<img src="{{ item.image_url }}" alt="" loading="lazy" class="image-fallback">
<img src="{{ item.image_url }}" alt="{{ item.title }}" loading="lazy" class="image-fallback">
</div>
{% endif %}
<div class="news-card-body">

View File

@ -15,7 +15,7 @@
<article class="news-card">
{% if item.image_url %}
<div class="news-card-image">
<img src="{{ item.image_url }}" alt="" loading="lazy" class="image-fallback">
<img src="{{ item.image_url }}" alt="{{ item.article['title'] }}" loading="lazy" class="image-fallback">
</div>
{% endif %}
<div class="news-card-body">

View File

@ -8,7 +8,7 @@
<article class="news-detail-card">
{% if image_url %}
<div class="news-detail-image">
<img src="{{ image_url }}" alt="" loading="lazy" class="image-fallback">
<img src="{{ image_url }}" alt="{{ article['title'] }}" loading="lazy" class="image-fallback">
</div>
{% endif %}

View File

@ -135,19 +135,20 @@ def seed_data(environment, **kwargs):
logger.info(f"Created {len(POST_UIDS)} seed posts")
for seed in SEED_USERS[:2]:
try:
opener = logged_in_opener(seed["email"])
body = urllib.parse.urlencode({
"content": f"Seed comment by {seed['username']}",
"post_uid": POST_UIDS[0],
}).encode()
opener.open(
urllib.request.Request(f"{host}/comments/create", data=body),
timeout=10,
)
except Exception as e:
logger.warning(f"Create seed comment failed: {e}")
for post_uid in POST_UIDS:
for seed in SEED_USERS[:2]:
try:
opener = logged_in_opener(seed["email"])
body = urllib.parse.urlencode({
"content": f"Seed comment by {seed['username']}",
"post_uid": post_uid,
}).encode()
opener.open(
urllib.request.Request(f"{host}/comments/create", data=body),
timeout=10,
)
except Exception as e:
logger.warning(f"Create seed comment failed: {e}")
for seed in SEED_USERS[:2]:
try:
@ -306,11 +307,11 @@ def seed_data(environment, **kwargs):
except Exception as e:
logger.warning(f"Harvest from {seed['username']} failed: {e}")
if POST_UIDS:
for post_uid in POST_UIDS:
try:
opener = logged_in_opener(SEED_USERS[0]["email"])
resp = opener.open(
urllib.request.Request(f"{host}/posts/{POST_UIDS[0]}"),
urllib.request.Request(f"{host}/posts/{post_uid}"),
timeout=10,
)
html = resp.read().decode("utf-8", errors="replace")
@ -1021,3 +1022,70 @@ class AdminUser(HttpUser):
self.client.post(
f"/admin/users/{uid}/toggle", name="admin/users/toggle"
)
class AnonymousUser(HttpUser):
weight = 3
wait_time = between(1, 5)
@task(5)
def browse_feed(self):
self.client.get("/feed", name="feed")
@task(3)
def browse_post(self):
if POST_SLUGS:
self.client.get(
f"/posts/{random.choice(POST_SLUGS)}", name="posts/[uid]"
)
@task(2)
def browse_public(self):
path = random.choice([
"/", "/news", "/gists", "/projects", "/leaderboard",
])
self.client.get(path, name="public/[page]")
@task(2)
def browse_detail(self):
if NEWS_SLUGS:
self.client.get(
f"/news/{random.choice(NEWS_SLUGS)}", name="news/[slug]"
)
if GIST_SLUGS:
self.client.get(
f"/gists/{random.choice(GIST_SLUGS)}", name="gists/[slug]"
)
if PROJECT_SLUGS:
self.client.get(
f"/projects/{random.choice(PROJECT_SLUGS)}", name="projects/[slug]"
)
@task(2)
def guest_vote_redirects_to_login(self):
if not POST_UIDS:
return
uid = random.choice(POST_UIDS)
with self.client.post(
f"/votes/post/{uid}", data={"value": 1},
headers={"x-requested-with": "fetch"},
catch_response=True, name="votes/post/guest-redirect",
) as resp:
if "/auth/login" in resp.url:
resp.success()
else:
resp.failure(f"guest vote not redirected to login: {resp.url}")
@task(1)
def guest_guarded_get_redirects_to_login(self):
with self.client.get(
"/messages", catch_response=True, name="guarded/guest-redirect",
) as resp:
if "/auth/login" in resp.url:
resp.success()
else:
resp.failure(f"guarded GET not redirected to login: {resp.url}")
@task(1)
def login_with_next(self):
self.client.get("/auth/login", params={"next": "/feed"}, name="auth/login?next")

View File

@ -25,7 +25,7 @@ dependencies = [
devplace = "devplacepy.cli:main"
[project.optional-dependencies]
dev = ["pytest", "playwright", "pytest-xdist", "requests"]
dev = ["pytest", "playwright", "pytest-xdist", "requests", "coverage"]
[build-system]
requires = ["hatchling"]

View File

@ -229,3 +229,10 @@ def bob(browser, seeded_db):
yield p, seeded_db["bob"]
p.close()
ctx.close()
@pytest.fixture(scope="session")
def local_db():
from devplacepy.database import init_db, db
init_db()
return db

85
tests/test_cache.py Normal file
View File

@ -0,0 +1,85 @@
import devplacepy.cache as cache_mod
from devplacepy.cache import TTLCache
class FakeClock:
def __init__(self, start=1000.0):
self.now = start
def time(self):
return self.now
def test_get_miss_returns_none():
cache = TTLCache(ttl=60)
assert cache.get("absent") is None
def test_set_and_get_round_trip():
cache = TTLCache(ttl=60)
cache.set("key", "value")
assert cache.get("key") == "value"
def test_entry_expires_after_ttl(monkeypatch):
clock = FakeClock()
monkeypatch.setattr(cache_mod, "time", clock)
cache = TTLCache(ttl=10)
cache.set("key", "value")
clock.now += 11
assert cache.get("key") is None
def test_lru_evicts_oldest_when_over_max_size():
cache = TTLCache(ttl=60, max_size=2)
cache.set("a", 1)
cache.set("b", 2)
cache.set("c", 3)
assert cache.get("a") is None
assert cache.get("b") == 2
assert cache.get("c") == 3
def test_get_promotes_recency():
cache = TTLCache(ttl=60, max_size=2)
cache.set("a", 1)
cache.set("b", 2)
cache.get("a")
cache.set("c", 3)
assert cache.get("a") == 1
assert cache.get("b") is None
assert cache.get("c") == 3
def test_pop_removes_entry():
cache = TTLCache(ttl=60)
cache.set("key", "value")
cache.pop("key")
assert cache.get("key") is None
def test_pop_missing_key_is_noop():
cache = TTLCache(ttl=60)
cache.pop("absent")
def test_clear_empties_cache():
cache = TTLCache(ttl=60)
cache.set("a", 1)
cache.set("b", 2)
cache.clear()
assert cache.get("a") is None
assert cache.get("b") is None
def test_items_excludes_expired_entries(monkeypatch):
clock = FakeClock()
monkeypatch.setattr(cache_mod, "time", clock)
cache = TTLCache(ttl=10)
cache.set("old", 1)
clock.now += 5
cache.set("fresh", 2)
clock.now += 7
items = dict(cache.items())
assert "old" not in items
assert items["fresh"] == 2

102
tests/test_cli.py Normal file
View File

@ -0,0 +1,102 @@
import argparse
from datetime import datetime, timezone, timedelta
import pytest
from devplacepy import cli
from devplacepy.database import get_table
from devplacepy.utils import generate_uid
def _make_user(role="Member"):
username = f"cli_{generate_uid()[:8]}"
get_table("users").insert({
"uid": generate_uid(),
"username": username,
"email": f"{username}@t.dev",
"role": role,
"created_at": datetime.now(timezone.utc).isoformat(),
})
return username
def test_role_get_prints_lowercased_role(local_db, capsys):
username = _make_user(role="Admin")
cli.cmd_role_get(argparse.Namespace(username=username))
assert capsys.readouterr().out.strip() == "admin"
def test_role_get_missing_user_exits(local_db):
with pytest.raises(SystemExit):
cli.cmd_role_get(argparse.Namespace(username="cli_absent_xyz"))
def test_role_set_updates_role(local_db):
username = _make_user(role="Member")
cli.cmd_role_set(argparse.Namespace(username=username, role="admin"))
user = get_table("users").find_one(username=username)
assert user["role"] == "Admin"
def test_role_set_invalid_role_exits(local_db):
username = _make_user()
with pytest.raises(SystemExit):
cli.cmd_role_set(argparse.Namespace(username=username, role="superuser"))
def test_role_set_missing_user_exits(local_db):
with pytest.raises(SystemExit):
cli.cmd_role_set(argparse.Namespace(username="cli_absent_xyz", role="admin"))
def test_news_clear_deletes_all_tables(local_db, capsys):
for table in ("news", "news_images", "news_sync"):
get_table(table).insert({"uid": generate_uid(), "marker": "cli"})
cli.cmd_news_clear(argparse.Namespace())
for table in ("news", "news_images", "news_sync"):
assert get_table(table).count() == 0
assert "News data cleared" in capsys.readouterr().out
def test_news_sanitize_strips_html(local_db, capsys):
uid = generate_uid()
get_table("news").insert({
"uid": uid,
"description": "<b>Bold</b> &amp; clean",
"content": "<p>Body</p>",
})
cli.cmd_news_sanitize(argparse.Namespace())
row = get_table("news").find_one(uid=uid)
assert row["description"] == "Bold & clean"
assert row["content"] == "Body"
assert "Sanitized" in capsys.readouterr().out
def test_attachments_prune_removes_only_stale_orphans(local_db, capsys):
old_uid = generate_uid()
fresh_uid = generate_uid()
attachments = get_table("attachments")
attachments.insert({
"uid": old_uid, "target_type": "", "target_uid": "",
"created_at": "2000-01-01T00:00:00+00:00",
})
attachments.insert({
"uid": fresh_uid, "target_type": "", "target_uid": "",
"created_at": datetime.now(timezone.utc).isoformat(),
})
cli.cmd_attachments_prune(argparse.Namespace(hours=24))
assert attachments.find_one(uid=old_uid) is None
assert attachments.find_one(uid=fresh_uid) is not None
def test_main_without_command_exits(monkeypatch):
monkeypatch.setattr("sys.argv", ["devplace"])
with pytest.raises(SystemExit):
cli.main()
def test_main_dispatches_subcommand(local_db, monkeypatch, capsys):
username = _make_user(role="Admin")
monkeypatch.setattr("sys.argv", ["devplace", "role", "get", username])
cli.main()
assert capsys.readouterr().out.strip() == "admin"

View File

@ -0,0 +1,58 @@
from datetime import datetime, timezone
from devplacepy.content import is_owner, canonical_redirect, first_image_url, enrich_items
from devplacepy.database import get_table, get_users_by_uids
from devplacepy.utils import generate_uid
def test_is_owner_matches_user_uid():
assert is_owner({"user_uid": "u1"}, {"uid": "u1"}) is True
assert is_owner({"user_uid": "u1"}, {"uid": "u2"}) is False
assert is_owner(None, {"uid": "u1"}) is False
assert is_owner({"user_uid": "u1"}, None) is False
def test_canonical_redirect_when_slug_differs():
item = {"slug": "abcd1234-title", "uid": "abcd1234"}
assert canonical_redirect("posts", item, "abcd1234-title") is None
response = canonical_redirect("posts", item, "abcd1234")
assert response is not None
assert response.status_code == 301
assert response.headers["location"] == "/posts/abcd1234-title"
def test_first_image_url_prefers_inline():
assert first_image_url({"image": "pic.png"}, None) == "/static/uploads/pic.png"
def test_first_image_url_uses_first_image_attachment():
item = {"image": None}
attachments = [{"is_image": False, "url": "/a"}, {"is_image": True, "url": "/b"}]
assert first_image_url(item, attachments) == "/b"
def test_first_image_url_none_when_absent():
assert first_image_url({}, []) is None
def test_enrich_items_attaches_author_and_extras(local_db):
uid = generate_uid()
username = f"enrich_{uid[:8]}"
get_table("users").insert({
"uid": uid, "username": username, "email": f"{username}@t.dev",
"created_at": datetime.now(timezone.utc).isoformat(),
})
item_uid = generate_uid()
items = [{"uid": item_uid, "user_uid": uid, "created_at": datetime.now(timezone.utc).isoformat()}]
authors = get_users_by_uids([uid])
enriched = enrich_items(
items, "post", authors,
extra_maps={"comment_count": {item_uid: 4}, "flag": lambda item: "ok"},
)
entry = enriched[0]
assert entry["post"] is items[0]
assert entry["author"]["username"] == username
assert entry["time_ago"] == "just now"
assert entry["my_vote"] == 0
assert entry["comment_count"] == 4
assert entry["flag"] == "ok"

115
tests/test_db_helpers.py Normal file
View File

@ -0,0 +1,115 @@
from datetime import datetime, timezone
from devplacepy.database import (
get_table,
get_users_by_uids,
get_vote_counts,
get_user_votes,
get_comment_counts_by_post_uids,
resolve_by_slug,
get_user_stars,
get_user_rank,
build_pagination,
)
from devplacepy.utils import generate_uid
def _now():
return datetime.now(timezone.utc).isoformat()
def _user():
uid = generate_uid()
get_table("users").insert({
"uid": uid, "username": f"dbh_{uid[:8]}", "email": f"{uid[:8]}@t.dev",
"created_at": _now(),
})
return uid
def _post(owner, **extra):
uid = generate_uid()
get_table("posts").insert({
"uid": uid, "user_uid": owner, "slug": f"{uid[:8]}-post",
"title": None, "content": "db helper post body", "topic": "random",
"project_uid": None, "image": None, "stars": 0, "created_at": _now(),
**extra,
})
return uid
def test_get_users_by_uids_dedupes(local_db):
a, b = generate_uid(), generate_uid()
users = get_table("users")
users.insert({"uid": a, "username": f"u_{a[:6]}", "created_at": _now()})
users.insert({"uid": b, "username": f"u_{b[:6]}", "created_at": _now()})
result = get_users_by_uids([a, b, a])
assert set(result.keys()) == {a, b}
def test_get_vote_counts_groups_up_and_down(local_db):
target = generate_uid()
votes = get_table("votes")
for value in (1, 1, -1):
votes.insert({
"uid": generate_uid(), "user_uid": generate_uid(),
"target_uid": target, "target_type": "post", "value": value,
"created_at": _now(),
})
ups, downs = get_vote_counts([target])
assert ups[target] == 2
assert downs[target] == 1
def test_get_user_votes_returns_user_value(local_db):
target = generate_uid()
user = generate_uid()
get_table("votes").insert({
"uid": generate_uid(), "user_uid": user,
"target_uid": target, "target_type": "post", "value": 1, "created_at": _now(),
})
assert get_user_votes(user, [target]) == {target: 1}
def test_get_comment_counts_by_post_uids(local_db):
post = generate_uid()
comments = get_table("comments")
for _ in range(2):
comments.insert({
"uid": generate_uid(), "user_uid": generate_uid(),
"target_type": "post", "target_uid": post, "content": "hi", "created_at": _now(),
})
assert get_comment_counts_by_post_uids([post]) == {post: 2}
def test_resolve_by_slug_finds_by_slug_or_uid(local_db):
owner = _user()
uid = _post(owner)
posts = get_table("posts")
slug = posts.find_one(uid=uid)["slug"]
assert resolve_by_slug(posts, slug)["uid"] == uid
assert resolve_by_slug(posts, uid)["uid"] == uid
assert resolve_by_slug(posts, "missing-slug-xyz") is None
def test_get_user_stars_sums_content_stars(local_db):
user = _user()
_post(user, stars=7)
assert get_user_stars(user) == 7
def test_get_user_rank_unknown_user_is_none(local_db):
assert get_user_rank(generate_uid()) is None
def test_build_pagination_metadata():
first = build_pagination(1, 100, 25)
assert first["total_pages"] == 4
assert first["has_prev"] is False
assert first["has_next"] is True
middle = build_pagination(2, 100, 25)
assert middle["has_prev"] is True
assert middle["has_next"] is True
clamped = build_pagination(99, 10, 25)
assert clamped["page"] == 1
assert clamped["total_pages"] == 1

75
tests/test_follow_api.py Normal file
View File

@ -0,0 +1,75 @@
import time
import requests
from tests.conftest import BASE_URL
from devplacepy.database import get_table
_counter = [0]
def _session():
_counter[0] += 1
name = f"flw{int(time.time() * 1000)}{_counter[0]}"
s = requests.Session()
s.post(f"{BASE_URL}/auth/signup", data={
"username": name, "email": f"{name}@t.dev",
"password": "secret123", "confirm_password": "secret123",
}, allow_redirects=True)
return s, name
def _uid(username):
return get_table("users").find_one(username=username)["uid"]
def test_follow_creates_notification_and_awards_xp(app_server):
s_a, a_name = _session()
_, b_name = _session()
a_uid, b_uid = _uid(a_name), _uid(b_name)
before = get_table("users").find_one(uid=b_uid).get("xp", 0) or 0
s_a.post(f"{BASE_URL}/follow/{b_name}", allow_redirects=False)
assert get_table("follows").count(follower_uid=a_uid, following_uid=b_uid) == 1
assert get_table("notifications").count(user_uid=b_uid, type="follow") == 1
after = get_table("users").find_one(uid=b_uid).get("xp", 0) or 0
assert after - before == 5
def test_duplicate_follow_is_idempotent(app_server):
s_a, a_name = _session()
_, b_name = _session()
a_uid, b_uid = _uid(a_name), _uid(b_name)
s_a.post(f"{BASE_URL}/follow/{b_name}", allow_redirects=False)
s_a.post(f"{BASE_URL}/follow/{b_name}", allow_redirects=False)
assert get_table("follows").count(follower_uid=a_uid, following_uid=b_uid) == 1
assert get_table("notifications").count(user_uid=b_uid, type="follow") == 1
def test_self_follow_rejected(app_server):
s_a, a_name = _session()
a_uid = _uid(a_name)
s_a.post(f"{BASE_URL}/follow/{a_name}", allow_redirects=False)
assert get_table("follows").count(follower_uid=a_uid, following_uid=a_uid) == 0
def test_follow_nonexistent_user_redirects(app_server):
s_a, _ = _session()
r = s_a.post(f"{BASE_URL}/follow/no_such_user_zzz", allow_redirects=False)
assert r.status_code == 302
def test_unfollow_removes_follow(app_server):
s_a, a_name = _session()
_, b_name = _session()
a_uid, b_uid = _uid(a_name), _uid(b_name)
s_a.post(f"{BASE_URL}/follow/{b_name}", allow_redirects=False)
s_a.post(f"{BASE_URL}/follow/unfollow/{b_name}", allow_redirects=False)
assert get_table("follows").count(follower_uid=a_uid, following_uid=b_uid) == 0
r = s_a.post(f"{BASE_URL}/follow/unfollow/{b_name}", allow_redirects=False)
assert r.status_code == 302

219
tests/test_news_service.py Normal file
View File

@ -0,0 +1,219 @@
import asyncio
import httpx
from devplacepy.services import news as news_mod
from devplacepy.services.news import NewsService, _extract_grade, _get_ai_key, _get_article_images
from devplacepy.database import get_table
from devplacepy.utils import generate_uid
API_URL = "http://news.test/api"
AI_URL = "http://ai.test/v1/chat"
LINK_HIGH = "http://news.test/high"
LINK_LOW = "http://news.test/low"
class FakeResp:
def __init__(self, json_data=None, text="", status=200):
self._json = json_data
self.text = text
self.status_code = status
def raise_for_status(self):
if self.status_code >= 400:
raise httpx.HTTPError(f"status {self.status_code}")
def json(self):
return self._json
class FakeClient:
def __init__(self, articles):
self.articles = articles
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
async def get(self, url, timeout=None):
if url == API_URL:
return FakeResp(json_data={"articles": self.articles})
return FakeResp(text='<img src="http://img.test/a.png">')
async def post(self, url, json=None, headers=None, timeout=None):
prompt = json["messages"][0]["content"]
if "FailArticle" in prompt:
return FakeResp(status=500, text="err")
if "EmptyArticle" in prompt:
return FakeResp(json_data={"choices": [{"message": {"content": ""}}]})
if "BadArticle" in prompt:
return FakeResp(json_data={"choices": [{"message": {"content": "no number"}}]})
grade = "9" if "HighArticle" in prompt else "3"
return FakeResp(json_data={"choices": [{"message": {"content": grade}}]})
class FailingApiClient(FakeClient):
async def get(self, url, timeout=None):
if url == API_URL:
raise httpx.HTTPError("api down")
return FakeResp(text="")
def _settings_stub(threshold="7"):
def fake_get_setting(key, default=None):
return {
"news_api_url": API_URL,
"news_ai_url": AI_URL,
"news_ai_model": "test-model",
"news_grade_threshold": threshold,
"news_ai_key": "",
}.get(key, default)
return fake_get_setting
def test_extract_grade_parsing():
assert _extract_grade("8") == 8
assert _extract_grade("Grade: 9") == 9
assert _extract_grade("Score is 7/10") == 7
assert _extract_grade("0") is None
assert _extract_grade("11") is None
assert _extract_grade("not a number") is None
def test_get_ai_key_env_precedence(monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setenv("NEWS_AI_KEY", "primary")
assert _get_ai_key() == "primary"
def test_get_ai_key_setting_fallback(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", lambda key, default=None: "from-setting" if key == "news_ai_key" else default)
assert _get_ai_key() == "from-setting"
def test_get_ai_key_openrouter_fallback(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setenv("OPENROUTER_API_KEY", "router-key")
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
assert _get_ai_key() == "router-key"
def test_get_ai_key_openai_fallback(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setenv("OPENAI_API_KEY", "openai-key")
assert _get_ai_key() == "openai-key"
def test_get_ai_key_empty_when_unset(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
assert _get_ai_key() == ""
def test_grade_article_empty_content_returns_none(local_db, monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
grade = asyncio.run(NewsService()._grade_article(
{"title": "EmptyArticle", "description": "d", "content": "c"}, AI_URL, "m", FakeClient([])))
assert grade is None
def test_grade_article_unparseable_returns_none(local_db, monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
grade = asyncio.run(NewsService()._grade_article(
{"title": "BadArticle", "description": "d", "content": "c"}, AI_URL, "m", FakeClient([])))
assert grade is None
def test_run_once_handles_api_failure(local_db, monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setattr(news_mod.httpx, "AsyncClient", lambda *a, **k: FailingApiClient([]))
asyncio.run(NewsService().run_once())
def test_run_once_updates_existing_news_row(local_db, monkeypatch):
external_id = f"news-{generate_uid()}"
existing_uid = generate_uid()
get_table("news").insert({
"uid": existing_uid, "external_id": external_id, "slug": "",
"title": "Old Title", "status": "draft", "grade": 0, "synced_at": "2020-01-01",
})
articles = [{"guid": external_id, "title": "HighArticle", "description": "d", "content": "c",
"link": "", "feed_name": "Feed", "author": "A", "published": "2026-01-01"}]
monkeypatch.setattr(news_mod, "get_setting", _settings_stub(threshold="7"))
monkeypatch.setattr(news_mod.httpx, "AsyncClient", lambda *a, **k: FakeClient(articles))
asyncio.run(NewsService().run_once())
row = get_table("news").find_one(uid=existing_uid)
assert row["title"] == "HighArticle"
assert row["status"] == "published"
assert get_table("news").count(external_id=external_id) == 1
def test_get_article_images_filters_and_dedupes():
html = (
'<img src="http://img.test/a.png">'
"<img src='http://img.test/a.png'>"
'<img src="http://img.test/b.svg">'
'<img src="/relative.png">'
)
client = FakeClient([])
async def fake_get(url, timeout=None):
return FakeResp(text=html)
client.get = fake_get
images = asyncio.run(_get_article_images("http://news.test/page", client))
assert [img["url"] for img in images] == ["http://img.test/a.png"]
def test_get_article_images_network_error_returns_empty():
client = FakeClient([])
async def failing_get(url, timeout=None):
raise httpx.HTTPError("down")
client.get = failing_get
assert asyncio.run(_get_article_images("http://news.test/page", client)) == []
def test_run_once_publishes_grades_and_is_idempotent(local_db, monkeypatch):
g_high, g_low, g_fail = (f"news-{generate_uid()}" for _ in range(3))
articles = [
{"guid": g_high, "title": "HighArticle", "description": "d", "content": "c",
"link": LINK_HIGH, "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
{"guid": g_low, "title": "LowArticle", "description": "d", "content": "c",
"link": LINK_LOW, "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
{"guid": g_fail, "title": "FailArticle", "description": "d", "content": "c",
"link": "", "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
{"guid": "", "title": "NoGuid", "description": "d", "content": "c",
"link": "", "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
]
monkeypatch.setattr(news_mod, "get_setting", _settings_stub(threshold="7"))
monkeypatch.setattr(news_mod.httpx, "AsyncClient", lambda *a, **k: FakeClient(articles))
asyncio.run(NewsService().run_once())
news = get_table("news")
sync = get_table("news_sync")
assert news.find_one(external_id=g_high)["status"] == "published"
assert news.find_one(external_id=g_high)["grade"] == 9
assert news.find_one(external_id=g_low)["status"] == "draft"
assert news.find_one(external_id=g_low)["grade"] == 3
fail_row = news.find_one(external_id=g_fail)
assert fail_row["status"] == "draft"
assert fail_row["grade"] == 0
assert sync.find_one(external_id=g_fail)["status"] == "grading_failed"
assert sync.find_one(external_id=g_high)["status"] == "graded"
asyncio.run(NewsService().run_once())
assert news.count(external_id=g_high) == 1
assert news.count(external_id=g_fail) == 1

View File

@ -62,3 +62,30 @@ def test_service_worker_served(app_server):
def test_manifest_served(app_server):
r = requests.get(f"{BASE_URL}/manifest.json")
assert r.status_code == 200
def test_browser_base64_is_url_safe_unpadded():
import base64
from devplacepy import push
encoded = push.browser_base64(b"\xff\xfe\x00 hello")
assert "=" not in encoded
assert "+" not in encoded and "/" not in encoded
assert base64.urlsafe_b64decode(encoded + "==") == b"\xff\xfe\x00 hello"
def test_hkdf_returns_requested_length():
from devplacepy import push
derived = push.hkdf(b"input-key-material", b"salt", b"info", 16)
assert isinstance(derived, bytes)
assert len(derived) == 16
def test_public_key_standard_b64_non_empty():
from devplacepy import push
assert push.public_key_standard_b64()
def test_create_notification_authorization_is_jwt():
from devplacepy import push
token = push.create_notification_authorization("https://push.example.com/endpoint")
assert token.count(".") == 2

View File

@ -213,3 +213,184 @@ def test_news_detail_has_newsarticle_schema(page, app_server):
text = " ".join(scripts.nth(i).text_content() for i in range(scripts.count()))
assert "NewsArticle" in text
assert page.locator('meta[property="og:type"]').get_attribute("content") == "article"
def _seed_owner():
from datetime import datetime, timezone
from uuid import uuid4
from devplacepy.database import get_table
uid = str(uuid4())
get_table("users").insert({
"uid": uid,
"username": f"seo_{uid[:8]}",
"email": f"{uid[:8]}@seo.test",
"password_hash": "x",
"role": "Member",
"is_active": True,
"created_at": datetime.now(timezone.utc).isoformat(),
})
return uid
def _seed_post(image=None):
from datetime import datetime, timezone
from uuid import uuid4
from devplacepy.database import get_table
from devplacepy.utils import make_combined_slug
owner = _seed_owner()
uid = str(uuid4())
slug = make_combined_slug("SEO Detail Post", uid)
get_table("posts").insert({
"uid": uid, "user_uid": owner, "slug": slug,
"title": "SEO Detail Post", "content": "Body text for the SEO detail post.",
"topic": "general", "project_uid": None, "image": image,
"stars": 0, "created_at": datetime.now(timezone.utc).isoformat(),
})
return slug, uid
def _seed_gist():
from datetime import datetime, timezone
from uuid import uuid4
from devplacepy.database import get_table
from devplacepy.utils import make_combined_slug
owner = _seed_owner()
uid = str(uuid4())
slug = make_combined_slug("SEO Detail Gist", uid)
get_table("gists").insert({
"uid": uid, "user_uid": owner, "slug": slug,
"title": "SEO Detail Gist", "description": "Gist description for SEO tests.",
"source_code": "print('seo')", "language": "python",
"stars": 0, "created_at": datetime.now(timezone.utc).isoformat(),
})
return slug, uid
def _seed_project():
from datetime import datetime, timezone
from uuid import uuid4
from devplacepy.database import get_table
from devplacepy.utils import make_combined_slug
owner = _seed_owner()
uid = str(uuid4())
slug = make_combined_slug("SEO Detail Project", uid)
get_table("projects").insert({
"uid": uid, "user_uid": owner, "slug": slug,
"title": "SEO Detail Project", "description": "Project description for SEO tests.",
"project_type": "software", "platforms": "Linux", "status": "Released",
"stars": 0, "created_at": datetime.now(timezone.utc).isoformat(),
})
return slug, uid
def _seed_news_image(news_uid):
from devplacepy.database import get_table
get_table("news_images").insert({
"news_uid": news_uid,
"url": "https://example.com/seo-news-image.jpg",
})
def _seed_feed_posts(count):
from datetime import datetime, timezone, timedelta
from uuid import uuid4
from devplacepy.database import get_table
from devplacepy.utils import make_combined_slug
owner = _seed_owner()
topic = f"seopag{owner[:8]}"
base = datetime(2026, 1, 1, tzinfo=timezone.utc)
posts = get_table("posts")
for i in range(count):
uid = str(uuid4())
posts.insert({
"uid": uid, "user_uid": owner, "slug": make_combined_slug(f"seo pag {i}", uid),
"title": None, "content": f"seo pag post {i}", "topic": topic, "project_uid": None,
"image": None, "stars": 0,
"created_at": (base - timedelta(seconds=i)).isoformat(),
})
return topic
def test_post_uid_redirects_to_canonical_slug(app_server):
slug, uid = _seed_post()
r = requests.get(f"{BASE_URL}/posts/{uid}", allow_redirects=False)
assert r.status_code == 301
assert r.headers["location"].endswith(f"/posts/{slug}"), r.headers.get("location")
def test_post_slug_served_without_redirect(app_server):
slug, uid = _seed_post()
r = requests.get(f"{BASE_URL}/posts/{slug}", allow_redirects=False)
assert r.status_code == 200
def test_gist_uid_redirects_to_canonical_slug(app_server):
slug, uid = _seed_gist()
r = requests.get(f"{BASE_URL}/gists/{uid}", allow_redirects=False)
assert r.status_code == 301
assert r.headers["location"].endswith(f"/gists/{slug}"), r.headers.get("location")
def test_project_uid_redirects_to_canonical_slug(app_server):
slug, uid = _seed_project()
r = requests.get(f"{BASE_URL}/projects/{uid}", allow_redirects=False)
assert r.status_code == 301
assert r.headers["location"].endswith(f"/projects/{slug}"), r.headers.get("location")
def test_news_uid_redirects_to_canonical_slug(app_server):
slug, uid = _seed_news()
r = requests.get(f"{BASE_URL}/news/{uid}", allow_redirects=False)
assert r.status_code == 301
assert r.headers["location"].endswith(f"/news/{slug}"), r.headers.get("location")
def test_missing_profile_returns_404(app_server):
r = requests.get(f"{BASE_URL}/profile/no-such-user-xyz", allow_redirects=False)
assert r.status_code == 404
def test_post_og_image_uses_content_image(page, app_server):
slug, _ = _seed_post(image="seo-content.png")
page.goto(f"{BASE_URL}/posts/{slug}", wait_until="domcontentloaded")
og = page.locator('meta[property="og:image"]').get_attribute("content")
assert og.endswith("/static/uploads/seo-content.png"), og
def test_post_page_has_breadcrumb_schema(page, app_server):
slug, _ = _seed_post()
page.goto(f"{BASE_URL}/posts/{slug}", wait_until="domcontentloaded")
scripts = page.locator('script[type="application/ld+json"]')
text = " ".join(scripts.nth(i).text_content() for i in range(scripts.count()))
assert "BreadcrumbList" in text
def test_post_schema_has_date_modified(page, app_server):
slug, _ = _seed_post()
page.goto(f"{BASE_URL}/posts/{slug}", wait_until="domcontentloaded")
scripts = page.locator('script[type="application/ld+json"]')
text = " ".join(scripts.nth(i).text_content() for i in range(scripts.count()))
assert "dateModified" in text
def test_news_detail_image_has_alt_text(page, app_server):
slug, uid = _seed_news()
_seed_news_image(uid)
page.goto(f"{BASE_URL}/news/{slug}", wait_until="domcontentloaded")
alt = page.locator(".news-detail-image img").get_attribute("alt")
assert alt and alt.strip()
def test_feed_pagination_emits_rel_next(page, app_server):
topic = _seed_feed_posts(26)
page.goto(f"{BASE_URL}/feed?topic={topic}", wait_until="domcontentloaded")
nxt = page.locator('link[rel="next"]')
assert nxt.count() == 1
href = nxt.get_attribute("href")
assert "before=" in href and f"topic={topic}" in href, href
def test_post_detail_has_no_rel_next(page, app_server):
slug, _ = _seed_post()
page.goto(f"{BASE_URL}/posts/{slug}", wait_until="domcontentloaded")
assert page.locator('link[rel="next"]').count() == 0

View File

@ -104,3 +104,78 @@ def test_badge_info_unknown_fallback():
meta = badge_info("Nonexistent Badge")
assert meta["icon"]
assert meta["description"] == "Nonexistent Badge"
from devplacepy.utils import (
safe_next, strip_html, extract_mentions, award_badge, award_xp,
check_milestone_badges, create_mention_notifications,
)
from devplacepy.database import get_table
def _seed_user(role="Member", xp=0, level=1):
uid = generate_uid()
username = f"ut_{uid[:8]}"
get_table("users").insert({
"uid": uid, "username": username, "email": f"{username}@t.dev",
"role": role, "xp": xp, "level": level,
"created_at": datetime.now(timezone.utc).isoformat(),
})
return uid, username
def test_safe_next_allows_internal_rejects_external():
assert safe_next("/gists") == "/gists"
assert safe_next("//evil.com") == "/feed"
assert safe_next("http://evil.com") == "/feed"
assert safe_next(None) == "/feed"
def test_strip_html_removes_tags_and_unescapes():
assert strip_html("<b>Hi</b> &amp; bye") == "Hi & bye"
assert strip_html("") == ""
def test_extract_mentions_finds_usernames():
mentions = extract_mentions("hi @alice and (@bob_1), mail a@b.com")
assert mentions == ["alice", "bob_1"]
def test_award_badge_is_idempotent(local_db):
uid, _ = _seed_user()
assert award_badge(uid, "First Post") is True
assert award_badge(uid, "First Post") is False
def test_award_xp_levels_up_and_notifies(local_db):
uid, _ = _seed_user(xp=95, level=1)
result = award_xp(uid, 10)
assert result["xp"] == 105
assert result["level"] == 2
assert result["leveled_up"] is True
assert get_table("notifications").count(user_uid=uid, type="level") == 1
def test_check_milestone_badges_awards_prolific(local_db):
uid, _ = _seed_user()
posts = get_table("posts")
for index in range(10):
post_uid = generate_uid()
posts.insert({"uid": post_uid, "user_uid": uid, "slug": f"{post_uid[:8]}-p",
"title": None, "content": f"milestone post {index}", "topic": "random",
"project_uid": None, "image": None, "stars": 0,
"created_at": datetime.now(timezone.utc).isoformat()})
awarded = check_milestone_badges(uid)
assert "Prolific" in awarded
assert get_table("badges").find_one(user_uid=uid, badge_name="Prolific") is not None
def test_create_mention_notifications_targets_known_users(local_db):
actor_uid, actor_name = _seed_user()
target_uid, target_name = _seed_user()
create_mention_notifications(
f"hey @{target_name} and @{actor_name} and @ghost_zzz",
actor_uid, "/posts/x",
)
assert get_table("notifications").count(user_uid=target_uid, type="mention") == 1
assert get_table("notifications").count(user_uid=actor_uid, type="mention") == 0

101
tests/test_votes.py Normal file
View File

@ -0,0 +1,101 @@
import time
from datetime import datetime, timezone
import requests
from tests.conftest import BASE_URL
from devplacepy.database import get_table
from devplacepy.utils import generate_uid
_counter = [0]
AJAX = {"X-Requested-With": "fetch"}
def _session():
_counter[0] += 1
name = f"vot{int(time.time() * 1000)}{_counter[0]}"
s = requests.Session()
s.post(f"{BASE_URL}/auth/signup", data={
"username": name, "email": f"{name}@t.dev",
"password": "secret123", "confirm_password": "secret123",
}, allow_redirects=True)
return s, name
def _uid(username):
return get_table("users").find_one(username=username)["uid"]
def _make_post(owner_uid):
uid = generate_uid()
get_table("posts").insert({
"uid": uid, "user_uid": owner_uid, "slug": f"{uid[:8]}-vote-post",
"title": None, "content": "vote target content", "topic": "random",
"project_uid": None, "image": None, "stars": 0,
"created_at": datetime.now(timezone.utc).isoformat(),
})
return uid
def test_upvote_returns_ajax_payload(app_server):
s_a, a_name = _session()
s_b, _ = _session()
post_uid = _make_post(_uid(a_name))
r = s_b.post(f"{BASE_URL}/votes/post/{post_uid}", data={"value": "1"}, headers=AJAX)
assert r.json() == {"net": 1, "up": 1, "down": 0, "value": 1}
def test_repeated_vote_toggles_off(app_server):
s_a, a_name = _session()
s_b, _ = _session()
post_uid = _make_post(_uid(a_name))
s_b.post(f"{BASE_URL}/votes/post/{post_uid}", data={"value": "1"}, headers=AJAX)
r = s_b.post(f"{BASE_URL}/votes/post/{post_uid}", data={"value": "1"}, headers=AJAX)
assert r.json()["net"] == 0
assert r.json()["value"] == 0
def test_switch_upvote_to_downvote(app_server):
s_a, a_name = _session()
s_b, _ = _session()
post_uid = _make_post(_uid(a_name))
s_b.post(f"{BASE_URL}/votes/post/{post_uid}", data={"value": "1"}, headers=AJAX)
r = s_b.post(f"{BASE_URL}/votes/post/{post_uid}", data={"value": "-1"}, headers=AJAX)
assert r.json()["net"] == -1
assert r.json()["value"] == -1
def test_upvote_notifies_owner_and_awards_xp(app_server):
s_a, a_name = _session()
s_b, _ = _session()
a_uid = _uid(a_name)
post_uid = _make_post(a_uid)
before = get_table("users").find_one(uid=a_uid).get("xp", 0) or 0
s_b.post(f"{BASE_URL}/votes/post/{post_uid}", data={"value": "1"}, headers=AJAX)
assert get_table("notifications").count(user_uid=a_uid, type="vote") == 1
after = get_table("users").find_one(uid=a_uid).get("xp", 0) or 0
assert after - before == 5
def test_self_vote_does_not_notify(app_server):
s_a, a_name = _session()
a_uid = _uid(a_name)
post_uid = _make_post(a_uid)
before = get_table("notifications").count(user_uid=a_uid, type="vote")
s_a.post(f"{BASE_URL}/votes/post/{post_uid}", data={"value": "1"}, headers=AJAX)
after = get_table("notifications").count(user_uid=a_uid, type="vote")
assert after == before
def test_non_ajax_vote_redirects_to_referer(app_server):
s_a, a_name = _session()
s_b, _ = _session()
post_uid = _make_post(_uid(a_name))
r = s_b.post(
f"{BASE_URL}/votes/post/{post_uid}",
data={"value": "1"},
headers={"Referer": f"{BASE_URL}/feed"},
allow_redirects=False,
)
assert r.status_code == 302
assert r.headers["location"] == f"{BASE_URL}/feed"