Add isolated advertiser authentication, account hierarchy and project APIs

tests/test_advertiser.py

@@ -0,0 +1,187 @@
+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
+    )