fix(worker): delegate production ddl to cd
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 59s
CD Pipeline / build-and-deploy (push) Successful in 7m9s
CD Pipeline / post-deploy-checks (push) Successful in 1m53s

This commit is contained in:
ogt
2026-07-11 20:12:02 +08:00
parent 99e69001e9
commit 8424af5343
4 changed files with 50 additions and 9 deletions

View File

@@ -8,10 +8,13 @@ Runtime bootstrap guard tests.
from __future__ import annotations
import inspect
from collections.abc import Awaitable
from pathlib import Path
from typing import Any
import pytest
import yaml
from sqlalchemy.exc import DBAPIError
@@ -63,6 +66,24 @@ class _ConnectionBudgetEngine:
raise RuntimeError('too many connections for role "awoooi"')
def test_production_worker_delegates_bootstrap_ddl_to_cd() -> None:
from src.workers import signal_worker
root = Path(__file__).resolve().parents[3]
manifest = yaml.safe_load(
(root / "k8s/awoooi-prod/08-deployment-worker.yaml").read_text(
encoding="utf-8"
)
)
container = manifest["spec"]["template"]["spec"]["containers"][0]
env = {item["name"]: item.get("value") for item in container["env"]}
source = inspect.getsource(signal_worker._main)
assert env["DATABASE_BOOTSTRAP_DDL_ENABLED"] == "false"
assert "if settings.DATABASE_BOOTSTRAP_DDL_ENABLED" in source
assert 'owner="cd_migration_preflight"' in source
def test_get_engine_uses_database_pool_budget(monkeypatch):
from src.db import base as db_base