Oliver/oliver-sales-ops-platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
oliver-sales-ops-platform/backend/tests/test_team_shape.py
DJP 96b8e2dc3d Stage 15 stub: pitch deck markdown export 
Phase 1 stub for the proposal deck. Pulls every upstream artifact
together into a 5-section markdown doc:

  1. The opportunity at a glance — intake summary + brands + delivery model
  2. What's being asked for — channels, markets, capabilities, deliverables, ambiguities
  3. Why we're a fit — qualification % + recommendation + delivery model + per-stage split
  4. Commercial framing — caveats, assumptions, KPIs, capability gaps
  5. AI cost so far — running total + call count

GET /opportunities/{id}/pitch-deck/markdown returns the .md file with the
right Content-Disposition. Saves a stage_artifact (type='pitch_deck_markdown').

Phase 2 will swap this for a python-pptx deck assembled from a template
library (the per-section content stays the same — just the rendering
changes).

Smoke-tested: 6.5KB doc on the Versuni opportunity, all 5 sections
populated from real artifacts.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-27 14:24:30 -04:00

527 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Stage 11 — team shape (FTE per role) coverage.
Team shape is pure-Python (no Claude) so it's exercised end-to-end against
the live backend. We seed ClientAsset + Match rows directly via psycopg2
(same backdoor `test_ratecard.py` uses), call the real `/ratecard/build`,
then exercise the `GET /team-shape` query-param surface:
- empty ratecard
- per-1-asset hours x volume aggregation (V1 bug-4 invariant)
- blanket `efficiency_pct` with programme roles excluded
- `efficiency_pct=100` is *not* capped (only `discipline_overrides` is)
- `discipline_overrides` cap at MAX_EFFICIENCY=90
- `discipline_overrides` is malformed JSON => 400
- precedence: when `discipline_overrides` is provided, disciplines NOT in
the map fall back to 0% (the blanket pct is ignored — see service code)
- 404 on unknown opp
ClientAsset / Match / RatecardLine / StageArtifact rows all cascade on
opportunity delete, so the per-test `_custom_opp` helper handles teardown.
"""
from __future__ import annotations
import os
from contextlib import asynccontextmanager
from datetime import date
import httpx
import psycopg2
import pytest
_DB_DSN = os.environ.get(
"OSOP_TEST_DSN",
"host=127.0.0.1 port=5435 dbname=oliver_sales_ops user=osop_user password=osop_pass_2026",
)
# --------------------------------------------------------------------------- #
# Helpers (parallel to test_ratecard.py — kept local so this file is
# self-contained and doesn't depend on the order tests load).
# --------------------------------------------------------------------------- #
def _seed_client_asset(
opp_id: int,
raw_name: str = "Test asset",
volume: int = 1,
sort_order: int = 0,
) -> int:
conn = psycopg2.connect(_DB_DSN)
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO client_assets
(opportunity_id, raw_name, volume, sort_order)
VALUES (%s, %s, %s, %s)
RETURNING id
""",
(opp_id, raw_name, volume, sort_order),
)
new_id = cur.fetchone()[0]
conn.commit()
finally:
conn.close()
return new_id
def _seed_match(
client_asset_id: int,
gmal_asset_id: int,
is_selected: bool = True,
confidence: str = "exact",
confidence_score: float = 0.95,
rank: int = 1,
) -> int:
conn = psycopg2.connect(_DB_DSN)
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO matches
(client_asset_id, gmal_asset_id, confidence, confidence_score,
is_selected, rank)
VALUES (%s, %s, %s::matchconfidence, %s, %s, %s)
RETURNING id
""",
(
client_asset_id,
gmal_asset_id,
confidence.upper(),
confidence_score,
is_selected,
rank,
),
)
new_id = cur.fetchone()[0]
conn.commit()
finally:
conn.close()
return new_id
def _pick_gmal_for_model(model_type: str = "CURRENT_OPLUS", min_rows: int = 2) -> int:
"""Same pattern as test_ratecard.py — return a GMAL with at least `min_rows`
GmalHours entries for the given model_type."""
conn = psycopg2.connect(_DB_DSN)
try:
with conn.cursor() as cur:
cur.execute(
"""
SELECT gmal_asset_id, COUNT(*)
FROM gmal_hours
WHERE model_type = %s::modeltype
GROUP BY gmal_asset_id
HAVING COUNT(*) >= %s
ORDER BY gmal_asset_id
LIMIT 1
""",
(model_type, min_rows),
)
row = cur.fetchone()
finally:
conn.close()
assert row is not None, (
f"expected a GMAL with {min_rows}+ {model_type} hour rows — "
"seed your gmal_hours table first"
)
return row[0]
def _opportunity_payload(model_type: str, name_suffix: str = "") -> dict:
return {
"name": f"PYTEST TeamShape {name_suffix}".strip(),
"client_name": "Test Client Ltd",
"region": "EMEA",
"brands": "BrandA",
"service_types": "Content Production",
"description": "Team shape test opportunity.",
"deadline": date(2027, 1, 15).isoformat(),
"go_live": date(2027, 3, 1).isoformat(),
"model_type": model_type,
}
@asynccontextmanager
async def _custom_opp(client: httpx.AsyncClient, model_type: str, suffix: str):
"""Create+yield+delete a fresh opportunity with the given model_type."""
payload = _opportunity_payload(model_type, name_suffix=suffix)
r = await client.post("/opportunities", json=payload)
r.raise_for_status()
opp = r.json()
try:
yield opp
finally:
try:
await client.delete(f"/opportunities/{opp['id']}")
except Exception:
pass
def _has_creative_role(roles: list[dict]) -> bool:
"""The first GMAL (id=1) is known to have a Creative-discipline role; the
discipline_override tests assert against that. If our chosen GMAL doesn't
have one, we'd need to broaden the test — fail loudly here instead."""
return any(r["discipline"] == "Creative" for r in roles)
# --------------------------------------------------------------------------- #
# 1. Empty ratecard — response shape OK, roles=[], totals=0
# --------------------------------------------------------------------------- #
async def test_team_shape_empty_ratecard(client: httpx.AsyncClient):
"""An opportunity with no RatecardLine rows should return an empty roles
list and zeroed totals (the route still returns 200 — the service short-
circuits when there are no lines)."""
async with _custom_opp(client, "current_oplus", suffix="(empty)") as opp:
r = await client.get(f"/opportunities/{opp['id']}/team-shape")
assert r.status_code == 200, r.text
body = r.json()
assert body["project_id"] == opp["id"]
assert body["roles"] == []
assert body["total_hours"] == 0
assert body["total_fte"] == 0
assert body["adjusted_hours"] == 0
assert body["adjusted_fte"] == 0
assert body["delivery_fte"] == 0
assert body["programme_fte"] == 0
assert body["hours_saved"] == 0
assert body["fte_saved"] == 0
assert body["hours_per_fte"] == 1800
assert body["efficiency_pct"] == 0.0
assert body["discipline_overrides"] is None
# --------------------------------------------------------------------------- #
# 2. Bug-4 invariant: total_hours = base_hours x volume, and FTE math
# --------------------------------------------------------------------------- #
async def test_team_shape_zero_efficiency_volume_aggregation(client: httpx.AsyncClient):
"""Seed 1 ClientAsset (volume=10) + selected Match, build the ratecard,
GET /team-shape with efficiency_pct=0 and verify:
- each role's `total_hours == base_hours × 10` (bug-4 fix invariant —
per-1-asset hours on the line are aggregated × line.volume in the
team-shape service)
- `total_fte = total_hours / 1800` and matches the response field
- `adjusted_hours == total_hours` and `fte_saved == 0` when no
efficiency is requested."""
async with _custom_opp(client, "current_oplus", suffix="(volume)") as opp:
opp_id = opp["id"]
gmal_asset_id = _pick_gmal_for_model("CURRENT_OPLUS", min_rows=2)
ca_id = _seed_client_asset(opp_id, raw_name="Hero PDP", volume=10, sort_order=0)
_seed_match(ca_id, gmal_asset_id, is_selected=True, rank=1)
# Build the ratecard via the real endpoint.
rb = await client.post(f"/opportunities/{opp_id}/ratecard/build")
assert rb.status_code == 200, rb.text
# Pull line-level hours so we can independently compute the expected
# aggregate (per-1-asset base_hours × volume).
rc = await client.get(f"/opportunities/{opp_id}/ratecard")
assert rc.status_code == 200, rc.text
lines = rc.json()["lines"]
assert len(lines) >= 2
# Sum base_hours × volume per role_id (the team_shape service does
# this internally).
expected_per_role: dict[int, float] = {}
for ln in lines:
# Role id isn't directly on the ratecard line response, so we'll
# match by `role_title` which is unique within an asset's lines.
key = ln["role_title"]
expected_per_role[key] = (
expected_per_role.get(key, 0.0)
+ float(ln["base_hours"]) * ln["volume"]
)
r = await client.get(f"/opportunities/{opp_id}/team-shape?efficiency_pct=0")
assert r.status_code == 200, r.text
body = r.json()
assert body["efficiency_pct"] == 0.0
assert body["discipline_overrides"] is None
# Same number of roles as ratecard lines (1 role per line for this
# single GMAL).
roles = body["roles"]
assert len(roles) == len(lines)
for role in roles:
title = role["role_title"]
expected_total = round(expected_per_role[title], 2)
assert role["total_hours"] == expected_total, (
f"bug-4 invariant: role {title!r} should have "
f"base_hours × volume = {expected_total}, got {role['total_hours']}"
)
# FTE math
expected_fte = round(role["total_hours"] / 1800, 4)
assert role["fte"] == expected_fte
# Zero efficiency => adjusted == total, fte_saved == 0
assert role["adjusted_hours"] == role["total_hours"]
assert role["adjusted_fte"] == role["fte"]
assert role["fte_saved"] == 0
assert role["hours_saved"] == 0
assert role["efficiency_pct"] == 0
# Aggregate FTE matches sum of role FTEs (rounded to 4dp at the
# service-level top totals).
total_fte_expected = round(sum(r["fte"] for r in roles), 4)
# The route rounds the aggregate to 4dp independently from the role
# rounding, so allow a tiny float tolerance.
assert body["total_fte"] == pytest.approx(total_fte_expected, abs=1e-3)
assert body["adjusted_fte"] == pytest.approx(total_fte_expected, abs=1e-3)
assert body["fte_saved"] == 0
assert body["hours_saved"] == 0
# --------------------------------------------------------------------------- #
# 3. Programme roles never reduced; non-programme cut by blanket pct
# --------------------------------------------------------------------------- #
async def test_team_shape_blanket_50_pct_excludes_programme(client: httpx.AsyncClient):
"""With a blanket 50% efficiency, programme roles must keep
adjusted_hours == total_hours (efficiency_pct=0 on those rows) while
every other role has adjusted_hours == total_hours × 0.5.
Also asserts delivery_fte + programme_fte == total_fte (within a 2dp
rounding tolerance — the route rounds aggregates independently)."""
async with _custom_opp(client, "current_oplus", suffix="(50pct)") as opp:
opp_id = opp["id"]
gmal_asset_id = _pick_gmal_for_model("CURRENT_OPLUS", min_rows=2)
ca_id = _seed_client_asset(opp_id, raw_name="Hero PDP", volume=10, sort_order=0)
_seed_match(ca_id, gmal_asset_id, is_selected=True, rank=1)
rb = await client.post(f"/opportunities/{opp_id}/ratecard/build")
assert rb.status_code == 200, rb.text
r = await client.get(
f"/opportunities/{opp_id}/team-shape?efficiency_pct=50"
)
assert r.status_code == 200, r.text
body = r.json()
assert body["efficiency_pct"] == 50.0
roles = body["roles"]
assert len(roles) > 0
saw_programme = False
saw_delivery = False
for role in roles:
if role["is_programme_role"]:
saw_programme = True
assert role["efficiency_pct"] == 0, (
f"programme role {role['role_title']!r} should have "
f"efficiency_pct=0 even with blanket 50%, "
f"got {role['efficiency_pct']}"
)
assert role["adjusted_hours"] == role["total_hours"]
assert role["adjusted_fte"] == role["fte"]
assert role["fte_saved"] == 0
else:
saw_delivery = True
assert role["efficiency_pct"] == 50.0
expected_adj = round(role["total_hours"] * 0.5, 2)
assert role["adjusted_hours"] == expected_adj, (
f"non-programme role {role['role_title']!r} should be "
f"halved: expected {expected_adj}, got {role['adjusted_hours']}"
)
assert saw_programme, (
"expected at least one programme role on the chosen GMAL — "
"if this fails, the test seed data has changed"
)
assert saw_delivery, (
"expected at least one non-programme role on the chosen GMAL"
)
# delivery_fte + programme_fte ~= total_fte (each rounded
# independently by the route, so tolerate a 2dp drift).
assert body["delivery_fte"] + body["programme_fte"] == pytest.approx(
body["total_fte"], abs=0.01
)
# --------------------------------------------------------------------------- #
# 4. efficiency_pct=100 — NOT capped (V2-as-shipped behaviour)
# --------------------------------------------------------------------------- #
async def test_team_shape_blanket_100_not_capped(client: httpx.AsyncClient):
"""V2-as-shipped behaviour: the MAX_EFFICIENCY=90 cap is enforced ONLY
when `discipline_overrides` is used. The blanket `efficiency_pct` query
parameter has NO cap — passing 100 collapses every non-programme role to
zero adjusted hours.
This test pins the current behaviour. If a future release adds a cap to
the blanket path, expect this assertion to flip; that's intentional."""
async with _custom_opp(client, "current_oplus", suffix="(100pct)") as opp:
opp_id = opp["id"]
gmal_asset_id = _pick_gmal_for_model("CURRENT_OPLUS", min_rows=2)
ca_id = _seed_client_asset(opp_id, raw_name="Hero PDP", volume=10, sort_order=0)
_seed_match(ca_id, gmal_asset_id, is_selected=True, rank=1)
rb = await client.post(f"/opportunities/{opp_id}/ratecard/build")
assert rb.status_code == 200, rb.text
r = await client.get(
f"/opportunities/{opp_id}/team-shape?efficiency_pct=100"
)
assert r.status_code == 200, r.text
body = r.json()
assert body["efficiency_pct"] == 100.0
for role in body["roles"]:
if role["is_programme_role"]:
# Programme roles always exempt
assert role["efficiency_pct"] == 0
assert role["adjusted_hours"] == role["total_hours"]
else:
# Blanket 100% — uncapped — collapses to zero adjusted hours.
assert role["efficiency_pct"] == 100.0
assert role["adjusted_hours"] == 0, (
"blanket efficiency_pct=100 is uncapped (V2-as-shipped); "
f"role {role['role_title']!r} adjusted_hours should be 0, "
f"got {role['adjusted_hours']}"
)
assert role["adjusted_fte"] == 0
# --------------------------------------------------------------------------- #
# 5. discipline_overrides clamps to MAX_EFFICIENCY=90
# --------------------------------------------------------------------------- #
async def test_team_shape_discipline_overrides_capped_at_90(client: httpx.AsyncClient):
"""`discipline_overrides={"Creative": 95}` — the service caps the value
at MAX_EFFICIENCY=90, so the response should report `efficiency_pct=90.0`
on every Creative-discipline non-programme role (and the adjusted hours
should reflect a 90% cut, not 95%)."""
async with _custom_opp(client, "current_oplus", suffix="(cap)") as opp:
opp_id = opp["id"]
gmal_asset_id = _pick_gmal_for_model("CURRENT_OPLUS", min_rows=2)
ca_id = _seed_client_asset(opp_id, raw_name="Hero PDP", volume=10, sort_order=0)
_seed_match(ca_id, gmal_asset_id, is_selected=True, rank=1)
rb = await client.post(f"/opportunities/{opp_id}/ratecard/build")
assert rb.status_code == 200, rb.text
# Use httpx params to handle URL encoding of the JSON value.
r = await client.get(
f"/opportunities/{opp_id}/team-shape",
params={"discipline_overrides": '{"Creative": 95}'},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["discipline_overrides"] == {"Creative": 95.0}
roles = body["roles"]
assert _has_creative_role(roles), (
"expected a Creative-discipline role on the chosen GMAL — "
"if this fails, the test seed data has changed"
)
for role in roles:
if role["is_programme_role"]:
# Programme always exempt — Delivery discipline is not in the
# override map anyway; even if it were, programme is exempt.
assert role["efficiency_pct"] == 0
elif role["discipline"] == "Creative":
assert role["efficiency_pct"] == 90.0, (
f"Creative role {role['role_title']!r} should be capped at "
f"MAX_EFFICIENCY=90, got {role['efficiency_pct']}"
)
expected_adj = round(role["total_hours"] * 0.10, 2)
assert role["adjusted_hours"] == expected_adj
else:
# Disciplines NOT in the override map fall back to 0% — the
# presence of `discipline_overrides` switches the service
# away from the blanket path (covered separately in test 7).
assert role["efficiency_pct"] == 0
# --------------------------------------------------------------------------- #
# 6. Malformed discipline_overrides JSON => 400
# --------------------------------------------------------------------------- #
async def test_team_shape_invalid_overrides_json_400(client: httpx.AsyncClient):
"""`discipline_overrides=not-json` is a 400 with a hint about JSON."""
async with _custom_opp(client, "current_oplus", suffix="(bad-json)") as opp:
r = await client.get(
f"/opportunities/{opp['id']}/team-shape",
params={"discipline_overrides": "not-json"},
)
assert r.status_code == 400, r.text
detail = r.json().get("detail", "").lower()
assert "json" in detail or "discipline" in detail, (
f"detail should mention JSON / discipline, got: {detail!r}"
)
# --------------------------------------------------------------------------- #
# 7. Precedence: discipline_overrides wins, blanket is ignored for unmapped
# disciplines (they fall back to 0, NOT the blanket pct)
# --------------------------------------------------------------------------- #
async def test_team_shape_overrides_precedence_over_blanket(client: httpx.AsyncClient):
"""When `discipline_overrides` is provided alongside `efficiency_pct`,
the per-discipline map wins. The service only consults the blanket
`efficiency_pct` when `discipline_overrides` is empty/None — so
disciplines NOT listed in the override map fall back to 0% efficiency,
NOT to the blanket value.
Read: services/team_shape.py:74-82 — programme→0, then
`if discipline_efficiency:` short-circuits before the `efficiency_pct`
branch can run."""
async with _custom_opp(client, "current_oplus", suffix="(precedence)") as opp:
opp_id = opp["id"]
gmal_asset_id = _pick_gmal_for_model("CURRENT_OPLUS", min_rows=2)
ca_id = _seed_client_asset(opp_id, raw_name="Hero PDP", volume=10, sort_order=0)
_seed_match(ca_id, gmal_asset_id, is_selected=True, rank=1)
rb = await client.post(f"/opportunities/{opp_id}/ratecard/build")
assert rb.status_code == 200, rb.text
r = await client.get(
f"/opportunities/{opp_id}/team-shape",
params={
"discipline_overrides": '{"Creative": 50}',
"efficiency_pct": 75, # should be ignored
},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["discipline_overrides"] == {"Creative": 50.0}
assert body["efficiency_pct"] == 75.0 # echoed, but unused
roles = body["roles"]
for role in roles:
if role["is_programme_role"]:
assert role["efficiency_pct"] == 0
elif role["discipline"] == "Creative":
assert role["efficiency_pct"] == 50.0, (
f"Creative override should apply: expected 50, got "
f"{role['efficiency_pct']}"
)
else:
# NOT in the override map — falls back to 0, NOT the
# blanket 75. This is the precedence behaviour.
assert role["efficiency_pct"] == 0, (
f"role {role['role_title']!r} ({role['discipline']!r}) is "
f"not in the override map — should fall back to 0, NOT "
f"the blanket 75. got {role['efficiency_pct']}"
)
# --------------------------------------------------------------------------- #
# 8. 404 on unknown opportunity
# --------------------------------------------------------------------------- #
async def test_team_shape_unknown_opportunity_404(client: httpx.AsyncClient):
r = await client.get("/opportunities/9999999/team-shape")
assert r.status_code == 404, r.text
Reference in a new issue View git blame Copy permalink