5936ab167e
Completed slices: - S01: Desire Embedding & Clustering - S02: Fulfillment Flow & Frontend Branch: milestone/M001
294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""Tests for desire fulfillment endpoint and cluster_count annotation.
|
|
|
|
Covers:
|
|
- fulfill_desire endpoint: happy path, not-found, not-open, shader validation
|
|
(tested via source assertions since FastAPI isn't in the test environment)
|
|
- cluster_count annotation: batch query pattern, single desire query
|
|
- Schema field: cluster_count exists in DesirePublic
|
|
|
|
Approach: Per K005, router functions can't be imported without FastAPI installed.
|
|
We verify correctness through:
|
|
1. Source-level structure assertions (endpoint wiring, imports, validation logic)
|
|
2. Isolated logic unit tests (annotation loop, status transitions)
|
|
3. Schema field verification via Pydantic model introspection
|
|
"""
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _router_source() -> str:
|
|
"""Read the desires router source code."""
|
|
return (
|
|
Path(__file__).resolve().parent.parent
|
|
/ "app"
|
|
/ "routers"
|
|
/ "desires.py"
|
|
).read_text(encoding="utf-8")
|
|
|
|
|
|
def _schema_source() -> str:
|
|
"""Read the schemas source code."""
|
|
return (
|
|
Path(__file__).resolve().parent.parent
|
|
/ "app"
|
|
/ "schemas"
|
|
/ "schemas.py"
|
|
).read_text(encoding="utf-8")
|
|
|
|
|
|
def _make_mock_desire(
|
|
*,
|
|
desire_id=None,
|
|
status="open",
|
|
heat_score=1.0,
|
|
):
|
|
"""Create a mock object simulating a Desire ORM instance."""
|
|
d = MagicMock()
|
|
d.id = desire_id or uuid.uuid4()
|
|
d.status = status
|
|
d.heat_score = heat_score
|
|
d.cluster_count = 0 # default before annotation
|
|
return d
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fulfill_desire — happy path structure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFulfillHappyPath:
|
|
"""Verify the fulfill endpoint's happy-path logic via source analysis."""
|
|
|
|
def test_fulfill_sets_status_to_fulfilled(self):
|
|
"""The endpoint sets desire.status = 'fulfilled' on success."""
|
|
source = _router_source()
|
|
assert 'desire.status = "fulfilled"' in source
|
|
|
|
def test_fulfill_sets_fulfilled_by_shader(self):
|
|
"""The endpoint sets desire.fulfilled_by_shader = shader_id."""
|
|
source = _router_source()
|
|
assert "desire.fulfilled_by_shader = shader_id" in source
|
|
|
|
def test_fulfill_sets_fulfilled_at_timestamp(self):
|
|
"""The endpoint sets desire.fulfilled_at to current UTC time."""
|
|
source = _router_source()
|
|
assert "desire.fulfilled_at" in source
|
|
assert "datetime.now(timezone.utc)" in source
|
|
|
|
def test_fulfill_returns_status_response(self):
|
|
"""The endpoint returns a dict with status, desire_id, shader_id."""
|
|
source = _router_source()
|
|
assert '"status": "fulfilled"' in source
|
|
assert '"desire_id"' in source
|
|
assert '"shader_id"' in source
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fulfill_desire — error paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFulfillDesireNotFound:
|
|
"""404 when desire doesn't exist."""
|
|
|
|
def test_desire_not_found_raises_404(self):
|
|
source = _router_source()
|
|
# After desire lookup, checks scalar_one_or_none result
|
|
assert "Desire not found" in source
|
|
|
|
|
|
class TestFulfillDesireNotOpen:
|
|
"""400 when desire is not in 'open' status."""
|
|
|
|
def test_desire_not_open_check_exists(self):
|
|
source = _router_source()
|
|
assert 'desire.status != "open"' in source
|
|
|
|
def test_desire_not_open_error_message(self):
|
|
source = _router_source()
|
|
assert "Desire is not open" in source
|
|
|
|
|
|
class TestFulfillShaderNotFound:
|
|
"""404 when shader_id doesn't match any shader."""
|
|
|
|
def test_shader_lookup_exists(self):
|
|
source = _router_source()
|
|
assert "select(Shader).where(Shader.id == shader_id)" in source
|
|
|
|
def test_shader_not_found_raises_404(self):
|
|
source = _router_source()
|
|
assert "Shader not found" in source
|
|
|
|
|
|
class TestFulfillShaderNotPublished:
|
|
"""400 when shader status is not 'published'."""
|
|
|
|
def test_shader_status_validation(self):
|
|
source = _router_source()
|
|
assert 'shader.status != "published"' in source
|
|
|
|
def test_shader_not_published_error_message(self):
|
|
source = _router_source()
|
|
assert "Shader must be published to fulfill a desire" in source
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cluster_count annotation — logic unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestClusterCountAnnotation:
|
|
"""Verify cluster_count annotation logic patterns."""
|
|
|
|
def test_list_desires_has_batch_cluster_query(self):
|
|
"""list_desires uses a batch query with ANY(:desire_ids)."""
|
|
source = _router_source()
|
|
assert "ANY(:desire_ids)" in source
|
|
assert "desire_clusters dc1" in source
|
|
assert "desire_clusters dc2" in source
|
|
|
|
def test_list_desires_avoids_n_plus_1(self):
|
|
"""Cluster counts are fetched in a single batch, not per-desire."""
|
|
source = _router_source()
|
|
# The pattern: build dict from batch query, then loop to annotate
|
|
assert "cluster_counts = {" in source
|
|
assert "cluster_counts.get(d.id, 0)" in source
|
|
|
|
def test_list_desires_skips_cluster_query_when_empty(self):
|
|
"""When no desires are returned, cluster query is skipped."""
|
|
source = _router_source()
|
|
assert "if desire_ids:" in source
|
|
|
|
def test_get_desire_annotates_single_cluster_count(self):
|
|
"""get_desire runs a cluster count query for the single desire."""
|
|
source = _router_source()
|
|
# Should have a cluster query scoped to a single desire_id
|
|
assert "WHERE dc1.desire_id = :desire_id" in source
|
|
|
|
def test_annotation_loop_sets_default_zero(self):
|
|
"""Desires without cluster entries default to cluster_count = 0."""
|
|
source = _router_source()
|
|
assert "cluster_counts.get(d.id, 0)" in source
|
|
|
|
def test_annotation_loop_logic(self):
|
|
"""Unit test: the annotation loop correctly maps cluster counts to desires."""
|
|
# Simulate the annotation loop from list_desires
|
|
d1 = _make_mock_desire()
|
|
d2 = _make_mock_desire()
|
|
d3 = _make_mock_desire()
|
|
desires = [d1, d2, d3]
|
|
|
|
# Simulate cluster query result: d1 has 3 in cluster, d3 has 2
|
|
cluster_counts = {d1.id: 3, d3.id: 2}
|
|
|
|
# This is the exact logic from the router
|
|
for d in desires:
|
|
d.cluster_count = cluster_counts.get(d.id, 0)
|
|
|
|
assert d1.cluster_count == 3
|
|
assert d2.cluster_count == 0 # not in any cluster
|
|
assert d3.cluster_count == 2
|
|
|
|
def test_get_desire_cluster_count_fallback(self):
|
|
"""get_desire sets cluster_count=0 when no cluster row exists."""
|
|
source = _router_source()
|
|
# The router checks `row[0] if row else 0`
|
|
assert "row[0] if row else 0" in source
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schema field verification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDesirePublicSchema:
|
|
"""Verify DesirePublic schema has the cluster_count field."""
|
|
|
|
def test_cluster_count_field_in_schema_source(self):
|
|
"""DesirePublic schema source contains cluster_count field."""
|
|
source = _schema_source()
|
|
assert "cluster_count" in source
|
|
|
|
def test_cluster_count_default_zero(self):
|
|
"""cluster_count defaults to 0 in the schema."""
|
|
source = _schema_source()
|
|
assert "cluster_count: int = 0" in source
|
|
|
|
def test_schema_from_attributes_enabled(self):
|
|
"""DesirePublic uses from_attributes=True for ORM compatibility."""
|
|
source = _schema_source()
|
|
# Find the DesirePublic class section
|
|
desire_public_idx = source.index("class DesirePublic")
|
|
desire_public_section = source[desire_public_idx:desire_public_idx + 200]
|
|
assert "from_attributes=True" in desire_public_section
|
|
|
|
def test_cluster_count_pydantic_model(self):
|
|
"""DesirePublic schema has cluster_count as an int field with default 0."""
|
|
source = _schema_source()
|
|
# Find the DesirePublic class and verify cluster_count is between
|
|
# heat_score and fulfilled_by_shader (correct field ordering)
|
|
desire_idx = source.index("class DesirePublic")
|
|
desire_section = source[desire_idx:desire_idx + 500]
|
|
heat_pos = desire_section.index("heat_score")
|
|
cluster_pos = desire_section.index("cluster_count")
|
|
fulfilled_pos = desire_section.index("fulfilled_by_shader")
|
|
assert heat_pos < cluster_pos < fulfilled_pos, (
|
|
"cluster_count should be between heat_score and fulfilled_by_shader"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Wiring assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFulfillmentWiring:
|
|
"""Structural assertions that the router is properly wired."""
|
|
|
|
def test_router_imports_shader_model(self):
|
|
"""desires.py imports Shader for shader validation."""
|
|
source = _router_source()
|
|
assert "Shader" in source.split("\n")[8] # near top imports
|
|
|
|
def test_router_imports_text_from_sqlalchemy(self):
|
|
"""desires.py imports text from sqlalchemy for raw SQL."""
|
|
source = _router_source()
|
|
assert "from sqlalchemy import" in source
|
|
assert "text" in source
|
|
|
|
def test_fulfill_endpoint_requires_auth(self):
|
|
"""fulfill_desire uses get_current_user dependency."""
|
|
source = _router_source()
|
|
# Find the fulfill_desire function
|
|
fulfill_idx = source.index("async def fulfill_desire")
|
|
fulfill_section = source[fulfill_idx:fulfill_idx + 500]
|
|
assert "get_current_user" in fulfill_section
|
|
|
|
def test_fulfill_endpoint_takes_shader_id_param(self):
|
|
"""fulfill_desire accepts shader_id as a query parameter."""
|
|
source = _router_source()
|
|
fulfill_idx = source.index("async def fulfill_desire")
|
|
fulfill_section = source[fulfill_idx:fulfill_idx + 300]
|
|
assert "shader_id" in fulfill_section
|
|
|
|
def test_list_desires_returns_desire_public(self):
|
|
"""list_desires endpoint uses DesirePublic response model."""
|
|
source = _router_source()
|
|
assert "response_model=list[DesirePublic]" in source
|
|
|
|
def test_get_desire_returns_desire_public(self):
|
|
"""get_desire endpoint uses DesirePublic response model."""
|
|
source = _router_source()
|
|
# Find the get_desire endpoint specifically
|
|
lines = source.split("\n")
|
|
for i, line in enumerate(lines):
|
|
if "async def get_desire" in line:
|
|
# Check the decorator line above
|
|
decorator_line = lines[i - 1]
|
|
assert "response_model=DesirePublic" in decorator_line
|
|
break
|