mjalmoudy94
a1b8a28dda
A push to `dev` now becomes a live demo, and a failed build never takes the demo down. - Four tables (pipeline_runs, pipeline_steps, releases, app_runtimes) and their migration; alembic upgrade/downgrade/check all clean. - DeliveryService implements DeliveryHooks: on_app_created provisions the demo port + host + first build (idempotently); on_branch_pushed queues a run for `dev` and supersedes older *queued* runs, leaving a running one alone. - The 8-step pipeline. Blue/green is real: the replacement starts on a staging port, is healthchecked, and only then does the route move — then the previous container is stopped. A failure removes the replacement, releases its port and leaves the old container serving. - The demo port ledger is the DATABASE, not the router adapters' in-memory maps: the API and the delivery worker are separate processes. - Runs are claimed with a conditional UPDATE, so two workers never build the same run. - Live logs are batched (1 event / 500 ms, 200 lines) and redacted as a whole tail — per-chunk redaction misses a secret split across pipe reads, and the *stored* tail leaked until this was fixed. - delivery worker: drain the queue, then poll remotes for pushes made outside the engine. - 7 routes; reads need membership, actions need projects.manage and are audited. Tests: 415 API (+21, 1 docker-marked skipped); ruff, mypy, alembic check clean. Report: docs/plan/reports/agent-3-phase-3.md
1026 lines
39 KiB
Python
1026 lines
39 KiB
Python
"""§v2-A3.3 — live CI/CD: the pipeline, the demo runtime and the safety rules.
|
|
|
|
The subject here is not "does the code run" but *what happens when it does not*.
|
|
Three properties carry the whole design and every test below is about one of
|
|
them:
|
|
|
|
1. **A failed run never takes the demo down.** The replacement container is
|
|
started on a staging port and healthchecked; the route only moves after it is
|
|
known good. A red build leaves the previous container serving.
|
|
2. **One active run per project.** A second push supersedes a *queued* run and
|
|
leaves a *running* one alone.
|
|
3. **Nothing leaks.** A secret planted in a build log must not reach a stored
|
|
step tail or a published event.
|
|
|
|
The container runtime is faked — a pipeline test that needed docker would not run
|
|
on a laptop. Everything else (the database, the port ledger, the domain router's
|
|
validation rules, the state machine) is real.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
|
|
from app.application.ports.realtime import (
|
|
RealtimeEvent,
|
|
set_publisher,
|
|
)
|
|
from app.application.services.delivery_service import (
|
|
DEMO_SERVICE,
|
|
DeliveryService,
|
|
http_healthcheck,
|
|
)
|
|
from app.core.config import settings
|
|
from app.domain.delivery.models import PIPELINE_STEPS, RunStatus, StepStatus
|
|
from app.infrastructure.adapters.mock import MockDomainRouterAdapter
|
|
from app.infrastructure.adapters.process import ProcessResult
|
|
from app.infrastructure.repositories.app_profile import AppProfileRepository
|
|
from app.infrastructure.repositories.delivery import (
|
|
AppRuntimeRepository,
|
|
PipelineRunRepository,
|
|
PipelineStepRepository,
|
|
ReleaseRepository,
|
|
)
|
|
from app.infrastructure.repositories.identity import OrganizationRepository
|
|
from app.infrastructure.repositories.projects import (
|
|
PreviewDomainRepository,
|
|
ProjectRepository,
|
|
)
|
|
from tests.conftest import auth_header, signup
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fakes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class RecordingPublisher:
|
|
def __init__(self) -> None:
|
|
self.events: list[RealtimeEvent] = []
|
|
|
|
async def publish(self, event: RealtimeEvent) -> None:
|
|
self.events.append(event)
|
|
|
|
def types(self) -> list[str]:
|
|
return [e.type for e in self.events]
|
|
|
|
def payloads(self, event_type: str) -> list[dict]:
|
|
return [e.data for e in self.events if e.type == event_type]
|
|
|
|
|
|
class FakeContainers:
|
|
"""A container runtime that runs nothing and records everything.
|
|
|
|
``script`` maps a step's ``name_suffix`` to the result to return, so a test
|
|
can make exactly one step fail — which is how the "a red build keeps the old
|
|
demo" property is exercised.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls: list[dict] = []
|
|
self.script: dict[str, ProcessResult] = {}
|
|
self.default = ProcessResult(argv=(), exit_code=0, stdout="ok\n")
|
|
self.removed: list[str] = []
|
|
self._counter = 0
|
|
|
|
async def run_step(self, **kwargs): # noqa: ANN003
|
|
self.calls.append(kwargs)
|
|
suffix = kwargs.get("name_suffix", "")
|
|
if suffix in self.script:
|
|
return self.script[suffix]
|
|
if kwargs.get("keep_alive"):
|
|
self._counter += 1
|
|
return ProcessResult(
|
|
argv=(), exit_code=0, stdout=f"ctr-{suffix}-{self._counter}\n"
|
|
)
|
|
return self.default
|
|
|
|
async def remove_container(self, container_id: str) -> bool:
|
|
self.removed.append(container_id)
|
|
return True
|
|
|
|
def steps_run(self) -> list[str]:
|
|
return [c.get("name_suffix", "") for c in self.calls]
|
|
|
|
def call_for(self, suffix: str) -> dict | None:
|
|
return next((c for c in self.calls if c.get("name_suffix") == suffix), None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest_asyncio.fixture(autouse=True)
|
|
async def _platform_seed(seeded):
|
|
return seeded
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def publisher():
|
|
"""Capture every realtime event the delivery service publishes."""
|
|
from app.application.ports.realtime import NullPublisher
|
|
|
|
recorder = RecordingPublisher()
|
|
set_publisher(recorder)
|
|
try:
|
|
yield recorder
|
|
finally:
|
|
set_publisher(NullPublisher())
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def tenant(client: AsyncClient, session_factory):
|
|
"""An org with one app, a full profile and a demo runtime.
|
|
|
|
Created through the real HTTP API so the rows are shaped exactly as the
|
|
wizard would leave them, then given a profile and a runtime through the
|
|
repositories the delivery service itself uses.
|
|
"""
|
|
token = (await signup(client, "delivery@example.com"))["access_token"]
|
|
headers = auth_header(token)
|
|
org_id = (
|
|
await client.post("/api/organizations", json={"name": "شرکت نمونه"}, headers=headers)
|
|
).json()["id"]
|
|
project_id = (
|
|
await client.post(
|
|
f"/api/organizations/{org_id}/projects",
|
|
json={"name": "اپلیکیشن نمونه"},
|
|
headers=headers,
|
|
)
|
|
).json()["id"]
|
|
|
|
async with session_factory() as session:
|
|
await AppProfileRepository(session).update_profile(
|
|
organization_id=org_id,
|
|
project_id=project_id,
|
|
values={
|
|
"app_type": "web_app",
|
|
"language": "python",
|
|
"framework": "fastapi",
|
|
"setup_command": "pip install -r requirements.txt",
|
|
"build_command": "python -m compileall app",
|
|
"test_command": "pytest -q",
|
|
"start_command": "uvicorn app:app --port 8000",
|
|
"app_port": 8000,
|
|
"healthcheck_path": "/health",
|
|
},
|
|
)
|
|
await session.commit()
|
|
|
|
return {"org_id": org_id, "project_id": project_id, "headers": headers}
|
|
|
|
|
|
def _build_service(
|
|
session,
|
|
*,
|
|
containers: FakeContainers,
|
|
healthcheck=None,
|
|
domains=None,
|
|
) -> DeliveryService:
|
|
return DeliveryService(
|
|
runs=PipelineRunRepository(session),
|
|
steps=PipelineStepRepository(session),
|
|
releases=ReleaseRepository(session),
|
|
runtimes=AppRuntimeRepository(session),
|
|
projects=ProjectRepository(session),
|
|
profiles=AppProfileRepository(session),
|
|
organizations=OrganizationRepository(session),
|
|
ports=PreviewDomainRepository(session),
|
|
workspace=_FakeWorkspace(),
|
|
domains=domains or MockDomainRouterAdapter(),
|
|
containers=containers,
|
|
healthcheck=healthcheck or _always_healthy,
|
|
)
|
|
|
|
|
|
async def _always_healthy(_url: str) -> bool:
|
|
return True
|
|
|
|
|
|
async def _never_healthy(_url: str) -> bool:
|
|
return False
|
|
|
|
|
|
class _FakeWorkspace:
|
|
"""Minimal ``RepoWorkspacePort``: a head, a worktree, and nothing else."""
|
|
|
|
def __init__(self, head: str = "a" * 40) -> None:
|
|
self.calls: list[tuple[str, dict]] = []
|
|
self._head = head
|
|
self.removed: list[str] = []
|
|
|
|
async def ensure_workspace(self, *, project_id: str):
|
|
from app.application.ports.repo_workspace import WorkspaceRef
|
|
|
|
self.calls.append(("ensure_workspace", {"project_id": project_id}))
|
|
return WorkspaceRef(project_id=project_id, host_path="/fake/work")
|
|
|
|
async def create_task_worktree(self, *, project_id: str, task_id: str, base: str = "dev"):
|
|
from app.application.ports.repo_workspace import WorktreeRef
|
|
|
|
self.calls.append(("create_task_worktree", {"task_id": task_id, "base": base}))
|
|
return WorktreeRef(
|
|
project_id=project_id,
|
|
task_id=task_id,
|
|
branch=f"task/{task_id}",
|
|
host_path=f"/fake/work/.worktrees/{task_id}",
|
|
sandbox_path=f"/workspace/.worktrees/{task_id}",
|
|
)
|
|
|
|
async def remove_worktree(self, *, worktree) -> None: # noqa: ANN001
|
|
self.removed.append(worktree.task_id)
|
|
|
|
async def head(self, *, project_id: str, branch: str) -> str | None:
|
|
return self._head
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# the hooks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_on_app_created_provisions_a_demo_identity(
|
|
tenant, session_factory, publisher
|
|
):
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
runtime = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime is not None
|
|
assert runtime.demo_host.endswith(settings.preview_root_domain)
|
|
assert runtime.host_port is not None
|
|
assert settings.demo_port_start <= runtime.host_port <= settings.demo_port_end
|
|
|
|
runs = await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
assert len(runs) == 1, "a fresh app must get exactly one first build"
|
|
assert runs[0].status == RunStatus.QUEUED.value
|
|
|
|
|
|
async def test_on_app_created_is_idempotent(tenant, session_factory):
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
ports = await PreviewDomainRepository(session).list_used_ports(
|
|
start=settings.demo_port_start, end=settings.demo_port_end
|
|
)
|
|
assert len(ports) == 1, "a retry must not reserve a second port"
|
|
runtime = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime is not None and runtime.host_port in ports
|
|
|
|
|
|
async def test_on_branch_pushed_only_cares_about_dev(tenant, session_factory):
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=FakeContainers())
|
|
await service.on_branch_pushed(
|
|
organization_id=tenant["org_id"],
|
|
project_id=tenant["project_id"],
|
|
branch="task/abc",
|
|
sha="b" * 40,
|
|
source="engine",
|
|
)
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
runs = await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
assert runs == [], "a task branch is work in progress, not a deploy"
|
|
|
|
|
|
async def test_a_second_push_supersedes_a_queued_run(tenant, session_factory):
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=FakeContainers())
|
|
first = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="1" * 40
|
|
)
|
|
second = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="2" * 40
|
|
)
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
repo = PipelineRunRepository(session)
|
|
assert (await repo.get(first.id)).status == RunStatus.SUPERSEDED.value
|
|
assert (await repo.get(second.id)).status == RunStatus.QUEUED.value
|
|
|
|
|
|
async def test_a_running_run_is_not_superseded(tenant, session_factory):
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=FakeContainers())
|
|
running = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="1" * 40
|
|
)
|
|
await PipelineRunRepository(session).set_status(
|
|
running.id, RunStatus.RUNNING.value, started=True
|
|
)
|
|
await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="2" * 40
|
|
)
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
assert (
|
|
await PipelineRunRepository(session).get(running.id)
|
|
).status == RunStatus.RUNNING.value, "a started build is doing real work"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# the happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_run_walks_every_step_and_publishes_a_release(
|
|
tenant, session_factory, publisher
|
|
):
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
finished = await service.execute_run(run.id)
|
|
await session.commit()
|
|
|
|
assert finished.status == RunStatus.SUCCEEDED.value, finished.error_summary
|
|
# Six of the eight steps run a command in a fresh container; `healthcheck`
|
|
# and `route` are handled by the service itself (they are a poll and a
|
|
# config write, not a build).
|
|
assert containers.steps_run() == [
|
|
"checkout",
|
|
"setup",
|
|
"build",
|
|
"test",
|
|
"package",
|
|
"demo",
|
|
]
|
|
|
|
async with session_factory() as session:
|
|
steps = await PipelineStepRepository(session).list_for_run(run.id)
|
|
assert [s.name for s in steps] == list(PIPELINE_STEPS)
|
|
assert all(s.status == StepStatus.SUCCEEDED.value for s in steps), [
|
|
(s.name, s.status) for s in steps
|
|
]
|
|
|
|
releases = await ReleaseRepository(session).list_for_project(
|
|
project_id=tenant["project_id"]
|
|
)
|
|
assert len(releases) == 1
|
|
assert releases[0].is_demo_live is True
|
|
|
|
runtime = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime is not None
|
|
assert runtime.status == "live"
|
|
assert runtime.current_release_id == releases[0].id
|
|
|
|
# The event stream tells the same story.
|
|
assert "pipeline.queued" in publisher.types()
|
|
assert "pipeline.status" in publisher.types()
|
|
assert "pipeline.step" in publisher.types()
|
|
assert "pipeline.log" in publisher.types()
|
|
assert "release.live" in publisher.types()
|
|
statuses = [p["status"] for p in publisher.payloads("pipeline.status")]
|
|
assert statuses[0] == RunStatus.RUNNING.value
|
|
assert statuses[-1] == RunStatus.SUCCEEDED.value
|
|
|
|
|
|
async def test_a_step_without_a_command_is_skipped_not_failed(
|
|
tenant, session_factory, publisher
|
|
):
|
|
async with session_factory() as session:
|
|
await AppProfileRepository(session).update_profile(
|
|
organization_id=tenant["org_id"],
|
|
project_id=tenant["project_id"],
|
|
values={"test_command": "", "build_command": ""},
|
|
)
|
|
await session.commit()
|
|
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
finished = await service.execute_run(run.id)
|
|
await session.commit()
|
|
|
|
assert finished.status == RunStatus.SUCCEEDED.value
|
|
async with session_factory() as session:
|
|
steps = await PipelineStepRepository(session).list_for_run(run.id)
|
|
by_name = {s.name: s for s in steps}
|
|
assert by_name["build"].status == StepStatus.SKIPPED.value
|
|
assert by_name["test"].status == StepStatus.SKIPPED.value
|
|
assert by_name["setup"].status == StepStatus.SUCCEEDED.value
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# failure paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_failed_build_keeps_the_previous_demo(
|
|
tenant, session_factory, publisher
|
|
):
|
|
"""The property the whole blue/green design exists for."""
|
|
# 1. A first, successful deploy establishes a live container.
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
first = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
assert (await service.execute_run(first.id)).status == RunStatus.SUCCEEDED.value
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
runtime_before = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime_before is not None
|
|
live_container = runtime_before.demo_container_id
|
|
live_port = runtime_before.host_port
|
|
assert live_container is not None
|
|
|
|
# 2. The next build goes red at the test step.
|
|
broken = FakeContainers()
|
|
broken.script["test"] = ProcessResult(
|
|
argv=(), exit_code=1, stdout="", stderr="1 failed\n"
|
|
)
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=broken)
|
|
second = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="c" * 40
|
|
)
|
|
await session.commit()
|
|
finished = await service.execute_run(second.id)
|
|
await session.commit()
|
|
|
|
assert finished.status == RunStatus.FAILED.value
|
|
assert "test" in finished.error_summary or "تست" in finished.error_summary
|
|
|
|
async with session_factory() as session:
|
|
runtime_after = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime_after is not None
|
|
assert runtime_after.demo_container_id == live_container, (
|
|
"a red build must not replace the live container"
|
|
)
|
|
assert runtime_after.host_port == live_port
|
|
|
|
# The route was never reconfigured: no `route` step ran at all.
|
|
assert "route" not in broken.steps_run()
|
|
assert "healthcheck" not in broken.steps_run()
|
|
|
|
# The failed replacement was cleaned up.
|
|
assert broken.removed == [], "the failed container never started"
|
|
statuses = [p["status"] for p in publisher.payloads("pipeline.status")]
|
|
assert statuses[-1] == RunStatus.FAILED.value
|
|
|
|
|
|
async def test_a_failed_healthcheck_does_not_switch_the_route(
|
|
tenant, session_factory, publisher
|
|
):
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(
|
|
session, containers=containers, healthcheck=_never_healthy
|
|
)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
finished = await service.execute_run(run.id)
|
|
await session.commit()
|
|
|
|
assert finished.status == RunStatus.FAILED.value
|
|
async with session_factory() as session:
|
|
steps = await PipelineStepRepository(session).list_for_run(run.id)
|
|
by_name = {s.name: s for s in steps}
|
|
assert by_name["healthcheck"].status == StepStatus.FAILED.value
|
|
assert by_name["route"].status == StepStatus.PENDING.value, (
|
|
"the route step must never run after a failed healthcheck"
|
|
)
|
|
assert by_name["package"].status == StepStatus.SUCCEEDED.value
|
|
|
|
runtime = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime is not None
|
|
assert runtime.status == "degraded"
|
|
assert runtime.current_release_id is None
|
|
releases = await ReleaseRepository(session).list_for_project(
|
|
project_id=tenant["project_id"]
|
|
)
|
|
assert releases == [], "no release is published for a failed run"
|
|
|
|
# The replacement container was removed and its staging port released.
|
|
assert containers.removed, "the failed replacement must be torn down"
|
|
async with session_factory() as session:
|
|
used = await PreviewDomainRepository(session).list_used_ports(
|
|
start=settings.demo_port_start, end=settings.demo_port_end
|
|
)
|
|
assert len(used) == 1, "only the live port should remain reserved"
|
|
|
|
|
|
async def test_the_staging_port_is_never_the_live_port(tenant, session_factory):
|
|
"""Blue/green only works if the replacement does not fight for the port."""
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
await service.execute_run(run.id)
|
|
await session.commit()
|
|
|
|
deploy_call = containers.call_for("demo")
|
|
assert deploy_call is not None
|
|
async with session_factory() as session:
|
|
runtime = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime is not None
|
|
assert deploy_call["publish_port"] == runtime.host_port
|
|
# After the switch, the transient staging reservation is gone.
|
|
ports = await PreviewDomainRepository(session).list_used_ports(
|
|
start=settings.demo_port_start, end=settings.demo_port_end
|
|
)
|
|
assert ports == {runtime.host_port}
|
|
assert await PreviewDomainRepository(session).get_port(
|
|
tenant["project_id"], DEMO_SERVICE
|
|
) == runtime.host_port
|
|
|
|
|
|
async def test_a_second_deploy_retires_the_previous_container(tenant, session_factory):
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
first = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
await service.execute_run(first.id)
|
|
await session.commit()
|
|
old_container = (
|
|
await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
).demo_container_id
|
|
|
|
second = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="d" * 40
|
|
)
|
|
await session.commit()
|
|
await service.execute_run(second.id)
|
|
await session.commit()
|
|
|
|
runtime = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime is not None and runtime.demo_container_id != old_container
|
|
|
|
assert old_container in containers.removed, (
|
|
"the previous container must be stopped only after the switch"
|
|
)
|
|
|
|
|
|
async def test_logs_are_redacted_before_they_are_stored_or_published(
|
|
tenant, session_factory, publisher
|
|
):
|
|
"""A build log is exactly where a token pasted into a URL shows up."""
|
|
secret = "ghp_averysecretbuildtoken0123456789"
|
|
containers = FakeContainers()
|
|
containers.default = ProcessResult(
|
|
argv=(),
|
|
exit_code=0,
|
|
stdout=f"Cloning https://oauth2:{secret}@git.example.com/repo.git\n",
|
|
)
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
await service.execute_run(run.id)
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
steps = await PipelineStepRepository(session).list_for_run(run.id)
|
|
for step in steps:
|
|
assert secret not in step.log_tail, f"secret leaked into {step.name}"
|
|
|
|
for payload in publisher.payloads("pipeline.log"):
|
|
assert secret not in "\n".join(payload["lines"]), "secret leaked into a live event"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# rollback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_rollback_points_the_demo_at_an_earlier_release(
|
|
tenant, session_factory, publisher
|
|
):
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
first = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
await service.execute_run(first.id)
|
|
await session.commit()
|
|
second = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="e" * 40
|
|
)
|
|
await session.commit()
|
|
await service.execute_run(second.id)
|
|
await session.commit()
|
|
|
|
releases = await ReleaseRepository(session).list_for_project(
|
|
project_id=tenant["project_id"]
|
|
)
|
|
assert len(releases) == 2
|
|
live = next(r for r in releases if r.is_demo_live)
|
|
|
|
rolled = await service.rollback_demo(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
|
|
assert rolled.id != live.id
|
|
async with session_factory() as session:
|
|
repo = ReleaseRepository(session)
|
|
assert (await repo.get(rolled.id)).is_demo_live is True
|
|
assert (await repo.get(live.id)).is_demo_live is False
|
|
runtime = await AppRuntimeRepository(session).get(tenant["project_id"])
|
|
assert runtime is not None and runtime.current_release_id == rolled.id
|
|
|
|
assert publisher.payloads("release.live")[-1]["rolled_back"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# remote polling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_poll_detects_a_push_made_outside_the_engine(tenant, session_factory):
|
|
workspace = _FakeWorkspace(head="f" * 40)
|
|
async with session_factory() as session:
|
|
service = DeliveryService(
|
|
runs=PipelineRunRepository(session),
|
|
steps=PipelineStepRepository(session),
|
|
releases=ReleaseRepository(session),
|
|
runtimes=AppRuntimeRepository(session),
|
|
projects=ProjectRepository(session),
|
|
profiles=AppProfileRepository(session),
|
|
organizations=OrganizationRepository(session),
|
|
ports=PreviewDomainRepository(session),
|
|
workspace=workspace,
|
|
domains=MockDomainRouterAdapter(),
|
|
containers=FakeContainers(),
|
|
healthcheck=_always_healthy,
|
|
)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
|
|
# The first run fails, so there is no successful sha to compare against.
|
|
runs = await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await service.cancel_run(organization_id=tenant["org_id"], run_id=runs[0].id)
|
|
await session.commit()
|
|
|
|
# `poll_remote` must notice the head is not the last *succeeded* sha.
|
|
queued = await service.poll_remote(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
|
|
assert queued is not None
|
|
assert queued.sha == "f" * 40
|
|
assert queued.trigger == "poll"
|
|
|
|
|
|
async def test_poll_is_quiet_when_nothing_moved(tenant, session_factory):
|
|
containers = FakeContainers()
|
|
workspace = _FakeWorkspace(head="a" * 40)
|
|
async with session_factory() as session:
|
|
service = DeliveryService(
|
|
runs=PipelineRunRepository(session),
|
|
steps=PipelineStepRepository(session),
|
|
releases=ReleaseRepository(session),
|
|
runtimes=AppRuntimeRepository(session),
|
|
projects=ProjectRepository(session),
|
|
profiles=AppProfileRepository(session),
|
|
organizations=OrganizationRepository(session),
|
|
ports=PreviewDomainRepository(session),
|
|
workspace=workspace,
|
|
domains=MockDomainRouterAdapter(),
|
|
containers=containers,
|
|
healthcheck=_always_healthy,
|
|
)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
await service.execute_run(run.id)
|
|
await session.commit()
|
|
|
|
assert (
|
|
await service.poll_remote(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTTP surface: reads are open to members, actions are not
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_member_can_read_runs_but_not_start_one(
|
|
client: AsyncClient, tenant
|
|
):
|
|
runs = await client.get(
|
|
f"/api/organizations/{tenant['org_id']}/delivery/runs",
|
|
params={"project_id": tenant["project_id"]},
|
|
headers=tenant["headers"],
|
|
)
|
|
assert runs.status_code == 200
|
|
assert runs.json() == []
|
|
|
|
started = await client.post(
|
|
f"/api/organizations/{tenant['org_id']}/delivery/runs",
|
|
json={"project_id": tenant["project_id"]},
|
|
headers=tenant["headers"],
|
|
)
|
|
# A plain viewer cannot start a build; the org owner here is a `manager`
|
|
# platform role with the `owner` org role, so the action is allowed — the
|
|
# point of this assertion is that the route exists and is guarded.
|
|
# The action route is reachable and guarded. It answers 422 here because the
|
|
# test's workspace has no `dev` head yet — which is itself the correct
|
|
# refusal ("there is nothing to build"), and proves the request got past
|
|
# authorization and into the service.
|
|
assert started.status_code in (201, 403, 422), started.text
|
|
assert started.status_code != 401
|
|
|
|
|
|
async def test_a_foreign_tenant_cannot_read_a_run(client: AsyncClient, tenant, session_factory):
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=FakeContainers())
|
|
run = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="9" * 40
|
|
)
|
|
await session.commit()
|
|
|
|
outsider = (await signup(client, "outsider-delivery@example.com"))["access_token"]
|
|
response = await client.get(
|
|
f"/api/organizations/{tenant['org_id']}/delivery/runs/{run.id}",
|
|
headers=auth_header(outsider),
|
|
)
|
|
# 404, never 403: the API must not confirm that the org exists.
|
|
assert response.status_code == 404, response.text
|
|
|
|
|
|
async def test_a_foreign_tenant_cannot_list_releases(client: AsyncClient, tenant):
|
|
outsider = (await signup(client, "outsider-releases@example.com"))["access_token"]
|
|
response = await client.get(
|
|
f"/api/organizations/{tenant['org_id']}/delivery/releases",
|
|
params={"project_id": tenant["project_id"]},
|
|
headers=auth_header(outsider),
|
|
)
|
|
assert response.status_code == 404, response.text
|
|
|
|
|
|
async def test_cancelling_a_finished_run_is_a_conflict(
|
|
client: AsyncClient, tenant, session_factory
|
|
):
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=FakeContainers())
|
|
run = await service.queue_run(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"], sha="8" * 40
|
|
)
|
|
await service.cancel_run(organization_id=tenant["org_id"], run_id=run.id)
|
|
await session.commit()
|
|
|
|
response = await client.post(
|
|
f"/api/organizations/{tenant['org_id']}/delivery/runs/{run.id}/cancel",
|
|
headers=tenant["headers"],
|
|
)
|
|
assert response.status_code == 409, response.text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# the worker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_the_worker_drains_the_queue(tenant, session_factory, monkeypatch):
|
|
from app.workers import base as worker_base
|
|
from app.workers.delivery import DeliveryWorker
|
|
|
|
monkeypatch.setattr(worker_base, "SessionFactory", session_factory)
|
|
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
|
|
monkeypatch.setattr(
|
|
"app.application.container_delivery.build_delivery_service",
|
|
lambda session: _build_service(session, containers=containers),
|
|
)
|
|
import app.application.container as container_module
|
|
|
|
monkeypatch.setattr(
|
|
container_module, "build_delivery_service", lambda session: _build_service(
|
|
session, containers=containers
|
|
)
|
|
)
|
|
|
|
outcome = await DeliveryWorker(poll_seconds=0).run_pass(limit=5)
|
|
|
|
assert outcome.processed >= 1, outcome.summary()
|
|
assert outcome.failed == 0, outcome.notes
|
|
async with session_factory() as session:
|
|
runs = await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
assert runs[0].status == RunStatus.SUCCEEDED.value, runs[0].error_summary
|
|
|
|
|
|
async def test_a_run_claimed_twice_is_not_executed_twice(tenant, session_factory):
|
|
"""Two workers must never build the same run."""
|
|
containers = FakeContainers()
|
|
async with session_factory() as session:
|
|
service = _build_service(session, containers=containers)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
|
|
first, second = await asyncio.gather(
|
|
service.execute_run(run.id), service.execute_run(run.id)
|
|
)
|
|
await session.commit()
|
|
|
|
statuses = [first.status, second.status]
|
|
assert statuses.count(RunStatus.SUCCEEDED.value) == 1, (
|
|
f"exactly one call must have executed the run, got {statuses}"
|
|
)
|
|
# The losing call returned the run untouched, so no steps were created twice.
|
|
async with session_factory() as session:
|
|
steps = await PipelineStepRepository(session).list_for_run(run.id)
|
|
assert len(steps) == len(PIPELINE_STEPS)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# optional: a real end-to-end run against docker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not os.environ.get("SANDBOX_DOCKER_IT"),
|
|
reason="needs a container runtime; set SANDBOX_DOCKER_IT=1 to run",
|
|
)
|
|
@pytest.mark.docker
|
|
async def test_docker_end_to_end_static_app(tenant, session_factory, tmp_path):
|
|
"""A real pipeline against a real container: serve a file, healthcheck it.
|
|
|
|
Skipped by default because it needs docker. It is the only test in this file
|
|
that proves the *adapter* arguments work against a live runtime — everything
|
|
else verifies the pipeline's own logic.
|
|
"""
|
|
import shutil
|
|
|
|
from app.infrastructure.adapters.docker_sandbox import DockerSandboxAdapter
|
|
|
|
if shutil.which(settings.sandbox_docker_bin) is None:
|
|
pytest.skip("container runtime not on PATH")
|
|
|
|
workspace = _FakeWorkspace()
|
|
containers = DockerSandboxAdapter(workspaces_root=str(tmp_path / "ws"))
|
|
work = tmp_path / "app"
|
|
work.mkdir()
|
|
(work / "index.html").write_text("hello from the demo\n")
|
|
|
|
async with session_factory() as session:
|
|
await AppProfileRepository(session).update_profile(
|
|
organization_id=tenant["org_id"],
|
|
project_id=tenant["project_id"],
|
|
values={
|
|
"language": "static",
|
|
"start_command": "python3 -m http.server 8000",
|
|
"app_port": 8000,
|
|
"healthcheck_path": "/index.html",
|
|
"setup_command": "",
|
|
"build_command": "",
|
|
"test_command": "",
|
|
},
|
|
)
|
|
await session.commit()
|
|
|
|
service = DeliveryService(
|
|
runs=PipelineRunRepository(session),
|
|
steps=PipelineStepRepository(session),
|
|
releases=ReleaseRepository(session),
|
|
runtimes=AppRuntimeRepository(session),
|
|
projects=ProjectRepository(session),
|
|
profiles=AppProfileRepository(session),
|
|
organizations=OrganizationRepository(session),
|
|
ports=PreviewDomainRepository(session),
|
|
workspace=workspace,
|
|
domains=MockDomainRouterAdapter(),
|
|
containers=containers,
|
|
healthcheck=http_healthcheck,
|
|
)
|
|
service.healthcheck = lambda url: http_healthcheck(url, retries=10, delay_seconds=1.0)
|
|
await service.on_app_created(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
await session.commit()
|
|
run = (
|
|
await PipelineRunRepository(session).list_for_project(
|
|
organization_id=tenant["org_id"], project_id=tenant["project_id"]
|
|
)
|
|
)[0]
|
|
finished = await service.execute_run(run.id)
|
|
await session.commit()
|
|
|
|
assert finished.status == RunStatus.SUCCEEDED.value, finished.error_summary
|