f0f6e7542f
- Add generate_articles task handler to worker.py - template_generate and template_regenerate now enqueue tasks instead of running inline (was blocking HTTP request for seconds with 1k articles) - rebuild_all enqueues per-template + inline rebuilds manual articles - Update tests to check task enqueue instead of immediate article creation Subtask 4 of CMS admin improvement. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1271 lines
53 KiB
Python
1271 lines
53 KiB
Python
"""
|
||
Tests for the programmatic SEO content generation engine.
|
||
|
||
Covers: slugify utility, migration 0010, generation pipeline,
|
||
scenario widget baking, public article serving, markets hub,
|
||
sitemap integration, admin CRUD routes, and path collision prevention.
|
||
"""
|
||
import importlib
|
||
import json
|
||
import sqlite3
|
||
from datetime import date, datetime
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from padelnomics.content.routes import (
|
||
RESERVED_PREFIXES,
|
||
SCENARIO_RE,
|
||
SECTION_TEMPLATES,
|
||
bake_scenario_cards,
|
||
is_reserved_path,
|
||
)
|
||
from padelnomics.core import execute, fetch_all, fetch_one, slugify
|
||
from padelnomics.planner.calculator import calc, validate_state
|
||
|
||
SCHEMA_PATH = Path(__file__).parent.parent / "src" / "padelnomics" / "migrations" / "schema.sql"
|
||
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────
|
||
|
||
def _table_names(conn):
|
||
rows = conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='table'"
|
||
" AND name NOT LIKE 'sqlite_%' ORDER BY name"
|
||
).fetchall()
|
||
return [r[0] for r in rows]
|
||
|
||
|
||
def _trigger_names(conn):
|
||
rows = conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='trigger' ORDER BY name"
|
||
).fetchall()
|
||
return [r[0] for r in rows]
|
||
|
||
|
||
def _index_names(conn):
|
||
rows = conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='index'"
|
||
" AND name NOT LIKE 'sqlite_%' ORDER BY name"
|
||
).fetchall()
|
||
return [r[0] for r in rows]
|
||
|
||
|
||
def _column_names(conn, table):
|
||
return [r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()]
|
||
|
||
|
||
async def _create_published_scenario(slug="test-scenario", city="TestCity", country="TC"):
|
||
"""Insert a published scenario with real calc data, return its id."""
|
||
state = validate_state({"dblCourts": 4, "sglCourts": 2})
|
||
d = calc(state)
|
||
return await execute(
|
||
"""INSERT INTO published_scenarios
|
||
(slug, title, subtitle, location, country, venue_type, ownership,
|
||
court_config, state_json, calc_json)
|
||
VALUES (?, ?, '', ?, ?, 'indoor', 'rent', '4 double + 2 single', ?, ?)""",
|
||
(slug, f"Scenario {city}", city, country, json.dumps(state), json.dumps(d)),
|
||
)
|
||
|
||
|
||
async def _create_article(slug="test-article", url_path="/test-article",
|
||
status="published", published_at=None):
|
||
"""Insert an article row, return its id."""
|
||
pub = published_at or datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
|
||
return await execute(
|
||
"""INSERT INTO articles
|
||
(url_path, slug, title, meta_description, country, region,
|
||
status, published_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
(url_path, slug, f"Title {slug}", f"Desc {slug}", "US", "North America",
|
||
status, pub),
|
||
)
|
||
|
||
|
||
TEST_TEMPLATE = """\
|
||
---
|
||
name: "Test City Analysis"
|
||
slug: test-city
|
||
content_type: calculator
|
||
data_table: serving.test_cities
|
||
natural_key: city_slug
|
||
languages: [en]
|
||
url_pattern: "/markets/{{ country | lower }}/{{ city_slug }}"
|
||
title_pattern: "Padel in {{ city }}"
|
||
meta_description_pattern: "Padel costs in {{ city }}"
|
||
schema_type: Article
|
||
---
|
||
# Padel in {{ city }}
|
||
|
||
Welcome to {{ city }}.
|
||
|
||
[scenario:{{ scenario_slug }}:capex]
|
||
"""
|
||
|
||
TEST_ROWS = [
|
||
{"city": "Miami", "city_slug": "miami", "country": "US", "region": "North America", "electricity": 700},
|
||
{"city": "Madrid", "city_slug": "madrid", "country": "ES", "region": "Europe", "electricity": 500},
|
||
{"city": "Berlin", "city_slug": "berlin", "country": "DE", "region": "Europe", "electricity": 550},
|
||
]
|
||
|
||
TEST_COLUMNS = [
|
||
{"column_name": "city", "data_type": "VARCHAR"},
|
||
{"column_name": "city_slug", "data_type": "VARCHAR"},
|
||
{"column_name": "country", "data_type": "VARCHAR"},
|
||
{"column_name": "region", "data_type": "VARCHAR"},
|
||
{"column_name": "electricity", "data_type": "INTEGER"},
|
||
]
|
||
|
||
|
||
@pytest.fixture
|
||
def pseo_env(tmp_path, monkeypatch):
|
||
"""Set up pSEO environment: temp template dir, build dir, mock DuckDB."""
|
||
import padelnomics.content as content_mod
|
||
|
||
tpl_dir = tmp_path / "templates"
|
||
tpl_dir.mkdir()
|
||
monkeypatch.setattr(content_mod, "TEMPLATES_DIR", tpl_dir)
|
||
|
||
build_dir = tmp_path / "build"
|
||
build_dir.mkdir()
|
||
monkeypatch.setattr(content_mod, "BUILD_DIR", build_dir)
|
||
|
||
(tpl_dir / "test-city.md.jinja").write_text(TEST_TEMPLATE)
|
||
|
||
async def mock_fetch_analytics(query, params=None):
|
||
if "information_schema" in query:
|
||
return TEST_COLUMNS
|
||
if "WHERE" in query and params:
|
||
# preview_article: filter by natural key value
|
||
return [r for r in TEST_ROWS if params[0] in r.values()]
|
||
return TEST_ROWS
|
||
|
||
monkeypatch.setattr(content_mod, "fetch_analytics", mock_fetch_analytics)
|
||
|
||
return {"tpl_dir": tpl_dir, "build_dir": build_dir}
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# slugify()
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestSlugify:
|
||
def test_basic(self):
|
||
assert slugify("Hello World") == "hello-world"
|
||
|
||
def test_unicode_stripped(self):
|
||
assert slugify("Padel Court Cost in München") == "padel-court-cost-in-munchen"
|
||
|
||
def test_special_chars_removed(self):
|
||
assert slugify("Test/Special@Chars!") == "testspecialchars"
|
||
|
||
def test_max_length(self):
|
||
result = slugify("a " * 100, max_length_chars=20)
|
||
assert len(result) <= 20
|
||
|
||
def test_empty_string(self):
|
||
assert slugify("") == ""
|
||
|
||
def test_leading_trailing_hyphens_stripped(self):
|
||
assert slugify("---hello---") == "hello"
|
||
|
||
def test_multiple_spaces_collapsed(self):
|
||
assert slugify("hello world") == "hello-world"
|
||
|
||
def test_accented_characters(self):
|
||
assert slugify("café résumé naïve") == "cafe-resume-naive"
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Migration 0010
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestMigration0010:
|
||
"""Synchronous tests — migration uses stdlib sqlite3."""
|
||
|
||
def _run_migration(self, db_path):
|
||
conn = sqlite3.connect(db_path)
|
||
mod = importlib.import_module("padelnomics.migrations.versions.0010_add_content_tables")
|
||
mod.up(conn)
|
||
conn.commit()
|
||
return conn
|
||
|
||
def test_creates_all_tables(self, tmp_path):
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
tables = _table_names(conn)
|
||
assert "published_scenarios" in tables
|
||
assert "article_templates" in tables
|
||
assert "template_data" in tables
|
||
assert "articles" in tables
|
||
conn.close()
|
||
|
||
def test_creates_fts_table(self, tmp_path):
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
tables = _table_names(conn)
|
||
assert "articles_fts" in tables
|
||
conn.close()
|
||
|
||
def test_creates_sync_triggers(self, tmp_path):
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
triggers = _trigger_names(conn)
|
||
assert "articles_ai" in triggers
|
||
assert "articles_ad" in triggers
|
||
assert "articles_au" in triggers
|
||
conn.close()
|
||
|
||
def test_creates_indexes(self, tmp_path):
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
indexes = _index_names(conn)
|
||
assert "idx_pub_scenarios_slug" in indexes
|
||
assert "idx_article_templates_slug" in indexes
|
||
assert "idx_template_data_template" in indexes
|
||
assert "idx_articles_url_path" in indexes
|
||
assert "idx_articles_slug" in indexes
|
||
assert "idx_articles_status" in indexes
|
||
conn.close()
|
||
|
||
def test_published_scenarios_columns(self, tmp_path):
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
cols = _column_names(conn, "published_scenarios")
|
||
for expected in ("id", "slug", "title", "location", "country",
|
||
"venue_type", "ownership", "court_config",
|
||
"state_json", "calc_json", "template_data_id"):
|
||
assert expected in cols, f"{expected} missing from published_scenarios"
|
||
conn.close()
|
||
|
||
def test_articles_columns(self, tmp_path):
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
cols = _column_names(conn, "articles")
|
||
for expected in ("id", "url_path", "slug", "title", "meta_description",
|
||
"country", "region", "status", "published_at",
|
||
"template_data_id"):
|
||
assert expected in cols, f"{expected} missing from articles"
|
||
conn.close()
|
||
|
||
def test_idempotent(self, tmp_path):
|
||
"""Running the migration twice does not raise."""
|
||
db_path = str(tmp_path / "test.db")
|
||
conn = self._run_migration(db_path)
|
||
conn.close()
|
||
# Run again
|
||
conn = self._run_migration(db_path)
|
||
tables = _table_names(conn)
|
||
assert "articles" in tables
|
||
conn.close()
|
||
|
||
def test_fts_sync_on_insert(self, tmp_path):
|
||
"""Inserting into articles populates articles_fts via trigger."""
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
conn.execute(
|
||
"""INSERT INTO articles (url_path, slug, title, meta_description,
|
||
country, region, status, published_at)
|
||
VALUES ('/test', 'test', 'Test Title', 'desc', 'US', 'NA',
|
||
'published', '2026-01-01')"""
|
||
)
|
||
conn.commit()
|
||
rows = conn.execute(
|
||
"SELECT * FROM articles_fts WHERE articles_fts MATCH 'Test'"
|
||
).fetchall()
|
||
assert len(rows) == 1
|
||
conn.close()
|
||
|
||
def test_fts_sync_on_delete(self, tmp_path):
|
||
"""Deleting from articles removes from articles_fts via trigger."""
|
||
conn = self._run_migration(str(tmp_path / "test.db"))
|
||
conn.execute(
|
||
"""INSERT INTO articles (url_path, slug, title, meta_description,
|
||
country, region, status, published_at)
|
||
VALUES ('/test', 'test', 'UniqueTitle', 'desc', 'US', 'NA',
|
||
'published', '2026-01-01')"""
|
||
)
|
||
conn.commit()
|
||
conn.execute("DELETE FROM articles WHERE slug = 'test'")
|
||
conn.commit()
|
||
rows = conn.execute(
|
||
"SELECT * FROM articles_fts WHERE articles_fts MATCH 'UniqueTitle'"
|
||
).fetchall()
|
||
assert len(rows) == 0
|
||
conn.close()
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Scenario regex
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestScenarioRegex:
|
||
def test_simple_marker(self):
|
||
m = SCENARIO_RE.search("[scenario:my-slug]")
|
||
assert m.group(1) == "my-slug"
|
||
assert m.group(2) is None
|
||
|
||
def test_section_marker(self):
|
||
m = SCENARIO_RE.search("[scenario:my-slug:capex]")
|
||
assert m.group(1) == "my-slug"
|
||
assert m.group(2) == "capex"
|
||
|
||
def test_all_sections_have_templates(self):
|
||
for section in (None, "capex", "operating", "cashflow", "returns", "full"):
|
||
assert section in SECTION_TEMPLATES
|
||
|
||
def test_no_match_on_invalid(self):
|
||
assert SCENARIO_RE.search("[scenario:UPPER]") is None
|
||
assert SCENARIO_RE.search("[scenario:]") is None
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Path collision prevention
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestReservedPaths:
|
||
def test_admin_reserved(self):
|
||
assert is_reserved_path("/admin/anything") is True
|
||
|
||
def test_planner_reserved(self):
|
||
assert is_reserved_path("/planner/") is True
|
||
|
||
def test_markets_reserved(self):
|
||
assert is_reserved_path("/markets") is True
|
||
|
||
def test_custom_path_allowed(self):
|
||
assert is_reserved_path("/padel-court-cost-miami") is False
|
||
|
||
def test_leading_slash_normalized(self):
|
||
assert is_reserved_path("admin/foo") is True
|
||
|
||
def test_all_prefixes_covered(self):
|
||
"""Every prefix in the tuple is actually checked."""
|
||
for prefix in RESERVED_PREFIXES:
|
||
assert is_reserved_path(prefix + "/test") is True
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Scenario card baking
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestBakeScenarioCards:
|
||
async def test_no_markers_returns_unchanged(self, db):
|
||
html = "<p>Hello world</p>"
|
||
result = await bake_scenario_cards(html)
|
||
assert result == html
|
||
|
||
async def test_unknown_slug_left_as_marker(self, db):
|
||
"""Markers referencing non-existent scenarios are removed (replaced with nothing)."""
|
||
html = "<p>[scenario:nonexistent]</p>"
|
||
result = await bake_scenario_cards(html)
|
||
# The marker position is replaced but scenario not found → no replacement
|
||
assert "scenario-widget" not in result
|
||
|
||
async def test_summary_card_baked(self, db):
|
||
await _create_published_scenario(slug="test-bake")
|
||
html = "<div>[scenario:test-bake]</div>"
|
||
result = await bake_scenario_cards(html)
|
||
assert "scenario-widget" in result
|
||
assert "scenario-summary" in result
|
||
assert "TestCity" in result
|
||
|
||
async def test_capex_section_baked(self, db):
|
||
await _create_published_scenario(slug="test-capex")
|
||
html = "[scenario:test-capex:capex]"
|
||
result = await bake_scenario_cards(html)
|
||
assert "scenario-capex" in result
|
||
assert "Investment Breakdown" in result
|
||
|
||
async def test_operating_section_baked(self, db):
|
||
await _create_published_scenario(slug="test-op")
|
||
html = "[scenario:test-op:operating]"
|
||
result = await bake_scenario_cards(html)
|
||
assert "scenario-operating" in result
|
||
|
||
async def test_cashflow_section_baked(self, db):
|
||
await _create_published_scenario(slug="test-cf")
|
||
html = "[scenario:test-cf:cashflow]"
|
||
result = await bake_scenario_cards(html)
|
||
assert "scenario-cashflow" in result
|
||
|
||
async def test_returns_section_baked(self, db):
|
||
await _create_published_scenario(slug="test-ret")
|
||
html = "[scenario:test-ret:returns]"
|
||
result = await bake_scenario_cards(html)
|
||
assert "scenario-returns" in result
|
||
|
||
async def test_full_section_baked(self, db):
|
||
await _create_published_scenario(slug="test-full")
|
||
html = "[scenario:test-full:full]"
|
||
result = await bake_scenario_cards(html)
|
||
# Full includes all sections
|
||
assert "scenario-summary" in result
|
||
assert "scenario-capex" in result
|
||
assert "scenario-operating" in result
|
||
assert "scenario-cashflow" in result
|
||
assert "scenario-returns" in result
|
||
|
||
async def test_multiple_markers_same_slug(self, db):
|
||
await _create_published_scenario(slug="multi")
|
||
html = "[scenario:multi]\n---\n[scenario:multi:capex]"
|
||
result = await bake_scenario_cards(html)
|
||
assert "scenario-summary" in result
|
||
assert "scenario-capex" in result
|
||
|
||
async def test_planner_cta_link(self, db):
|
||
await _create_published_scenario(slug="cta-test")
|
||
html = "[scenario:cta-test]"
|
||
result = await bake_scenario_cards(html)
|
||
assert "planner" in result
|
||
assert "Try with your own numbers" in result
|
||
|
||
async def test_invalid_section_ignored(self, db):
|
||
await _create_published_scenario(slug="bad-sec")
|
||
html = "[scenario:bad-sec:nonexistent]"
|
||
result = await bake_scenario_cards(html)
|
||
# Invalid section key → no template → marker untouched
|
||
assert "scenario-widget" not in result
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Generation pipeline
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestGenerationPipeline:
|
||
async def test_generates_correct_count(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
generated = await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
assert generated == 3 # 3 rows × 1 language
|
||
|
||
async def test_staggered_dates_two_per_day(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 2)
|
||
|
||
articles = await fetch_all("SELECT * FROM articles ORDER BY published_at")
|
||
assert len(articles) == 3
|
||
dates = [a["published_at"][:10] for a in articles]
|
||
# 2 on day 1, 1 on day 2
|
||
assert dates[0] == "2026-03-01"
|
||
assert dates[1] == "2026-03-01"
|
||
assert dates[2] == "2026-03-02"
|
||
|
||
async def test_staggered_dates_one_per_day(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 1)
|
||
|
||
articles = await fetch_all("SELECT * FROM articles ORDER BY published_at")
|
||
dates = sorted({a["published_at"][:10] for a in articles})
|
||
assert dates == ["2026-03-01", "2026-03-02", "2026-03-03"]
|
||
|
||
async def test_article_url_and_title(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
|
||
miami = await fetch_one("SELECT * FROM articles WHERE slug = 'test-city-en-miami'")
|
||
assert miami is not None
|
||
assert miami["url_path"] == "/en/markets/us/miami"
|
||
assert miami["title"] == "Padel in Miami"
|
||
assert miami["template_slug"] == "test-city"
|
||
assert miami["language"] == "en"
|
||
assert miami["status"] == "published"
|
||
|
||
async def test_scenario_created_per_row(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
|
||
scenarios = await fetch_all("SELECT * FROM published_scenarios")
|
||
assert len(scenarios) == 3
|
||
|
||
async def test_scenario_has_valid_calc_json(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
|
||
scenario = await fetch_one(
|
||
"SELECT * FROM published_scenarios WHERE slug = 'test-city-miami'"
|
||
)
|
||
assert scenario is not None
|
||
d = json.loads(scenario["calc_json"])
|
||
assert "capex" in d
|
||
assert "ebitdaMonth" in d
|
||
assert "irr" in d
|
||
assert d["capex"] > 0
|
||
|
||
async def test_build_files_written(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
|
||
build_dir = pseo_env["build_dir"]
|
||
articles = await fetch_all("SELECT slug FROM articles")
|
||
for a in articles:
|
||
build_path = build_dir / "en" / f"{a['slug']}.html"
|
||
assert build_path.exists(), f"Missing build file: {build_path}"
|
||
content = build_path.read_text()
|
||
assert len(content) > 50
|
||
|
||
async def test_updates_existing_on_regeneration(self, db, pseo_env):
|
||
"""Running generate twice updates articles, doesn't duplicate."""
|
||
from padelnomics.content import generate_articles
|
||
|
||
first = await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
assert first == 3
|
||
|
||
second = await generate_articles("test-city", date(2026, 3, 10), 10)
|
||
assert second == 3 # Updates existing
|
||
|
||
articles = await fetch_all("SELECT * FROM articles")
|
||
assert len(articles) == 3 # No duplicates
|
||
|
||
async def test_calc_overrides_applied(self, db, pseo_env):
|
||
"""Data row values that match DEFAULTS keys are used as calc overrides."""
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
|
||
# Miami had electricity=700, default is 600
|
||
scenario = await fetch_one(
|
||
"SELECT * FROM published_scenarios WHERE slug = 'test-city-miami'"
|
||
)
|
||
state = json.loads(scenario["state_json"])
|
||
assert state["electricity"] == 700
|
||
|
||
async def test_seo_head_populated(self, db, pseo_env):
|
||
from padelnomics.content import generate_articles
|
||
await generate_articles("test-city", date(2026, 3, 1), 10)
|
||
|
||
article = await fetch_one("SELECT * FROM articles WHERE slug = 'test-city-en-miami'")
|
||
assert article["seo_head"] is not None
|
||
assert 'rel="canonical"' in article["seo_head"]
|
||
assert 'application/ld+json' in article["seo_head"]
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Jinja string rendering
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestRenderPattern:
|
||
def test_simple(self):
|
||
from padelnomics.content import _render_pattern
|
||
assert _render_pattern("Hello {{ name }}!", {"name": "World"}) == "Hello World!"
|
||
|
||
def test_missing_var_empty(self):
|
||
from padelnomics.content import _render_pattern
|
||
result = _render_pattern("Hello {{ missing }}!", {})
|
||
assert result == "Hello !"
|
||
|
||
def test_url_pattern(self):
|
||
from padelnomics.content import _render_pattern
|
||
result = _render_pattern("/markets/{{ country | lower }}/{{ slug }}", {"country": "US", "slug": "miami"})
|
||
assert result == "/markets/us/miami"
|
||
|
||
def test_slugify_filter(self):
|
||
from padelnomics.content import _render_pattern
|
||
result = _render_pattern("{{ name | slugify }}", {"name": "Hello World"})
|
||
assert result == "hello-world"
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Template discovery & loading
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestDiscoverTemplates:
|
||
def test_discovers_templates(self, pseo_env):
|
||
from padelnomics.content import discover_templates
|
||
templates = discover_templates()
|
||
assert len(templates) == 1
|
||
assert templates[0]["slug"] == "test-city"
|
||
assert templates[0]["name"] == "Test City Analysis"
|
||
|
||
def test_empty_dir(self, tmp_path, monkeypatch):
|
||
import padelnomics.content as content_mod
|
||
monkeypatch.setattr(content_mod, "TEMPLATES_DIR", tmp_path / "nonexistent")
|
||
from padelnomics.content import discover_templates
|
||
assert discover_templates() == []
|
||
|
||
def test_skips_invalid_frontmatter(self, pseo_env):
|
||
from padelnomics.content import discover_templates
|
||
(pseo_env["tpl_dir"] / "bad.md.jinja").write_text("no frontmatter here")
|
||
templates = discover_templates()
|
||
assert len(templates) == 1 # Only the valid test-city template
|
||
|
||
def test_includes_path(self, pseo_env):
|
||
from padelnomics.content import discover_templates
|
||
templates = discover_templates()
|
||
assert "_path" in templates[0]
|
||
assert templates[0]["_path"].endswith("test-city.md.jinja")
|
||
|
||
|
||
class TestLoadTemplate:
|
||
def test_loads_config_and_body(self, pseo_env):
|
||
from padelnomics.content import load_template
|
||
config = load_template("test-city")
|
||
assert config["slug"] == "test-city"
|
||
assert config["content_type"] == "calculator"
|
||
assert config["data_table"] == "serving.test_cities"
|
||
assert "body_template" in config
|
||
assert "Padel in {{ city }}" in config["body_template"]
|
||
|
||
def test_missing_template_raises(self, pseo_env):
|
||
from padelnomics.content import load_template
|
||
with pytest.raises(AssertionError, match="Template not found"):
|
||
load_template("nonexistent")
|
||
|
||
def test_schema_type_normalized_to_list(self, pseo_env):
|
||
from padelnomics.content import load_template
|
||
config = load_template("test-city")
|
||
assert isinstance(config["schema_type"], list)
|
||
assert "Article" in config["schema_type"]
|
||
|
||
def test_languages_parsed(self, pseo_env):
|
||
from padelnomics.content import load_template
|
||
config = load_template("test-city")
|
||
assert config["languages"] == ["en"]
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# FAQ extraction
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestExtractFaqPairs:
|
||
def test_extracts_pairs(self):
|
||
from padelnomics.content import _extract_faq_pairs
|
||
md = (
|
||
"# Title\n\n"
|
||
"## FAQ\n\n"
|
||
"**How much does it cost?**\n"
|
||
"It costs about 500k.\n\n"
|
||
"**How long does it take?**\n"
|
||
"About 12 months.\n\n"
|
||
"## Other Section\n"
|
||
)
|
||
pairs = _extract_faq_pairs(md)
|
||
assert len(pairs) == 2
|
||
assert pairs[0]["question"] == "How much does it cost?"
|
||
assert "500k" in pairs[0]["answer"]
|
||
assert pairs[1]["question"] == "How long does it take?"
|
||
|
||
def test_no_faq_section(self):
|
||
from padelnomics.content import _extract_faq_pairs
|
||
assert _extract_faq_pairs("# Title\n\nSome content") == []
|
||
|
||
def test_faq_at_end_of_document(self):
|
||
from padelnomics.content import _extract_faq_pairs
|
||
md = "## FAQ\n\n**Question one?**\nAnswer one.\n"
|
||
pairs = _extract_faq_pairs(md)
|
||
assert len(pairs) == 1
|
||
assert pairs[0]["question"] == "Question one?"
|
||
assert pairs[0]["answer"] == "Answer one."
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Breadcrumbs
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestBuildBreadcrumbs:
|
||
def test_basic_path(self):
|
||
from padelnomics.content import _build_breadcrumbs
|
||
crumbs = _build_breadcrumbs("/en/markets/germany/berlin", "https://padelnomics.io")
|
||
assert len(crumbs) == 5
|
||
assert crumbs[0] == {"name": "Home", "url": "https://padelnomics.io/"}
|
||
assert crumbs[1] == {"name": "En", "url": "https://padelnomics.io/en"}
|
||
assert crumbs[2] == {"name": "Markets", "url": "https://padelnomics.io/en/markets"}
|
||
assert crumbs[3] == {"name": "Germany", "url": "https://padelnomics.io/en/markets/germany"}
|
||
assert crumbs[4] == {"name": "Berlin", "url": "https://padelnomics.io/en/markets/germany/berlin"}
|
||
|
||
def test_root_path(self):
|
||
from padelnomics.content import _build_breadcrumbs
|
||
crumbs = _build_breadcrumbs("/", "https://padelnomics.io")
|
||
assert len(crumbs) == 1
|
||
assert crumbs[0]["name"] == "Home"
|
||
|
||
def test_hyphenated_segments_titlecased(self):
|
||
from padelnomics.content import _build_breadcrumbs
|
||
crumbs = _build_breadcrumbs("/en/my-section", "https://padelnomics.io")
|
||
assert crumbs[2]["name"] == "My Section"
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# JSON-LD structured data
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestBuildJsonld:
|
||
_COMMON = dict(
|
||
title="Test Title",
|
||
description="Test description",
|
||
url="https://padelnomics.io/en/markets/us/miami",
|
||
published_at="2026-01-01T08:00:00",
|
||
date_modified="2026-01-02T10:00:00",
|
||
language="en",
|
||
breadcrumbs=[
|
||
{"name": "Home", "url": "https://padelnomics.io/"},
|
||
{"name": "Markets", "url": "https://padelnomics.io/en/markets"},
|
||
],
|
||
)
|
||
|
||
def test_always_includes_breadcrumbs(self):
|
||
from padelnomics.content import build_jsonld
|
||
objects = build_jsonld(["Article"], **self._COMMON)
|
||
types = [o["@type"] for o in objects]
|
||
assert "BreadcrumbList" in types
|
||
|
||
def test_breadcrumb_positions(self):
|
||
from padelnomics.content import build_jsonld
|
||
objects = build_jsonld(["Article"], **self._COMMON)
|
||
bc = [o for o in objects if o["@type"] == "BreadcrumbList"][0]
|
||
items = bc["itemListElement"]
|
||
assert items[0]["position"] == 1
|
||
assert items[0]["name"] == "Home"
|
||
assert items[1]["position"] == 2
|
||
|
||
def test_article_schema(self):
|
||
from padelnomics.content import build_jsonld
|
||
objects = build_jsonld(["Article"], **self._COMMON)
|
||
article = [o for o in objects if o["@type"] == "Article"][0]
|
||
assert article["headline"] == "Test Title"
|
||
assert article["inLanguage"] == "en"
|
||
assert article["datePublished"] == "2026-01-01T08:00:00"
|
||
assert article["dateModified"] == "2026-01-02T10:00:00"
|
||
assert article["publisher"]["name"] == "Padelnomics"
|
||
|
||
def test_headline_truncated_at_110(self):
|
||
from padelnomics.content import build_jsonld
|
||
long_title = "A" * 200
|
||
objects = build_jsonld(["Article"], **{**self._COMMON, "title": long_title})
|
||
article = [o for o in objects if o["@type"] == "Article"][0]
|
||
assert len(article["headline"]) == 110
|
||
|
||
def test_faqpage_schema(self):
|
||
from padelnomics.content import build_jsonld
|
||
faq_pairs = [
|
||
{"question": "How much?", "answer": "About 500k."},
|
||
{"question": "How long?", "answer": "12 months."},
|
||
]
|
||
objects = build_jsonld(["Article", "FAQPage"], **self._COMMON, faq_pairs=faq_pairs)
|
||
faq = [o for o in objects if o["@type"] == "FAQPage"][0]
|
||
assert len(faq["mainEntity"]) == 2
|
||
assert faq["mainEntity"][0]["name"] == "How much?"
|
||
assert faq["mainEntity"][0]["acceptedAnswer"]["text"] == "About 500k."
|
||
|
||
def test_faqpage_omitted_without_pairs(self):
|
||
from padelnomics.content import build_jsonld
|
||
objects = build_jsonld(["FAQPage"], **self._COMMON, faq_pairs=[])
|
||
types = [o["@type"] for o in objects]
|
||
assert "FAQPage" not in types
|
||
|
||
def test_no_article_when_not_in_types(self):
|
||
from padelnomics.content import build_jsonld
|
||
faq_pairs = [{"question": "Q?", "answer": "A."}]
|
||
objects = build_jsonld(["FAQPage"], **self._COMMON, faq_pairs=faq_pairs)
|
||
types = [o["@type"] for o in objects]
|
||
assert "Article" not in types
|
||
assert "FAQPage" in types
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Preview article
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestPreviewArticle:
|
||
async def test_preview_returns_rendered_data(self, db, pseo_env):
|
||
from padelnomics.content import preview_article
|
||
result = await preview_article("test-city", "miami")
|
||
assert result["title"] == "Padel in Miami"
|
||
assert result["url_path"] == "/en/markets/us/miami"
|
||
assert result["meta_description"] == "Padel costs in Miami"
|
||
assert "<h1>" in result["html"]
|
||
|
||
async def test_preview_unknown_row_raises(self, db, pseo_env):
|
||
from padelnomics.content import preview_article
|
||
with pytest.raises(AssertionError, match="No row found"):
|
||
await preview_article("test-city", "nonexistent")
|
||
|
||
async def test_preview_with_language(self, db, pseo_env):
|
||
from padelnomics.content import preview_article
|
||
result = await preview_article("test-city", "miami", lang="de")
|
||
assert result["url_path"] == "/de/markets/us/miami"
|
||
|
||
async def test_preview_unknown_template_raises(self, db, pseo_env):
|
||
from padelnomics.content import preview_article
|
||
with pytest.raises(AssertionError, match="Template not found"):
|
||
await preview_article("nonexistent", "miami")
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Public routes
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestMarketsHub:
|
||
async def test_markets_returns_200(self, client):
|
||
resp = await client.get("/en/markets")
|
||
assert resp.status_code == 200
|
||
|
||
async def test_markets_has_search(self, client):
|
||
resp = await client.get("/en/markets")
|
||
html = (await resp.data).decode()
|
||
assert 'id="market-q"' in html
|
||
|
||
async def test_markets_results_partial(self, client):
|
||
resp = await client.get("/en/markets/results")
|
||
assert resp.status_code == 200
|
||
|
||
async def test_markets_shows_published_articles(self, client, db):
|
||
await _create_article(slug="pub-test", url_path="/pub-test", status="published")
|
||
resp = await client.get("/en/markets")
|
||
html = (await resp.data).decode()
|
||
assert "Title pub-test" in html
|
||
|
||
async def test_markets_hides_draft_articles(self, client, db):
|
||
await _create_article(slug="draft-test", url_path="/draft-test", status="draft")
|
||
resp = await client.get("/en/markets")
|
||
html = (await resp.data).decode()
|
||
assert "Title draft-test" not in html
|
||
|
||
async def test_markets_hides_future_articles(self, client, db):
|
||
await _create_article(
|
||
slug="future-test", url_path="/future-test",
|
||
status="published", published_at="2099-01-01T00:00:00",
|
||
)
|
||
resp = await client.get("/en/markets")
|
||
html = (await resp.data).decode()
|
||
assert "Title future-test" not in html
|
||
|
||
async def test_markets_filter_by_country(self, client, db):
|
||
await _create_article(slug="us-art", url_path="/us-art")
|
||
resp = await client.get("/en/markets/results?country=US")
|
||
html = (await resp.data).decode()
|
||
assert "Title us-art" in html
|
||
|
||
resp = await client.get("/en/markets/results?country=DE")
|
||
html = (await resp.data).decode()
|
||
assert "Title us-art" not in html
|
||
|
||
|
||
class TestArticleServing:
|
||
async def test_nonexistent_article_returns_404(self, client):
|
||
resp = await client.get("/en/padel-court-cost-nonexistent")
|
||
assert resp.status_code == 404
|
||
|
||
async def test_draft_article_returns_404(self, client, db):
|
||
await _create_article(slug="draft-serve", url_path="/draft-serve", status="draft")
|
||
resp = await client.get("/en/draft-serve")
|
||
assert resp.status_code == 404
|
||
|
||
async def test_future_article_returns_404(self, client, db):
|
||
await _create_article(
|
||
slug="future-serve", url_path="/future-serve",
|
||
status="published", published_at="2099-01-01T00:00:00",
|
||
)
|
||
resp = await client.get("/en/future-serve")
|
||
assert resp.status_code == 404
|
||
|
||
async def test_published_article_served(self, client, db):
|
||
from padelnomics.content.routes import BUILD_DIR
|
||
await _create_article(slug="live-art", url_path="/live-art")
|
||
|
||
# Write a build file
|
||
BUILD_DIR.mkdir(parents=True, exist_ok=True)
|
||
build_path = BUILD_DIR / "live-art.html"
|
||
build_path.write_text("<p>Article body content</p>")
|
||
|
||
try:
|
||
resp = await client.get("/en/live-art")
|
||
assert resp.status_code == 200
|
||
html = (await resp.data).decode()
|
||
assert "Article body content" in html
|
||
assert "Title live-art" in html
|
||
finally:
|
||
if build_path.exists():
|
||
build_path.unlink()
|
||
|
||
async def test_article_missing_build_file_returns_404(self, client, db):
|
||
await _create_article(slug="no-build", url_path="/no-build")
|
||
resp = await client.get("/en/no-build")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Sitemap integration
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestSitemapContent:
|
||
async def test_sitemap_includes_markets(self, client):
|
||
resp = await client.get("/sitemap.xml")
|
||
xml = (await resp.data).decode()
|
||
assert "/markets" in xml
|
||
|
||
async def test_sitemap_includes_published_articles(self, client, db):
|
||
await _create_article(slug="sitemap-art", url_path="/sitemap-art")
|
||
resp = await client.get("/sitemap.xml")
|
||
xml = (await resp.data).decode()
|
||
assert "/sitemap-art" in xml
|
||
|
||
async def test_sitemap_excludes_draft_articles(self, client, db):
|
||
await _create_article(slug="sitemap-draft", url_path="/sitemap-draft", status="draft")
|
||
resp = await client.get("/sitemap.xml")
|
||
xml = (await resp.data).decode()
|
||
assert "/sitemap-draft" not in xml
|
||
|
||
async def test_sitemap_excludes_future_articles(self, client, db):
|
||
await _create_article(
|
||
slug="sitemap-future", url_path="/sitemap-future",
|
||
status="published", published_at="2099-01-01T00:00:00",
|
||
)
|
||
resp = await client.get("/sitemap.xml")
|
||
xml = (await resp.data).decode()
|
||
assert "/sitemap-future" not in xml
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Route registration
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestRouteRegistration:
|
||
def test_markets_route_registered(self, app):
|
||
rules = [r.rule for r in app.url_map.iter_rules()]
|
||
assert "/<lang>/markets" in rules
|
||
|
||
def test_admin_content_routes_registered(self, app):
|
||
rules = [r.rule for r in app.url_map.iter_rules()]
|
||
assert "/admin/templates" in rules
|
||
assert "/admin/scenarios" in rules
|
||
assert "/admin/articles" in rules
|
||
assert "/admin/scenarios/new" in rules
|
||
assert "/admin/rebuild-all" in rules
|
||
|
||
def test_catchall_route_registered(self, app):
|
||
rules = [r.rule for r in app.url_map.iter_rules()]
|
||
assert "/<lang>/<path:url_path>" in rules
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Admin routes (require admin session)
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
@pytest.fixture
|
||
async def admin_client(app, db):
|
||
"""Test client with admin user (has admin role)."""
|
||
from datetime import datetime
|
||
now = datetime.utcnow().isoformat()
|
||
async with db.execute(
|
||
"INSERT INTO users (email, name, created_at) VALUES (?, ?, ?)",
|
||
("admin@test.com", "Admin", now),
|
||
) as cursor:
|
||
admin_id = cursor.lastrowid
|
||
await db.execute(
|
||
"INSERT INTO user_roles (user_id, role) VALUES (?, 'admin')", (admin_id,)
|
||
)
|
||
await db.commit()
|
||
|
||
async with app.test_client() as c:
|
||
async with c.session_transaction() as sess:
|
||
sess["user_id"] = admin_id
|
||
yield c
|
||
|
||
|
||
class TestAdminTemplates:
|
||
async def test_template_list_requires_admin(self, client):
|
||
resp = await client.get("/admin/templates")
|
||
assert resp.status_code == 302
|
||
|
||
async def test_template_list(self, admin_client):
|
||
resp = await admin_client.get("/admin/templates")
|
||
assert resp.status_code == 200
|
||
|
||
async def test_template_list_shows_discovered(self, admin_client, pseo_env):
|
||
resp = await admin_client.get("/admin/templates")
|
||
assert resp.status_code == 200
|
||
html = (await resp.data).decode()
|
||
assert "Test City Analysis" in html
|
||
assert "test-city" in html
|
||
|
||
|
||
class TestAdminTemplateDetail:
|
||
async def test_detail_shows_config(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/test-city")
|
||
assert resp.status_code == 200
|
||
html = (await resp.data).decode()
|
||
assert "Test City Analysis" in html
|
||
assert "serving.test_cities" in html
|
||
|
||
async def test_detail_shows_columns(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/test-city")
|
||
html = (await resp.data).decode()
|
||
assert "city_slug" in html
|
||
assert "VARCHAR" in html
|
||
|
||
async def test_detail_shows_sample_data(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/test-city")
|
||
html = (await resp.data).decode()
|
||
assert "Miami" in html
|
||
assert "Berlin" in html
|
||
|
||
async def test_detail_unknown_slug_redirects(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/nonexistent")
|
||
assert resp.status_code == 302
|
||
|
||
|
||
class TestAdminTemplatePreview:
|
||
async def test_preview_renders_article(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/test-city/preview/miami")
|
||
assert resp.status_code == 200
|
||
html = (await resp.data).decode()
|
||
assert "Padel in Miami" in html
|
||
|
||
async def test_preview_bad_key_redirects(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/test-city/preview/nonexistent")
|
||
assert resp.status_code == 302
|
||
|
||
async def test_preview_bad_template_redirects(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/bad-slug/preview/miami")
|
||
assert resp.status_code == 302
|
||
|
||
|
||
class TestAdminTemplateGenerate:
|
||
async def test_generate_form(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/test-city/generate")
|
||
assert resp.status_code == 200
|
||
html = (await resp.data).decode()
|
||
assert "3" in html # 3 rows available
|
||
assert "Generate" in html
|
||
|
||
async def test_generate_enqueues_task(self, admin_client, db, pseo_env):
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post("/admin/templates/test-city/generate", form={
|
||
"csrf_token": "test",
|
||
"start_date": "2026-04-01",
|
||
"articles_per_day": "2",
|
||
})
|
||
assert resp.status_code == 302
|
||
|
||
# Generation is now queued, not inline
|
||
tasks = await fetch_all(
|
||
"SELECT * FROM tasks WHERE task_name = 'generate_articles'"
|
||
)
|
||
assert len(tasks) == 1
|
||
import json
|
||
payload = json.loads(tasks[0]["payload"])
|
||
assert payload["template_slug"] == "test-city"
|
||
assert payload["start_date"] == "2026-04-01"
|
||
assert payload["articles_per_day"] == 2
|
||
|
||
async def test_generate_unknown_template_redirects(self, admin_client, db, pseo_env):
|
||
resp = await admin_client.get("/admin/templates/nonexistent/generate")
|
||
assert resp.status_code == 302
|
||
|
||
|
||
class TestAdminTemplateRegenerate:
|
||
async def test_regenerate_enqueues_task(self, admin_client, db, pseo_env):
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post("/admin/templates/test-city/regenerate", form={
|
||
"csrf_token": "test",
|
||
})
|
||
assert resp.status_code == 302
|
||
|
||
# Regeneration is now queued, not inline
|
||
tasks = await fetch_all(
|
||
"SELECT * FROM tasks WHERE task_name = 'generate_articles'"
|
||
)
|
||
assert len(tasks) == 1
|
||
import json
|
||
payload = json.loads(tasks[0]["payload"])
|
||
assert payload["template_slug"] == "test-city"
|
||
|
||
async def test_regenerate_unknown_template_redirects(self, admin_client, db, pseo_env):
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post("/admin/templates/nonexistent/regenerate", form={
|
||
"csrf_token": "test",
|
||
})
|
||
assert resp.status_code == 302
|
||
|
||
|
||
class TestAdminScenarios:
|
||
async def test_scenario_list(self, admin_client):
|
||
resp = await admin_client.get("/admin/scenarios")
|
||
assert resp.status_code == 200
|
||
|
||
async def test_scenario_new_form(self, admin_client):
|
||
resp = await admin_client.get("/admin/scenarios/new")
|
||
assert resp.status_code == 200
|
||
|
||
async def test_scenario_create(self, admin_client, db):
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post("/admin/scenarios/new", form={
|
||
"csrf_token": "test",
|
||
"title": "Miami Test",
|
||
"slug": "miami-test",
|
||
"location": "Miami",
|
||
"country": "US",
|
||
"venue": "indoor",
|
||
"own": "rent",
|
||
"dblCourts": "4",
|
||
"sglCourts": "2",
|
||
})
|
||
assert resp.status_code == 302
|
||
|
||
row = await fetch_one("SELECT * FROM published_scenarios WHERE slug = 'miami-test'")
|
||
assert row is not None
|
||
assert row["title"] == "Miami Test"
|
||
d = json.loads(row["calc_json"])
|
||
assert d["capex"] > 0
|
||
|
||
async def test_scenario_edit_recalculates(self, admin_client, db):
|
||
scenario_id = await _create_published_scenario(slug="edit-sc")
|
||
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post(f"/admin/scenarios/{scenario_id}/edit", form={
|
||
"csrf_token": "test",
|
||
"title": "Updated",
|
||
"location": "TestCity",
|
||
"country": "TC",
|
||
"dblCourts": "6",
|
||
"sglCourts": "0",
|
||
})
|
||
assert resp.status_code == 302
|
||
|
||
row = await fetch_one("SELECT * FROM published_scenarios WHERE id = ?", (scenario_id,))
|
||
assert row["title"] == "Updated"
|
||
state = json.loads(row["state_json"])
|
||
assert state["dblCourts"] == 6
|
||
|
||
async def test_scenario_delete(self, admin_client, db):
|
||
scenario_id = await _create_published_scenario(slug="del-sc")
|
||
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post(f"/admin/scenarios/{scenario_id}/delete", form={
|
||
"csrf_token": "test",
|
||
})
|
||
assert resp.status_code == 302
|
||
assert await fetch_one("SELECT 1 FROM published_scenarios WHERE id = ?", (scenario_id,)) is None
|
||
|
||
async def test_scenario_preview(self, admin_client, db):
|
||
scenario_id = await _create_published_scenario(slug="preview-sc")
|
||
resp = await admin_client.get(f"/admin/scenarios/{scenario_id}/preview")
|
||
assert resp.status_code == 200
|
||
html = (await resp.data).decode()
|
||
assert "scenario-widget" in html
|
||
|
||
|
||
class TestAdminArticles:
|
||
async def test_article_list(self, admin_client):
|
||
resp = await admin_client.get("/admin/articles")
|
||
assert resp.status_code == 200
|
||
|
||
async def test_article_new_form(self, admin_client):
|
||
resp = await admin_client.get("/admin/articles/new")
|
||
assert resp.status_code == 200
|
||
|
||
async def test_article_create_manual(self, admin_client, db):
|
||
from padelnomics.content.routes import BUILD_DIR
|
||
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post("/admin/articles/new", form={
|
||
"csrf_token": "test",
|
||
"title": "Manual Article",
|
||
"slug": "manual-art",
|
||
"url_path": "/manual-art",
|
||
"meta_description": "A test article",
|
||
"country": "US",
|
||
"region": "North America",
|
||
"body": "# Hello\n\nThis is a test.",
|
||
"status": "published",
|
||
})
|
||
assert resp.status_code == 302
|
||
|
||
row = await fetch_one("SELECT * FROM articles WHERE slug = 'manual-art'")
|
||
assert row is not None
|
||
assert row["title"] == "Manual Article"
|
||
|
||
# Build file created
|
||
build_path = BUILD_DIR / "manual-art.html"
|
||
try:
|
||
assert build_path.exists()
|
||
content = build_path.read_text()
|
||
assert "<h1>Hello</h1>" in content
|
||
finally:
|
||
if build_path.exists():
|
||
build_path.unlink()
|
||
md_path = Path("data/content/articles/manual-art.md")
|
||
if md_path.exists():
|
||
md_path.unlink()
|
||
|
||
async def test_article_reserved_path_rejected(self, admin_client, db):
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post("/admin/articles/new", form={
|
||
"csrf_token": "test",
|
||
"title": "Bad Path",
|
||
"slug": "bad-path",
|
||
"url_path": "/admin/evil",
|
||
"body": "# Nope",
|
||
"status": "draft",
|
||
})
|
||
# Should re-render form with error, not redirect
|
||
assert resp.status_code == 200
|
||
html = (await resp.data).decode()
|
||
assert "reserved" in html.lower() or "conflict" in html.lower()
|
||
|
||
async def test_article_publish_toggle(self, admin_client, db):
|
||
article_id = await _create_article(slug="toggle-art", url_path="/toggle-art", status="draft")
|
||
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post(f"/admin/articles/{article_id}/publish", form={
|
||
"csrf_token": "test",
|
||
})
|
||
assert resp.status_code == 302
|
||
|
||
row = await fetch_one("SELECT status FROM articles WHERE id = ?", (article_id,))
|
||
assert row["status"] == "published"
|
||
|
||
# Toggle back
|
||
resp = await admin_client.post(f"/admin/articles/{article_id}/publish", form={
|
||
"csrf_token": "test",
|
||
})
|
||
row = await fetch_one("SELECT status FROM articles WHERE id = ?", (article_id,))
|
||
assert row["status"] == "draft"
|
||
|
||
async def test_article_delete(self, admin_client, db):
|
||
article_id = await _create_article(slug="del-art", url_path="/del-art")
|
||
|
||
async with admin_client.session_transaction() as sess:
|
||
sess["csrf_token"] = "test"
|
||
|
||
resp = await admin_client.post(f"/admin/articles/{article_id}/delete", form={
|
||
"csrf_token": "test",
|
||
})
|
||
assert resp.status_code == 302
|
||
assert await fetch_one("SELECT 1 FROM articles WHERE id = ?", (article_id,)) is None
|
||
|
||
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Admin dashboard quick links
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestAdminDashboardLinks:
|
||
async def test_dashboard_has_content_links(self, admin_client):
|
||
resp = await admin_client.get("/admin/")
|
||
html = (await resp.data).decode()
|
||
assert "/admin/templates" in html
|
||
assert "/admin/scenarios" in html
|
||
assert "/admin/articles" in html
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════
|
||
# Footer
|
||
# ════════════════════════════════════════════════════════════
|
||
|
||
class TestFooterMarkets:
|
||
async def test_footer_has_markets_link(self, client):
|
||
resp = await client.get("/en/")
|
||
html = (await resp.data).decode()
|
||
assert "/markets" in html
|