feat(affiliate): add program + URL assembly tests; update CHANGELOG + PROJECT.md
41 tests total (+15). New coverage: get_all_programs(), get_program(), get_program_by_slug(), build_affiliate_url() (program path, legacy fallback, no program_id, no program dict), program-based redirect, legacy redirect, migration seed assertion, ASIN backfill assertion. All ruff checks pass. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
@@ -2,7 +2,8 @@
|
||||
Tests for the affiliate product system.
|
||||
|
||||
Covers: hash_ip determinism, product CRUD, bake_product_cards marker replacement,
|
||||
click redirect (302 + logged), rate limiting, inactive product 404, multi-retailer.
|
||||
click redirect (302 + logged), rate limiting, inactive product 404, multi-retailer,
|
||||
program CRUD, build_affiliate_url(), program-based redirect.
|
||||
"""
|
||||
import json
|
||||
from datetime import date
|
||||
@@ -10,11 +11,15 @@ from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from padelnomics.affiliate import (
|
||||
build_affiliate_url,
|
||||
get_all_products,
|
||||
get_all_programs,
|
||||
get_click_counts,
|
||||
get_click_stats,
|
||||
get_product,
|
||||
get_products_by_category,
|
||||
get_program,
|
||||
get_program_by_slug,
|
||||
hash_ip,
|
||||
log_click,
|
||||
)
|
||||
@@ -330,3 +335,282 @@ async def test_affiliate_redirect_unknown_404(app, db):
|
||||
async with app.test_client() as client:
|
||||
response = await client.get("/go/totally-unknown-xyz")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ── affiliate_programs ────────────────────────────────────────────────────────
|
||||
|
||||
async def _insert_program(
|
||||
name="Test Shop",
|
||||
slug="test-shop",
|
||||
url_template="https://testshop.example.com/p/{product_id}?ref={tag}",
|
||||
tracking_tag="testref",
|
||||
commission_pct=5.0,
|
||||
homepage_url="https://testshop.example.com",
|
||||
status="active",
|
||||
) -> int:
|
||||
"""Insert an affiliate program, return its id."""
|
||||
return await execute(
|
||||
"""INSERT INTO affiliate_programs
|
||||
(name, slug, url_template, tracking_tag, commission_pct, homepage_url, status)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||||
(name, slug, url_template, tracking_tag, commission_pct, homepage_url, status),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_get_all_programs_returns_all(db):
|
||||
"""get_all_programs returns inserted programs sorted by name."""
|
||||
await _insert_program(slug="zebra-shop", name="Zebra Shop")
|
||||
await _insert_program(slug="alpha-shop", name="Alpha Shop")
|
||||
programs = await get_all_programs()
|
||||
names = [p["name"] for p in programs]
|
||||
assert "Alpha Shop" in names
|
||||
assert "Zebra Shop" in names
|
||||
# Sorted by name ascending
|
||||
assert names.index("Alpha Shop") < names.index("Zebra Shop")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_get_all_programs_status_filter(db):
|
||||
"""get_all_programs(status='active') excludes inactive programs."""
|
||||
await _insert_program(slug="inactive-prog", status="inactive")
|
||||
await _insert_program(slug="active-prog", name="Active Shop")
|
||||
active = await get_all_programs(status="active")
|
||||
statuses = [p["status"] for p in active]
|
||||
assert all(s == "active" for s in statuses)
|
||||
slugs = [p["slug"] for p in active]
|
||||
assert "inactive-prog" not in slugs
|
||||
assert "active-prog" in slugs
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_get_program_by_id(db):
|
||||
"""get_program returns a program by id."""
|
||||
prog_id = await _insert_program()
|
||||
prog = await get_program(prog_id)
|
||||
assert prog is not None
|
||||
assert prog["slug"] == "test-shop"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_get_program_not_found(db):
|
||||
"""get_program returns None for unknown id."""
|
||||
prog = await get_program(99999)
|
||||
assert prog is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_get_program_by_slug(db):
|
||||
"""get_program_by_slug returns the program for a known slug."""
|
||||
await _insert_program(slug="find-by-slug")
|
||||
prog = await get_program_by_slug("find-by-slug")
|
||||
assert prog is not None
|
||||
assert prog["name"] == "Test Shop"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_get_program_by_slug_not_found(db):
|
||||
"""get_program_by_slug returns None for unknown slug."""
|
||||
prog = await get_program_by_slug("nonexistent-slug-xyz")
|
||||
assert prog is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_get_all_programs_product_count(db):
|
||||
"""get_all_programs includes product_count for each program."""
|
||||
prog_id = await _insert_program(slug="counted-prog")
|
||||
await _insert_product(slug="p-for-count", program_id=prog_id)
|
||||
programs = await get_all_programs()
|
||||
prog = next(p for p in programs if p["slug"] == "counted-prog")
|
||||
assert prog["product_count"] == 1
|
||||
|
||||
|
||||
# ── build_affiliate_url ───────────────────────────────────────────────────────
|
||||
|
||||
def test_build_affiliate_url_with_program():
|
||||
"""build_affiliate_url assembles URL from program template."""
|
||||
product = {"program_id": 1, "product_identifier": "B0TESTTEST", "affiliate_url": ""}
|
||||
program = {"url_template": "https://amazon.de/dp/{product_id}?tag={tag}", "tracking_tag": "mysite-21"}
|
||||
url = build_affiliate_url(product, program)
|
||||
assert url == "https://amazon.de/dp/B0TESTTEST?tag=mysite-21"
|
||||
|
||||
|
||||
def test_build_affiliate_url_legacy_fallback():
|
||||
"""build_affiliate_url falls back to baked affiliate_url when no program."""
|
||||
product = {"program_id": None, "product_identifier": "", "affiliate_url": "https://baked.example.com/p?tag=x"}
|
||||
url = build_affiliate_url(product, None)
|
||||
assert url == "https://baked.example.com/p?tag=x"
|
||||
|
||||
|
||||
def test_build_affiliate_url_no_program_id():
|
||||
"""build_affiliate_url uses fallback when program_id is 0/falsy."""
|
||||
product = {"program_id": 0, "product_identifier": "B0IGNORED", "affiliate_url": "https://fallback.example.com"}
|
||||
program = {"url_template": "https://shop.example.com/{product_id}?ref={tag}", "tracking_tag": "tag123"}
|
||||
url = build_affiliate_url(product, program)
|
||||
# program_id is falsy → fallback
|
||||
assert url == "https://fallback.example.com"
|
||||
|
||||
|
||||
def test_build_affiliate_url_no_program_dict():
|
||||
"""build_affiliate_url uses fallback when program dict is None."""
|
||||
product = {"program_id": 5, "product_identifier": "ASIN123", "affiliate_url": "https://fallback.example.com"}
|
||||
url = build_affiliate_url(product, None)
|
||||
assert url == "https://fallback.example.com"
|
||||
|
||||
|
||||
# ── program-based redirect ────────────────────────────────────────────────────
|
||||
|
||||
async def _insert_product( # noqa: F811 — redefined to add program_id support
|
||||
slug="test-racket-amazon",
|
||||
name="Test Racket",
|
||||
brand="TestBrand",
|
||||
category="racket",
|
||||
retailer="Amazon",
|
||||
affiliate_url="https://amazon.de/dp/TEST?tag=test-21",
|
||||
status="active",
|
||||
language="de",
|
||||
price_cents=14999,
|
||||
pros=None,
|
||||
cons=None,
|
||||
sort_order=0,
|
||||
program_id=None,
|
||||
product_identifier="",
|
||||
) -> int:
|
||||
"""Insert an affiliate product with optional program_id, return its id."""
|
||||
return await execute(
|
||||
"""INSERT INTO affiliate_products
|
||||
(slug, name, brand, category, retailer, affiliate_url,
|
||||
price_cents, currency, status, language, pros, cons, sort_order,
|
||||
program_id, product_identifier)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, 'EUR', ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(
|
||||
slug, name, brand, category, retailer, affiliate_url,
|
||||
price_cents, status, language,
|
||||
json.dumps(pros or ["Gut"]),
|
||||
json.dumps(cons or ["Teuer"]),
|
||||
sort_order,
|
||||
program_id,
|
||||
product_identifier,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_affiliate_redirect_uses_program_url(app, db):
|
||||
"""Redirect assembles URL from program template when product has program_id."""
|
||||
prog_id = await _insert_program(
|
||||
slug="amzn-test",
|
||||
url_template="https://www.amazon.de/dp/{product_id}?tag={tag}",
|
||||
tracking_tag="testsite-21",
|
||||
)
|
||||
await _insert_product(
|
||||
slug="program-redirect-test",
|
||||
affiliate_url="",
|
||||
program_id=prog_id,
|
||||
product_identifier="B0PROGRAM01",
|
||||
)
|
||||
async with app.test_client() as client:
|
||||
response = await client.get("/go/program-redirect-test")
|
||||
assert response.status_code == 302
|
||||
location = response.headers.get("Location", "")
|
||||
assert "B0PROGRAM01" in location
|
||||
assert "testsite-21" in location
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("db")
|
||||
async def test_affiliate_redirect_legacy_url_still_works(app, db):
|
||||
"""Legacy products with baked affiliate_url still redirect correctly."""
|
||||
await _insert_product(
|
||||
slug="legacy-redirect-test",
|
||||
affiliate_url="https://amazon.de/dp/LEGACY?tag=old-21",
|
||||
program_id=None,
|
||||
product_identifier="",
|
||||
)
|
||||
async with app.test_client() as client:
|
||||
response = await client.get("/go/legacy-redirect-test")
|
||||
assert response.status_code == 302
|
||||
assert "LEGACY" in response.headers.get("Location", "")
|
||||
|
||||
|
||||
# ── migration backfill ────────────────────────────────────────────────────────
|
||||
|
||||
def _load_migration_0027():
|
||||
"""Import migration 0027 via importlib (filename starts with a digit)."""
|
||||
import importlib
|
||||
from pathlib import Path
|
||||
|
||||
versions_dir = Path(__file__).parent.parent / "src/padelnomics/migrations/versions"
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"migration_0027", versions_dir / "0027_affiliate_programs.py"
|
||||
)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
def _make_pre_migration_db():
|
||||
"""Create a minimal sqlite3 DB simulating state just before migration 0027.
|
||||
|
||||
Provides the affiliate_products table (migration ALTERs it), but not
|
||||
affiliate_programs (migration CREATEs it).
|
||||
"""
|
||||
import sqlite3
|
||||
|
||||
conn = sqlite3.connect(":memory:")
|
||||
conn.execute("""
|
||||
CREATE TABLE affiliate_products (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
slug TEXT NOT NULL,
|
||||
name TEXT NOT NULL DEFAULT '',
|
||||
affiliate_url TEXT NOT NULL DEFAULT '',
|
||||
UNIQUE(slug)
|
||||
)
|
||||
""")
|
||||
conn.commit()
|
||||
return conn
|
||||
|
||||
|
||||
def test_migration_seeds_amazon_program():
|
||||
"""Migration 0027 up() seeds the Amazon program with expected fields.
|
||||
|
||||
Tests the migration function directly against a real sqlite3 DB
|
||||
(the conftest only replays CREATE TABLE DDL, not INSERT seeds).
|
||||
"""
|
||||
migration = _load_migration_0027()
|
||||
conn = _make_pre_migration_db()
|
||||
migration.up(conn)
|
||||
conn.commit()
|
||||
|
||||
row = conn.execute(
|
||||
"SELECT * FROM affiliate_programs WHERE slug = 'amazon'"
|
||||
).fetchone()
|
||||
assert row is not None
|
||||
cols = [d[0] for d in conn.execute("SELECT * FROM affiliate_programs WHERE slug = 'amazon'").description]
|
||||
prog = dict(zip(cols, row))
|
||||
assert prog["name"] == "Amazon"
|
||||
assert "padelnomics-21" in prog["tracking_tag"]
|
||||
assert "{product_id}" in prog["url_template"]
|
||||
assert "{tag}" in prog["url_template"]
|
||||
assert prog["commission_pct"] == 3.0
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_migration_backfills_asin_from_url():
|
||||
"""Migration 0027 up() extracts ASINs from existing affiliate_url values."""
|
||||
migration = _load_migration_0027()
|
||||
conn = _make_pre_migration_db()
|
||||
conn.execute(
|
||||
"INSERT INTO affiliate_products (slug, affiliate_url) VALUES (?, ?)",
|
||||
("test-racket", "https://www.amazon.de/dp/B0ASIN1234?tag=test-21"),
|
||||
)
|
||||
conn.commit()
|
||||
migration.up(conn)
|
||||
conn.commit()
|
||||
|
||||
row = conn.execute(
|
||||
"SELECT program_id, product_identifier FROM affiliate_products WHERE slug = 'test-racket'"
|
||||
).fetchone()
|
||||
assert row is not None
|
||||
assert row[0] is not None # program_id set
|
||||
assert row[1] == "B0ASIN1234" # ASIN extracted correctly
|
||||
conn.close()
|
||||
|
||||
Reference in New Issue
Block a user