adcloud-server/tests/test_advertiser.py

import os
from uuid import uuid4

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select, text
from sqlalchemy.engine import make_url

from adcloud_server.database import profiles
from adcloud_server.main import create_app
from adcloud_server.settings import Settings

PREFIX = "/v1/advertiser"


@pytest.fixture(params=["sqlite", "postgres"])
def client(tmp_path, request):
    engine = None
    schema = None
    url = f"sqlite:///{tmp_path}/app.db"
    if request.param == "postgres":
        base = os.getenv("ADCLOUD_TEST_POSTGRES_URL")
        if not base:
            pytest.skip("No PostgreSQL integration URL configured")
        schema = "advertiser_test_" + uuid4().hex
        engine = create_engine(base)
        with engine.begin() as conn:
            conn.execute(text(f'CREATE SCHEMA "{schema}"'))
        url = (
            make_url(base)
            .update_query_dict({"options": f"-csearch_path={schema}"})
            .render_as_string(hide_password=False)
        )
    try:
        # Advertiser authentication also works without the visitor identity secret.
        with TestClient(
            create_app(Settings(database_url=url, cookie_secure=False))
        ) as test_client:
            yield test_client
    finally:
        if engine is not None:
            # Only this fixture's random schema in the disposable CI database.
            with engine.begin() as conn:
                conn.execute(text(f'DROP SCHEMA "{schema}" CASCADE'))
            engine.dispose()


def register(client, email="owner@example.com"):
    response = client.post(
        PREFIX + "/register",
        json={
            "email": email,
            "password": "long-password-123",
            "name": "测试广告主",
            "company": "测试企业",
        },
    )
    assert response.status_code == 201, response.text
    return response


def project():
    return {"name": "秋季推广", "budget": 300, "bid": 30, "start_date": "2026-09-16"}


def test_registration_session_and_identity_separation(client):
    response = register(client)
    assert "HttpOnly" in response.headers["set-cookie"]
    assert "password" not in response.json()
    assert client.get(PREFIX + "/me").status_code == 200
    with client.app.state.advertiser_database.transaction() as conn:
        assert conn.execute(select(profiles)).first() is None
    assert client.post(PREFIX + "/logout").status_code == 204
    assert client.get(PREFIX + "/me").status_code == 401
    assert (
        client.post(
            PREFIX + "/login",
            json={"email": "owner@example.com", "password": "wrong-password"},
        ).status_code
        == 401
    )
    assert (
        client.post(
            PREFIX + "/login",
            json={"email": "OWNER@example.com", "password": "long-password-123"},
        ).status_code
        == 200
    )
    assert (
        client.post(
            PREFIX + "/register",
            json={
                "email": "owner@example.com",
                "password": "long-password-123",
                "name": "重复",
                "company": "重复",
            },
        ).status_code
        == 409
    )


def test_hierarchy_projects_and_isolation(client):
    register(client)
    accounts = client.get(PREFIX + "/accounts").json()
    parent = next(a for a in accounts if not a["parent_id"])
    child = next(a for a in accounts if a["parent_id"])
    assert child["parent_id"] == parent["id"]
    root_path = PREFIX + "/accounts/" + parent["id"]
    path = PREFIX + "/accounts/" + child["id"]
    assert client.post(root_path + "/projects", json=project()).status_code == 422
    assert (
        client.post(
            PREFIX + "/accounts",
            json={"parent_id": child["id"], "name": "三级账户", "company": "公司"},
        ).status_code
        == 422
    )
    created = client.post(path + "/projects", json=project())
    assert created.status_code == 201
    detail = client.get(path + "/projects/" + created.json()["id"])
    assert detail.status_code == 200
    assert detail.json()["name"] == project()["name"]
    item = created.json()
    item.pop("id")
    item["name"] = "修改后的项目"
    project_path = path + "/projects/" + created.json()["id"]
    assert client.put(project_path, json=item).status_code == 200
    assert client.get(path + "/projects").json()[0]["name"] == "修改后的项目"
    assert (
        client.post(path + "/projects", json={**project(), "budget": 99}).status_code
        == 422
    )
    assert (
        client.post(
            path + "/projects", json={**project(), "end_date": "2026-01-01"}
        ).status_code
        == 422
    )
    assert (
        client.post(
            path + "/projects", json={**project(), "landing_url": "javascript:alert(1)"}
        ).status_code
        == 422
    )
    client.post(PREFIX + "/logout")
    register(client, "other@example.com")
    assert client.get(path + "/projects").status_code == 404
    assert client.get(project_path).status_code == 404
    assert client.put(project_path, json=item).status_code == 404
    assert (
        client.put(path, json={"name": "越权修改", "company": "公司"}).status_code
        == 404
    )
    assert (
        client.post(
            PREFIX + "/accounts",
            json={"name": "越权子账户", "company": "公司", "parent_id": parent["id"]},
        ).status_code
        == 404
    )


def test_origin_and_throttle(client):
    assert (
        client.post(
            PREFIX + "/login",
            headers={"Origin": "https://evil.example"},
            json={"email": "owner@example.com", "password": "long-password"},
        ).status_code
        == 403
    )
    for _ in range(10):
        assert (
            client.post(
                PREFIX + "/login",
                json={"email": "unknown@example.com", "password": "wrong-password"},
            ).status_code
            == 401
        )
    assert (
        client.post(
            PREFIX + "/login",
            json={"email": "unknown@example.com", "password": "wrong-password"},
        ).status_code
        == 429
    )