roberto/adiuva-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Step 12 - completed

Browse Source
This commit is contained in:
roberto
2026-03-03 14:53:34 +01:00
parent 5d485b3665
commit d0b303e745
13 changed files with 950 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
tests/conftest.py Normal file
View File

@@ -0,0 +1,208 @@
"""Shared test fixtures for database-backed tests.
Provides an async SQLite in-memory engine that auto-creates all tables,
a per-test session, and a FastAPI ``TestClient`` wired to use it.
"""
from __future__ import annotations
import json
import time
import uuid
from collections.abc import AsyncGenerator, Generator
import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from jose import jwt
from sqlalchemy import StaticPool, event
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.config.settings import settings
from app.db import Base, get_session
from app.main import app
from app.models import Plugin, Subscription, User
# ── Fixed test user IDs (one per tier) ───────────────────────────────
TEST_USER_IDS: dict[str, str] = {
"free": "00000000-0000-0000-0000-000000000001",
"pro": "00000000-0000-0000-0000-000000000002",
"power": "00000000-0000-0000-0000-000000000003",
"team": "00000000-0000-0000-0000-000000000004",
}
# ── Async SQLite engine ──────────────────────────────────────────────
_TEST_ENGINE = create_async_engine(
"sqlite+aiosqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
_TestSessionLocal = async_sessionmaker(
_TEST_ENGINE,
expire_on_commit=False,
)
# Enable foreign key enforcement for SQLite (off by default).
@event.listens_for(_TEST_ENGINE.sync_engine, "connect")
def _set_sqlite_pragma(dbapi_conn, _connection_record): # noqa: ANN001
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
# ── Fixtures ─────────────────────────────────────────────────────────
@pytest_asyncio.fixture(autouse=True)
async def _create_tables():
"""Create all tables before each test, seed test users, then drop after."""
async with _TEST_ENGINE.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Seed one User + Subscription per tier so FK constraints and auth work.
async with _TestSessionLocal() as session:
for tier, uid in TEST_USER_IDS.items():
session.add(User(
id=uid,
email=f"{tier}@test.com",
password_hash="$2b$12$fakehashfortesting000000000000000000000000000",
tier=tier,
))
session.add(Subscription(
id=str(uuid.uuid4()),
user_id=uid,
tier=tier,
stripe_subscription_id=f"sub_test_{tier}",
status="active",
))
await session.commit()
yield
async with _TEST_ENGINE.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def db_session() -> AsyncGenerator[AsyncSession, None]:
"""Yield a per-test async DB session."""
async with _TestSessionLocal() as session:
yield session
@pytest.fixture
def client(db_session: AsyncSession) -> Generator[TestClient, None, None]: # noqa: ANN001
"""FastAPI test client with ``get_session`` overridden to use the test DB."""
async def _override_get_session() -> AsyncGenerator[AsyncSession, None]:
yield db_session
app.dependency_overrides[get_session] = _override_get_session
with TestClient(app) as c:
yield c
app.dependency_overrides.pop(get_session, None)
# ── Seed data helpers ────────────────────────────────────────────────
_SEED_PLUGINS = [
Plugin(
id="plugin-github-sync",
name="GitHub Sync",
description="Sync tasks with GitHub Issues and pull requests.",
version="1.0.0",
author_name="Adiuva",
category="productivity",
price_cents=0,
permissions=json.dumps(["read:tasks", "write:tasks"]),
status="approved",
s3_package_key="plugins/plugin-github-sync/1.0.0/package.zip",
install_count=0,
avg_rating=0.0,
),
Plugin(
id="plugin-slack-notify",
name="Slack Notifier",
description="Post task and checkpoint updates to Slack channels.",
version="1.2.0",
author_name="Adiuva",
category="communication",
price_cents=499,
permissions=json.dumps(["read:tasks", "read:checkpoints"]),
status="approved",
s3_package_key="plugins/plugin-slack-notify/1.2.0/package.zip",
install_count=0,
avg_rating=0.0,
),
Plugin(
id="plugin-time-tracker",
name="Time Tracker",
description="Track time spent on tasks with automatic reporting.",
version="0.9.1",
author_name="Third Party",
category="productivity",
price_cents=999,
permissions=json.dumps(["read:tasks", "write:tasks"]),
status="approved",
s3_package_key="plugins/plugin-time-tracker/0.9.1/package.zip",
install_count=0,
avg_rating=0.0,
),
]
@pytest_asyncio.fixture
async def seed_plugins(db_session: AsyncSession) -> list[Plugin]:
"""Insert the 3 default approved plugins and return them."""
plugins = []
for template in _SEED_PLUGINS:
p = Plugin(
id=template.id,
name=template.name,
description=template.description,
version=template.version,
author_name=template.author_name,
category=template.category,
price_cents=template.price_cents,
permissions=template.permissions,
status=template.status,
s3_package_key=template.s3_package_key,
install_count=template.install_count,
avg_rating=template.avg_rating,
)
db_session.add(p)
plugins.append(p)
await db_session.commit()
return plugins
# ── JWT helpers ──────────────────────────────────────────────────────
def make_jwt(
tier: str = "power",
user_id: str | None = None,
email: str | None = None,
) -> str:
"""Create a signed test JWT.
Uses the fixed ``TEST_USER_IDS`` mapping so the auth middleware can
find the corresponding ``Subscription`` row in the test database.
"""
uid = user_id or TEST_USER_IDS.get(tier, str(uuid.uuid4()))
now = int(time.time())
payload = {
"sub": uid,
"email": email or f"{tier}@test.com",
"tier": tier,
"exp": now + 3600,
"iat": now,
}
return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
def auth_header(tier: str = "power", user_id: str | None = None) -> dict[str, str]:
"""Return an Authorization header dict for the given tier."""
return {"Authorization": f"Bearer {make_jwt(tier, user_id)}"}

24
tests/test_middleware.py
View File

@@ -18,13 +18,30 @@ from fastapi.testclient import TestClient
from jose import jwt
from app.config.settings import settings
from app.db import get_session
from app.main import app
from app.schemas import ChatResponse
from tests.conftest import TEST_USER_IDS
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Autouse: redirect all DB access to the in-memory SQLite test engine.
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _override_db(db_session):
"""Route all get_session calls to the test SQLite session."""
async def _gen():
yield db_session
app.dependency_overrides[get_session] = _gen
yield
app.dependency_overrides.pop(get_session, None)
_CHAT_BODY = {
"message": "hello",
"context": {
@@ -74,14 +91,15 @@ class TestAuthMiddleware:
"""Tests exercised via GET /api/v1/auth/me."""
def test_valid_token_returns_profile(self) -> None:
uid = str(uuid.uuid4())
token = _make_jwt(user_id=uid, email="alice@example.com", tier="pro")
# Use the seeded pro user so the subscription lookup returns 'pro'.
uid = TEST_USER_IDS["pro"]
token = _make_jwt(user_id=uid, email="pro@test.com", tier="pro")
with TestClient(app) as client:
resp = client.get("/api/v1/auth/me", headers=_auth_header(token))
assert resp.status_code == 200
data = resp.json()
assert data["id"] == uid
assert data["email"] == "alice@example.com"
assert data["email"] == "pro@test.com"
assert data["tier"] == "pro"
def test_missing_token_returns_401(self) -> None:

341
tests/test_plugins.py
View File

@@ -1,52 +1,34 @@
"""Tests for Step 10: Plugin Marketplace.
"""Tests for Step 10+12: Plugin Marketplace (DB-backed).
Covers:
- PluginRegistry: catalog management, filtering, sorting, install counts
- PluginRegistry: catalog management, filtering, sorting, install counts (PostgreSQL)
- ReviewQueue: pending queue, review decisions, manifest security checklist
- RevenueShare: install event recording, earnings aggregation
- RevenueShare: install event recording, earnings aggregation (PostgreSQL)
- Route integration: tier gate, list/get/install/uninstall via TestClient
"""
from __future__ import annotations
import time
import json
import uuid
import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from jose import jwt
from unittest.mock import patch
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config.settings import settings
from app.main import app
from app.marketplace.plugin_registry import PluginRegistry
from app.marketplace.plugin_review import ReviewQueue, validate_manifest
from app.marketplace.revenue_share import RevenueShare
from app.models import Plugin, PluginReview as PluginReviewModel, RevenueEvent
from app.schemas import PluginManifest
from tests.conftest import TEST_USER_IDS, auth_header
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_jwt(tier: str = "power", user_id: str | None = None) -> str:
uid = user_id or str(uuid.uuid4())
now = int(time.time())
payload = {
"sub": uid,
"email": f"{uid[:8]}@example.com",
"tier": tier,
"exp": now + 3600,
"iat": now,
}
return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
def _auth(tier: str = "power") -> dict[str, str]:
return {"Authorization": f"Bearer {_make_jwt(tier)}"}
def _fresh_manifest(
plugin_id: str | None = None,
category: str = "productivity",
@@ -67,118 +49,150 @@ def _fresh_manifest(
# ---------------------------------------------------------------------------
# PluginRegistry
# PluginRegistry (DB-backed)
# ---------------------------------------------------------------------------
class TestPluginRegistry:
"""Each test uses a fresh PluginRegistry instance to avoid catalog pollution."""
"""Each test uses the conftest db_session fixture with a fresh in-memory DB."""
@pytest.fixture
def reg(self) -> PluginRegistry:
return PluginRegistry()
@pytest.mark.asyncio
async def test_seed_plugins_are_approved(self, reg: PluginRegistry) -> None:
result = await reg.list_plugins()
async def test_seed_plugins_are_listed(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
result = await reg.list_plugins(db_session)
assert result.total == 3
assert all(p.id.startswith("plugin-") for p in result.plugins)
@pytest.mark.asyncio
async def test_list_approved_only(self, reg: PluginRegistry) -> None:
async def test_list_approved_only(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
manifest = _fresh_manifest()
await reg.submit_plugin(manifest, "plugins/key.zip")
result = await reg.list_plugins()
await reg.submit_plugin(db_session, manifest, "plugins/key.zip")
result = await reg.list_plugins(db_session)
ids = [p.id for p in result.plugins]
assert manifest.id not in ids # still pending
@pytest.mark.asyncio
async def test_list_filter_by_category(self, reg: PluginRegistry) -> None:
result = await reg.list_plugins(category="communication")
async def test_list_filter_by_category(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
result = await reg.list_plugins(db_session, category="communication")
assert result.total == 1
assert result.plugins[0].id == "plugin-slack-notify"
@pytest.mark.asyncio
async def test_list_filter_by_query(self, reg: PluginRegistry) -> None:
result = await reg.list_plugins(query="time")
async def test_list_filter_by_query(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
result = await reg.list_plugins(db_session, query="time")
assert result.total == 1
assert result.plugins[0].id == "plugin-time-tracker"
@pytest.mark.asyncio
async def test_list_sort_by_installs(self, reg: PluginRegistry) -> None:
await reg.record_install("plugin-slack-notify")
await reg.record_install("plugin-slack-notify")
result = await reg.list_plugins(sort="installs")
async def test_list_sort_by_installs(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
await reg.record_install(db_session, "plugin-slack-notify")
await reg.record_install(db_session, "plugin-slack-notify")
result = await reg.list_plugins(db_session, sort="installs")
assert result.plugins[0].id == "plugin-slack-notify"
@pytest.mark.asyncio
async def test_get_plugin_found(self, reg: PluginRegistry) -> None:
entry = await reg.get_plugin("plugin-github-sync")
async def test_get_plugin_found(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
entry = await reg.get_plugin(db_session, "plugin-github-sync")
assert entry is not None
assert entry["manifest"].id == "plugin-github-sync"
assert "install_count" in entry
@pytest.mark.asyncio
async def test_get_plugin_not_found(self, reg: PluginRegistry) -> None:
entry = await reg.get_plugin("no-such-plugin")
async def test_get_plugin_not_found(
self, reg: PluginRegistry, db_session: AsyncSession
) -> None:
entry = await reg.get_plugin(db_session, "no-such-plugin")
assert entry is None
@pytest.mark.asyncio
async def test_submit_sets_pending(self, reg: PluginRegistry) -> None:
async def test_submit_sets_pending(
self, reg: PluginRegistry, db_session: AsyncSession
) -> None:
manifest = _fresh_manifest()
plugin_id = await reg.submit_plugin(manifest, "key.zip")
plugin_id = await reg.submit_plugin(db_session, manifest, "key.zip")
assert plugin_id == manifest.id
assert reg._catalog[plugin_id]["status"] == "pending_review"
result = await db_session.execute(select(Plugin).where(Plugin.id == plugin_id))
row = result.scalar_one()
assert row.status == "pending_review"
@pytest.mark.asyncio
async def test_approve_makes_visible(self, reg: PluginRegistry) -> None:
async def test_approve_makes_visible(
self, reg: PluginRegistry, db_session: AsyncSession
) -> None:
manifest = _fresh_manifest()
await reg.submit_plugin(manifest, "key.zip")
await reg.approve_plugin(manifest.id)
result = await reg.list_plugins()
await reg.submit_plugin(db_session, manifest, "key.zip")
await reg.approve_plugin(db_session, manifest.id)
result = await reg.list_plugins(db_session)
assert manifest.id in [p.id for p in result.plugins]
@pytest.mark.asyncio
async def test_reject_stores_reason(self, reg: PluginRegistry) -> None:
async def test_reject_stores_reason(
self, reg: PluginRegistry, db_session: AsyncSession
) -> None:
manifest = _fresh_manifest()
await reg.submit_plugin(manifest, "key.zip")
await reg.reject_plugin(manifest.id, reason="Unsafe permissions")
assert reg._catalog[manifest.id]["status"] == "rejected"
assert reg._catalog[manifest.id]["rejection_reason"] == "Unsafe permissions"
result = await reg.list_plugins()
assert manifest.id not in [p.id for p in result.plugins]
await reg.submit_plugin(db_session, manifest, "key.zip")
await reg.reject_plugin(db_session, manifest.id, reason="Unsafe permissions")
result = await db_session.execute(select(Plugin).where(Plugin.id == manifest.id))
row = result.scalar_one()
assert row.status == "rejected"
assert row.rejection_reason == "Unsafe permissions"
listed = await reg.list_plugins(db_session)
assert manifest.id not in [p.id for p in listed.plugins]
@pytest.mark.asyncio
async def test_approve_unknown_raises_key_error(self, reg: PluginRegistry) -> None:
async def test_approve_unknown_raises_key_error(
self, reg: PluginRegistry, db_session: AsyncSession
) -> None:
with pytest.raises(KeyError):
await reg.approve_plugin("ghost-plugin")
await reg.approve_plugin(db_session, "ghost-plugin")
@pytest.mark.asyncio
async def test_record_install_increments_count(self, reg: PluginRegistry) -> None:
await reg.record_install("plugin-github-sync")
entry = await reg.get_plugin("plugin-github-sync")
async def test_record_install_increments_count(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
await reg.record_install(db_session, "plugin-github-sync")
entry = await reg.get_plugin(db_session, "plugin-github-sync")
assert entry is not None
assert entry["install_count"] == 1
@pytest.mark.asyncio
async def test_record_uninstall_decrements_count(self, reg: PluginRegistry) -> None:
await reg.record_install("plugin-github-sync")
await reg.record_install("plugin-github-sync")
await reg.record_uninstall("plugin-github-sync")
entry = await reg.get_plugin("plugin-github-sync")
async def test_record_uninstall_decrements_count(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
await reg.record_install(db_session, "plugin-github-sync")
await reg.record_install(db_session, "plugin-github-sync")
await reg.record_uninstall(db_session, "plugin-github-sync")
entry = await reg.get_plugin(db_session, "plugin-github-sync")
assert entry is not None
assert entry["install_count"] == 1
@pytest.mark.asyncio
async def test_record_uninstall_floors_at_zero(self, reg: PluginRegistry) -> None:
await reg.record_uninstall("plugin-github-sync") # already 0
entry = await reg.get_plugin("plugin-github-sync")
async def test_record_uninstall_floors_at_zero(
self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
await reg.record_uninstall(db_session, "plugin-github-sync")
entry = await reg.get_plugin(db_session, "plugin-github-sync")
assert entry is not None
assert entry["install_count"] == 0
# ---------------------------------------------------------------------------
# ReviewQueue
# ReviewQueue (DB-backed)
# ---------------------------------------------------------------------------
@@ -188,37 +202,47 @@ class TestReviewQueue:
return PluginRegistry()
@pytest.fixture
def queue(self, reg: PluginRegistry) -> ReviewQueue:
# Patch the 'registry' name as bound inside plugin_review.py
with patch("app.marketplace.plugin_review.registry", reg):
yield ReviewQueue()
def queue(self) -> ReviewQueue:
return ReviewQueue()
@pytest.mark.asyncio
async def test_get_pending_returns_submitted_plugins(
self, reg: PluginRegistry, queue: ReviewQueue
self, reg: PluginRegistry, queue: ReviewQueue, db_session: AsyncSession
) -> None:
manifest = _fresh_manifest()
await reg.submit_plugin(manifest, "key.zip")
pending = await queue.get_pending()
await reg.submit_plugin(db_session, manifest, "key.zip")
pending = await queue.get_pending(db_session)
assert any(p["plugin_id"] == manifest.id for p in pending)
@pytest.mark.asyncio
async def test_submit_review_approved(
self, reg: PluginRegistry, queue: ReviewQueue
self, reg: PluginRegistry, queue: ReviewQueue, db_session: AsyncSession
) -> None:
manifest = _fresh_manifest()
await reg.submit_plugin(manifest, "key.zip")
await queue.submit_review(manifest.id, "reviewer-1", "approved", "Looks good")
assert reg._catalog[manifest.id]["status"] == "approved"
await reg.submit_plugin(db_session, manifest, "key.zip")
await queue.submit_review(db_session, manifest.id, TEST_USER_IDS["power"], "approved", "Looks good")
result = await db_session.execute(select(Plugin).where(Plugin.id == manifest.id))
row = result.scalar_one()
assert row.status == "approved"
# Check review row was persisted
review_result = await db_session.execute(
select(PluginReviewModel).where(PluginReviewModel.plugin_id == manifest.id)
)
review = review_result.scalar_one()
assert review.decision == "approved"
@pytest.mark.asyncio
async def test_submit_review_rejected(
self, reg: PluginRegistry, queue: ReviewQueue
self, reg: PluginRegistry, queue: ReviewQueue, db_session: AsyncSession
) -> None:
manifest = _fresh_manifest()
await reg.submit_plugin(manifest, "key.zip")
await queue.submit_review(manifest.id, "reviewer-1", "rejected", "Bad permissions")
assert reg._catalog[manifest.id]["status"] == "rejected"
await reg.submit_plugin(db_session, manifest, "key.zip")
await queue.submit_review(
db_session, manifest.id, TEST_USER_IDS["power"], "rejected", "Bad permissions"
)
result = await db_session.execute(select(Plugin).where(Plugin.id == manifest.id))
row = result.scalar_one()
assert row.status == "rejected"
def test_validate_manifest_ok(self) -> None:
manifest = _fresh_manifest(permissions=["read:tasks", "write:notes"])
@@ -241,65 +265,66 @@ class TestReviewQueue:
# ---------------------------------------------------------------------------
# RevenueShare
# RevenueShare (DB-backed)
# ---------------------------------------------------------------------------
class TestRevenueShare:
@pytest.fixture
def reg(self) -> PluginRegistry:
return PluginRegistry()
@pytest.fixture
def rs(self, reg: PluginRegistry) -> RevenueShare:
# Patch the 'registry' name as bound inside revenue_share.py
with patch("app.marketplace.revenue_share.registry", reg):
yield RevenueShare()
def rs(self) -> RevenueShare:
return RevenueShare()
@pytest.mark.asyncio
async def test_record_install_free_plugin(
self, reg: PluginRegistry, rs: RevenueShare
self, rs: RevenueShare, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
await rs.record_install("plugin-github-sync", "user-1", amount_cents=0)
assert len(rs._events) == 1
assert rs._events[0]["developer_share_cents"] == 0
await rs.record_install(db_session, "plugin-github-sync", TEST_USER_IDS["power"], amount_cents=0)
result = await db_session.execute(
select(RevenueEvent).where(RevenueEvent.plugin_id == "plugin-github-sync")
)
event = result.scalar_one()
assert event.developer_share_cents == 0
@pytest.mark.asyncio
async def test_record_install_paid_plugin_no_stripe(
self, reg: PluginRegistry, rs: RevenueShare
self, rs: RevenueShare, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
# No STRIPE_SECRET_KEY configured in test env — should not crash
await rs.record_install("plugin-slack-notify", "user-2", amount_cents=499)
assert len(rs._events) == 1
assert rs._events[0]["amount_cents"] == 499
assert rs._events[0]["developer_share_cents"] == int(499 * 0.70)
await rs.record_install(
db_session, "plugin-slack-notify", TEST_USER_IDS["pro"], amount_cents=499
)
result = await db_session.execute(
select(RevenueEvent).where(RevenueEvent.plugin_id == "plugin-slack-notify")
)
event = result.scalar_one()
assert event.amount_cents == 499
assert event.developer_share_cents == int(499 * 0.70)
@pytest.mark.asyncio
async def test_record_install_increments_registry_count(
self, reg: PluginRegistry, rs: RevenueShare
self, rs: RevenueShare, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
await rs.record_install("plugin-github-sync", "user-1", amount_cents=0)
entry = await reg.get_plugin("plugin-github-sync")
reg = PluginRegistry()
await rs.record_install(db_session, "plugin-github-sync", TEST_USER_IDS["power"], amount_cents=0)
entry = await reg.get_plugin(db_session, "plugin-github-sync")
assert entry is not None
assert entry["install_count"] == 1
@pytest.mark.asyncio
async def test_get_earnings_empty(
self, reg: PluginRegistry, rs: RevenueShare
self, rs: RevenueShare, db_session: AsyncSession
) -> None:
result = await rs.get_earnings("unknown-dev")
result = await rs.get_earnings(db_session, "unknown-dev")
assert result["total_installs"] == 0
assert result["total_revenue_cents"] == 0
assert result["developer_share_cents"] == 0
@pytest.mark.asyncio
async def test_get_earnings_aggregates(
self, reg: PluginRegistry, rs: RevenueShare
self, rs: RevenueShare, db_session: AsyncSession, seed_plugins: list[Plugin]
) -> None:
# "Adiuva" is the author of the seeded plugins
await rs.record_install("plugin-slack-notify", "u1", amount_cents=499)
await rs.record_install("plugin-slack-notify", "u2", amount_cents=499)
result = await rs.get_earnings("Adiuva")
await rs.record_install(db_session, "plugin-slack-notify", TEST_USER_IDS["power"], amount_cents=499)
await rs.record_install(db_session, "plugin-slack-notify", TEST_USER_IDS["pro"], amount_cents=499)
result = await rs.get_earnings(db_session, "Adiuva")
assert result["total_installs"] == 2
assert result["total_revenue_cents"] == 998
assert result["developer_share_cents"] == int(499 * 0.70) * 2
@@ -311,77 +336,67 @@ class TestRevenueShare:
class TestPluginRoutes:
def test_list_plugins_requires_power_tier(self) -> None:
with TestClient(app) as client:
resp = client.get("/api/v1/plugins", headers=_auth("free"))
def test_list_plugins_requires_power_tier(self, client, seed_plugins) -> None:
resp = client.get("/api/v1/plugins", headers=auth_header("free"))
assert resp.status_code == 403
def test_list_plugins_pro_tier_blocked(self) -> None:
with TestClient(app) as client:
resp = client.get("/api/v1/plugins", headers=_auth("pro"))
def test_list_plugins_pro_tier_blocked(self, client, seed_plugins) -> None:
resp = client.get("/api/v1/plugins", headers=auth_header("pro"))
assert resp.status_code == 403
def test_list_plugins_power_tier_ok(self) -> None:
with TestClient(app) as client:
resp = client.get("/api/v1/plugins", headers=_auth("power"))
def test_list_plugins_power_tier_ok(self, client, seed_plugins) -> None:
resp = client.get("/api/v1/plugins", headers=auth_header("power"))
assert resp.status_code == 200
data = resp.json()
assert "plugins" in data
assert data["total"] >= 3
assert data["total"] == 3
def test_list_plugins_team_tier_ok(self) -> None:
with TestClient(app) as client:
resp = client.get("/api/v1/plugins", headers=_auth("team"))
def test_list_plugins_team_tier_ok(self, client, seed_plugins) -> None:
resp = client.get("/api/v1/plugins", headers=auth_header("team"))
assert resp.status_code == 200
def test_get_plugin_found(self) -> None:
with TestClient(app) as client:
resp = client.get("/api/v1/plugins/plugin-github-sync", headers=_auth())
def test_get_plugin_found(self, client, seed_plugins) -> None:
resp = client.get("/api/v1/plugins/plugin-github-sync", headers=auth_header())
assert resp.status_code == 200
data = resp.json()
assert data["plugin"]["id"] == "plugin-github-sync"
assert "install_count" in data
def test_get_plugin_not_found(self) -> None:
with TestClient(app) as client:
resp = client.get("/api/v1/plugins/no-such-plugin", headers=_auth())
def test_get_plugin_not_found(self, client, seed_plugins) -> None:
resp = client.get("/api/v1/plugins/no-such-plugin", headers=auth_header())
assert resp.status_code == 404
def test_install_plugin_free(self) -> None:
with TestClient(app) as client:
resp = client.post(
"/api/v1/plugins/plugin-github-sync/install",
json={"plugin_id": "plugin-github-sync"},
headers=_auth(),
)
def test_install_plugin_free(self, client, seed_plugins) -> None:
resp = client.post(
"/api/v1/plugins/plugin-github-sync/install",
json={"plugin_id": "plugin-github-sync"},
headers=auth_header(),
)
assert resp.status_code == 200
data = resp.json()
assert data["ok"] is True
assert "download_url" in data
def test_install_plugin_not_found(self) -> None:
with TestClient(app) as client:
resp = client.post(
"/api/v1/plugins/ghost/install",
json={"plugin_id": "ghost"},
headers=_auth(),
)
def test_install_plugin_not_found(self, client, seed_plugins) -> None:
resp = client.post(
"/api/v1/plugins/ghost/install",
json={"plugin_id": "ghost"},
headers=auth_header(),
)
assert resp.status_code == 404
def test_uninstall_plugin_ok(self) -> None:
with TestClient(app) as client:
resp = client.delete(
"/api/v1/plugins/plugin-github-sync/install",
headers=_auth(),
)
def test_uninstall_plugin_ok(self, client, seed_plugins) -> None:
resp = client.delete(
"/api/v1/plugins/plugin-github-sync/install",
headers=auth_header(),
)
assert resp.status_code == 200
assert resp.json()["ok"] is True
def test_install_requires_power_tier(self) -> None:
with TestClient(app) as client:
resp = client.post(
"/api/v1/plugins/plugin-github-sync/install",
json={"plugin_id": "plugin-github-sync"},
headers=_auth("free"),
)
def test_install_requires_power_tier(self, client, seed_plugins) -> None:
resp = client.post(
"/api/v1/plugins/plugin-github-sync/install",
json={"plugin_id": "plugin-github-sync"},
headers=auth_header("free"),
)
assert resp.status_code == 403