Files
Nguyen Minh Phuc 50c2fe2a1e
ci / lint (push) Successful in 1m19s
ci / unit (push) Failing after 1m2s
ci / integration (push) Has been skipped
ci / types (push) Successful in 1m37s
ci / security (push) Failing after 38s
ci / dockerfile (push) Successful in 14s
ci / image (api) (push) Has been skipped
ci / image (reconciler) (push) Has been skipped
ci / image (worker) (push) Has been skipped
ci / bump (push) Has been skipped
svcforge: reference implementation
Complete working build of the system learn-python/ teaches.
164 tests, mypy --strict clean, domain coverage 99%.
2026-07-17 10:44:54 +00:00

45 lines
1.3 KiB
Python

"""Builders for integration tests. Keeps the tests about the behaviour under test."""
from __future__ import annotations
from datetime import UTC, datetime
from uuid import UUID, uuid4
from svcforge_core.domain.models import Instance
from svcforge_core.domain.states import InstanceState
from svcforge_core.repo.db import DictPool
from svcforge_core.repo.instances import InstanceRepo
def build_instance(
team: str = "platform",
service_type: str = "elasticsearch",
size: str = "small",
state: InstanceState = InstanceState.REQUESTED,
chart_version: str = "21.3.19",
) -> Instance:
"""An Instance with a deterministic release_name, as the domain requires."""
iid = uuid4()
now = datetime.now(UTC)
return Instance(
id=iid,
team=team,
service_type=service_type,
size=size,
state=state,
namespace=f"tenant-{team}",
release_name=f"{team}-{service_type}-{str(iid)[:8]}",
chart_version=chart_version,
created_at=now,
updated_at=now,
)
async def make_instance(pool: DictPool, **kwargs: object) -> UUID:
"""Insert an instance and return its id."""
inst = build_instance(**kwargs) # type: ignore[arg-type]
repo = InstanceRepo(pool)
async with pool.connection() as conn:
await repo.create(conn, inst)
return inst.id