generated from coulomb/repo-seed
tegwick
77689fbfb2
Add a local /services router (source of truth for the catalog itself):
- GET /services/catalog with hosting_type / development_type / maturity_level /
status filters (eager-loads all four extensions)
- GET /services/{slug}
- POST /services/catalog upsert-by-slug, applying the dimension extensions;
first_party.repo_slug resolves to a managed_repos FK.
Extensions are read/written via session.get (not the relationship attribute) to
avoid async lazy-load. /tpsc/* is left intact for dependency snapshots. 7 tests.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
94 lines
3.8 KiB
Python
94 lines
3.8 KiB
Python
"""Tests for the two-dimension service catalog API (STATE-WP-0062)."""
|
|
from __future__ import annotations
|
|
|
|
|
|
def _svc(slug, hosting, development, **extra):
|
|
body = {
|
|
"slug": slug,
|
|
"name": slug.replace("-", " ").title(),
|
|
"hosting_type": hosting,
|
|
"development_type": development,
|
|
}
|
|
body.update(extra)
|
|
return body
|
|
|
|
|
|
async def test_upsert_cloud_third_party_with_extensions(client):
|
|
r = await client.post("/services/catalog", json=_svc(
|
|
"openai-api", "cloud_hosted", "third_party",
|
|
owner_or_provider="OpenAI",
|
|
cloud={"gdpr_maturity": "developing", "dpa_available": True},
|
|
third_party={"pricing_model": "usage_based", "support_url": "https://help.openai.com"},
|
|
))
|
|
assert r.status_code == 201, r.text
|
|
body = r.json()
|
|
assert body["hosting_type"] == "cloud_hosted"
|
|
assert body["development_type"] == "third_party"
|
|
assert body["cloud"]["gdpr_maturity"] == "developing"
|
|
assert body["cloud"]["dpa_available"] is True
|
|
assert body["third_party"]["pricing_model"] == "usage_based"
|
|
assert body["first_party"] is None
|
|
assert body["self_hosted"] is None
|
|
|
|
|
|
async def test_upsert_self_hosted_first_party_with_maturity(client):
|
|
r = await client.post("/services/catalog", json=_svc(
|
|
"state-hub", "self_hosted", "first_party",
|
|
maturity_level=2,
|
|
first_party={"owning_domain": "custodian"},
|
|
self_hosted={"helix_instance": "coulombcore", "runbook_ref": "make api"},
|
|
))
|
|
assert r.status_code == 201, r.text
|
|
body = r.json()
|
|
assert body["maturity_level"] == 2
|
|
assert body["first_party"]["owning_domain"] == "custodian"
|
|
assert body["self_hosted"]["helix_instance"] == "coulombcore"
|
|
assert body["cloud"] is None
|
|
|
|
|
|
async def test_dimension_filters(client):
|
|
await client.post("/services/catalog", json=_svc("gitea", "self_hosted", "third_party",
|
|
self_hosted={"upstream_oss_project": "Gitea"}))
|
|
await client.post("/services/catalog", json=_svc("stripe", "cloud_hosted", "third_party"))
|
|
await client.post("/services/catalog", json=_svc("hub", "self_hosted", "first_party", maturity_level=3))
|
|
|
|
self_hosted = (await client.get("/services/catalog?hosting_type=self_hosted")).json()
|
|
assert {s["slug"] for s in self_hosted} == {"gitea", "hub"}
|
|
|
|
first_party = (await client.get("/services/catalog?development_type=first_party")).json()
|
|
assert {s["slug"] for s in first_party} == {"hub"}
|
|
|
|
level3 = (await client.get("/services/catalog?maturity_level=3")).json()
|
|
assert {s["slug"] for s in level3} == {"hub"}
|
|
|
|
|
|
async def test_upsert_is_idempotent_on_slug(client):
|
|
await client.post("/services/catalog", json=_svc("svc-x", "cloud_hosted", "third_party",
|
|
category="search"))
|
|
r = await client.post("/services/catalog", json=_svc("svc-x", "cloud_hosted", "third_party",
|
|
category="storage"))
|
|
assert r.status_code == 201, r.text
|
|
assert r.json()["category"] == "storage"
|
|
listed = (await client.get("/services/catalog")).json()
|
|
assert sum(1 for s in listed if s["slug"] == "svc-x") == 1
|
|
|
|
|
|
async def test_invalid_dimensions_rejected(client):
|
|
r = await client.post("/services/catalog", json=_svc("bad", "on_prem", "third_party"))
|
|
assert r.status_code == 422
|
|
r = await client.post("/services/catalog", json=_svc("bad2", "cloud_hosted", "nobody"))
|
|
assert r.status_code == 422
|
|
|
|
|
|
async def test_first_party_unknown_repo_slug_404(client):
|
|
r = await client.post("/services/catalog", json=_svc(
|
|
"svc-y", "self_hosted", "first_party",
|
|
first_party={"repo_slug": "does-not-exist"},
|
|
))
|
|
assert r.status_code == 404
|
|
|
|
|
|
async def test_get_unknown_service_404(client):
|
|
r = await client.get("/services/nope")
|
|
assert r.status_code == 404
|