Compare commits
4 Commits
aca07828f1
...
0aa9b1f561
| Author | SHA1 | Date | |
|---|---|---|---|
| 0aa9b1f561 | |||
| a98ded68e4 | |||
| 8d4a82965d | |||
| 9e836ea152 |
@ -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
5
.gitignore
vendored
@ -15,3 +15,8 @@ devplacepy/static/uploads/*.jpg
|
||||
devplacepy/static/uploads/*.jpeg
|
||||
devplacepy/static/uploads/*.gif
|
||||
devplacepy/static/uploads/*.webp
|
||||
|
||||
# coverage
|
||||
.coverage
|
||||
.coverage.*
|
||||
htmlcov/
|
||||
|
||||
20
Makefile
20
Makefile
@ -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); \
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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"},
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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),
|
||||
|
||||
@ -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']}"},
|
||||
|
||||
@ -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"},
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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}"
|
||||
|
||||
@ -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' }}">
|
||||
|
||||
@ -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">
|
||||
|
||||
@ -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">
|
||||
|
||||
@ -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 %}
|
||||
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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"]
|
||||
|
||||
@ -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
85
tests/test_cache.py
Normal 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
102
tests/test_cli.py
Normal 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> & 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"
|
||||
58
tests/test_content_unit.py
Normal file
58
tests/test_content_unit.py
Normal 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
115
tests/test_db_helpers.py
Normal 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
75
tests/test_follow_api.py
Normal 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
219
tests/test_news_service.py
Normal 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
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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> & 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
101
tests/test_votes.py
Normal 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"
|
||||
Loading…
Reference in New Issue
Block a user