325 lines
12 KiB
Python
325 lines
12 KiB
Python
"""Contract tests for the versioned strategy catalog API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from contextlib import ExitStack
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import StrategySet
|
|
from app.main import create_app
|
|
|
|
|
|
class FakeStrategyRepo:
|
|
def __init__(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
self.strategies = {
|
|
"active-1": StrategySet(
|
|
id="active-1", name="target_value", config="{}", nick="target_value:base",
|
|
status="active", revision=1, created_at=now, updated_at=now,
|
|
),
|
|
"inactive-1": StrategySet(
|
|
id="inactive-1", name="archived", config="{}", nick="archived:base",
|
|
status="inactive", revision=3, created_at=now, updated_at=now,
|
|
),
|
|
}
|
|
self.referenced: set[str] = set()
|
|
self.closed = False
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
def list_strategy_sets(self, *, include_inactive: bool = True) -> list[StrategySet]:
|
|
values = list(self.strategies.values())
|
|
return values if include_inactive else [item for item in values if item.status == "active"]
|
|
|
|
def get_strategy_set(self, id: str) -> StrategySet | None:
|
|
return self.strategies.get(id)
|
|
|
|
def get_or_create_strategy_set(self, name: str, config: str) -> StrategySet:
|
|
return self.create_strategy_set(name, config)
|
|
|
|
def create_strategy_set(
|
|
self,
|
|
name: str,
|
|
config: str,
|
|
*,
|
|
nick: str | None = None,
|
|
status: str = "active",
|
|
) -> StrategySet:
|
|
existing = next(
|
|
(item for item in self.strategies.values() if item.name == name and item.config == config),
|
|
None,
|
|
)
|
|
if existing is not None:
|
|
return existing
|
|
final_nick = nick or f"{name}:{hashlib.sha1(config.encode()).hexdigest()[:12]}"
|
|
if any(item.nick == final_nick for item in self.strategies.values()):
|
|
raise ValueError("동일한 nick 전략이 이미 존재합니다")
|
|
now = datetime.now(timezone.utc)
|
|
item = StrategySet(
|
|
id=f"created-{len(self.strategies)}",
|
|
name=name,
|
|
config=config,
|
|
nick=final_nick,
|
|
status=status,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.strategies[item.id] = item
|
|
return item
|
|
|
|
def update_strategy_set(
|
|
self,
|
|
id: str,
|
|
*,
|
|
name: str | None = None,
|
|
nick: str | None = None,
|
|
config: str | None = None,
|
|
status: str | None = None,
|
|
) -> StrategySet:
|
|
item = self.strategies.get(id)
|
|
if item is None:
|
|
raise ValueError("전략을 찾을 수 없습니다")
|
|
final_name = name if name is not None else item.name
|
|
final_nick = nick if nick is not None else item.nick
|
|
final_config = config if config is not None else item.config
|
|
if any(
|
|
other.id != id and other.name == final_name and other.config == final_config
|
|
for other in self.strategies.values()
|
|
):
|
|
raise ValueError("동일한 name/config 전략이 이미 존재합니다")
|
|
if any(other.id != id and other.nick == final_nick for other in self.strategies.values()):
|
|
raise ValueError("동일한 nick 전략이 이미 존재합니다")
|
|
final_status = status if status is not None else item.status
|
|
if (final_name, final_nick, final_config, final_status) == (
|
|
item.name,
|
|
item.nick,
|
|
item.config,
|
|
item.status,
|
|
):
|
|
return item
|
|
item.name = final_name
|
|
item.nick = final_nick
|
|
item.config = final_config
|
|
item.status = final_status
|
|
item.revision += 1
|
|
item.updated_at = datetime.now(timezone.utc)
|
|
return item
|
|
|
|
def delete_strategy_set(self, id: str) -> None:
|
|
if id in self.referenced:
|
|
raise ValueError("Strategy set is in use by 1 portfolio(s) and cannot be deleted")
|
|
if id not in self.strategies:
|
|
raise ValueError("전략을 찾을 수 없습니다")
|
|
del self.strategies[id]
|
|
|
|
|
|
def _client() -> TestClient:
|
|
settings = Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
admin_tokens_raw="admin-token",
|
|
read_tokens_raw="read-token",
|
|
)
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
return TestClient(app)
|
|
|
|
|
|
def _repos(repo: FakeStrategyRepo) -> ExitStack:
|
|
stack = ExitStack()
|
|
stack.enter_context(patch("app.api.v1_strategies.get_portfolio_repository_from_settings", return_value=repo))
|
|
stack.enter_context(patch("app.core.auth.get_portfolio_repository_from_settings", return_value=repo))
|
|
return stack
|
|
|
|
|
|
def _auth(token: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def test_public_catalog_excludes_inactive_and_inactive_lookup_conflicts() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
response = _client().get("/v1/strategies", headers=_auth("read-token"))
|
|
active = _client().get("/v1/strategies/active-1", headers=_auth("read-token"))
|
|
inactive = _client().get("/v1/strategies/inactive-1", headers=_auth("read-token"))
|
|
assert response.status_code == 200
|
|
assert [item["strategy_id"] for item in response.json()] == ["active-1"]
|
|
assert response.json()[0]["nick"] == "target_value:base"
|
|
assert active.status_code == 200
|
|
assert active.json()["nick"] == "target_value:base"
|
|
assert inactive.status_code == 409
|
|
|
|
|
|
def test_admin_catalog_requires_admin_and_reports_missing_strategy() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
forbidden = _client().get("/v1/admin/strategies", headers=_auth("read-token"))
|
|
missing = _client().get("/v1/admin/strategies/missing", headers=_auth("admin-token"))
|
|
assert forbidden.status_code == 403
|
|
assert missing.status_code == 404
|
|
|
|
|
|
def test_admin_create_custom_nick_and_patch_nick_only_preserve_id_and_increment_revision() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
created = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "momentum", "nick": "mom-20", "config": {"window": 20}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
strategy_id = created.json()["strategy_id"]
|
|
changed = _client().patch(
|
|
f"/v1/admin/strategies/{strategy_id}",
|
|
json={"nick": "mom-primary"},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
no_op = _client().patch(
|
|
f"/v1/admin/strategies/{strategy_id}",
|
|
json={"nick": "mom-primary"},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
assert created.status_code == 201
|
|
assert created.json()["nick"] == "mom-20"
|
|
assert created.json()["revision"] == 1
|
|
assert changed.status_code == 200
|
|
assert changed.json()["strategy_id"] == strategy_id
|
|
assert changed.json()["nick"] == "mom-primary"
|
|
assert changed.json()["revision"] == 2
|
|
assert no_op.json()["revision"] == 2
|
|
assert repo.strategies[strategy_id].nick == "mom-primary"
|
|
|
|
|
|
def test_admin_create_trims_nick() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
created = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "momentum", "nick": " mom-20 ", "config": {"window": 20}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
assert created.status_code == 201
|
|
assert created.json()["nick"] == "mom-20"
|
|
|
|
|
|
def test_admin_create_and_patch_reject_whitespace_only_nick() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
created = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "momentum", "nick": " \t ", "config": {"window": 20}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
changed = _client().patch(
|
|
"/v1/admin/strategies/active-1",
|
|
json={"nick": " \n "},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
assert created.status_code == 422
|
|
assert changed.status_code == 422
|
|
assert repo.strategies["active-1"].nick == "target_value:base"
|
|
|
|
|
|
def test_admin_patch_omitting_nick_preserves_custom_nick_when_name_and_config_change() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
created = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "momentum", "nick": "mom-custom", "config": {"window": 20}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
strategy_id = created.json()["strategy_id"]
|
|
changed = _client().patch(
|
|
f"/v1/admin/strategies/{strategy_id}",
|
|
json={"name": "momentum-v2", "config": {"window": 30}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
assert changed.status_code == 200
|
|
assert changed.json()["nick"] == "mom-custom"
|
|
assert changed.json()["revision"] == 2
|
|
|
|
|
|
def test_admin_create_omitting_nick_retains_legacy_generated_value() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
created = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "legacy", "config": {"window": 20}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
canonical_config = '{"window":20}'
|
|
expected = f"legacy:{hashlib.sha1(canonical_config.encode()).hexdigest()[:12]}"
|
|
assert created.status_code == 201
|
|
assert created.json()["nick"] == expected
|
|
|
|
|
|
def test_admin_create_and_patch_reject_explicit_null_nick_without_mutation() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
created = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "legacy-null", "nick": None, "config": {}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
changed = _client().patch(
|
|
"/v1/admin/strategies/active-1",
|
|
json={"nick": None},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
assert created.status_code == 422
|
|
assert changed.status_code == 422
|
|
assert set(repo.strategies) == {"active-1", "inactive-1"}
|
|
assert repo.strategies["active-1"].nick == "target_value:base"
|
|
assert repo.strategies["active-1"].revision == 1
|
|
|
|
|
|
def test_admin_duplicate_nick_returns_conflict() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
duplicate = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "duplicate", "nick": "target_value:base", "config": {}},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
assert duplicate.status_code == 409
|
|
|
|
|
|
def test_admin_create_inactive_starts_at_revision_one() -> None:
|
|
repo = FakeStrategyRepo()
|
|
with _repos(repo):
|
|
created = _client().post(
|
|
"/v1/admin/strategies",
|
|
json={"name": "retired", "config": {}, "status": "inactive"},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
assert created.status_code == 201
|
|
assert created.json()["status"] == "inactive"
|
|
assert created.json()["revision"] == 1
|
|
|
|
|
|
def test_admin_patch_validation_conflict_and_reference_safe_delete() -> None:
|
|
repo = FakeStrategyRepo()
|
|
repo.referenced.add("active-1")
|
|
with _repos(repo):
|
|
empty = _client().patch("/v1/admin/strategies/active-1", json={}, headers=_auth("admin-token"))
|
|
conflict = _client().patch(
|
|
"/v1/admin/strategies/inactive-1",
|
|
json={"name": "target_value"},
|
|
headers=_auth("admin-token"),
|
|
)
|
|
referenced = _client().delete("/v1/admin/strategies/active-1", headers=_auth("admin-token"))
|
|
deleted = _client().delete("/v1/admin/strategies/inactive-1", headers=_auth("admin-token"))
|
|
assert empty.status_code == 422
|
|
assert conflict.status_code == 409
|
|
assert referenced.status_code == 409
|
|
assert deleted.status_code == 204
|
|
|
|
|
|
def test_legacy_strategy_set_routes_are_removed() -> None:
|
|
response = _client().get("/strategy-sets", headers=_auth("admin-token"))
|
|
assert response.status_code == 404
|