mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
* [env-bran] EnvSyncEngine: orchestrator-side prod→dev cascade (default-off) - EnvSyncEngine mirrors CiWatchEngine: cascade ladder_pairs top-down via GitHub merges API; clean→auto-push lower rung, conflict→one sync PR + tracked MAIN_PM task + stop. Never pushes prod (lower rung is never prod by construction). - GitService.sync_env_branch (merges API) + open_sync_pr (idempotent) + _env_merge_status/_post_sync_pr helpers (constants for 201/204/409). - TaskService.ENV_SYNC_SOURCE + list_open_env_sync_tasks (per-repo dedup). - config env_sync_enabled/_interval_seconds(1800)/_max_open_tasks(3)/_max_per_cycle(1). - Orchestrator 4-touch registration + _load_env_sync_set (ladder+token opt-in). - Feature-flags card + settings FEATURE_FLAGS entry for ROBOCO_ENV_SYNC_ENABLED. * [env-bran] Panel: environment ladder editor + types + validation - EnvironmentRung type + environments on Project/ProjectCreate/ProjectUpdate. - EnvironmentLadderEditor (plain useState, add/remove/up-down reorder, head/ prod labels) reused by create + edit project dialogs. - validateLadder (non-empty name+branch, no duplicate branches) shared, toast.error on submit; empty editor => null => inherits default_branch shim. - default_branch input kept with override-hint; API client passthrough. - 6 unit tests for validateLadder. * [env-bran] Tests + gate green: env ladder, EnvSyncEngine, promotion chain - tests/unit/models/test_env_branches.py: shim, head/prod, ladder_pairs, promotion_chain, normalize (20 tests) - tests/integration/services/test_env_sync_engine.py: cascade clean/conflict/ missing_ref/tokenless/degenerate/caps/dedup/disabled (9 tests, DB) - tests/integration/test_migration_env_branches.py: 073 defaults null + round-trip - tests/unit/services/test_release_executor*.py: add env_chain=[] to _ReleaseContext constructions (promotion_chain field is now required) - tests/unit/runtime/test_orchestrator_shutdown_drain.py: register _env_sync_task in the stop()-drain fixture (new named background loop) - roboco/services/git.py: revert _project_head_branch rename back to _project_default_branch (modify-in-place per plan); the rename in the consumers commit broke ~15 unit-test mocks that bind the original name - roboco/services/env_sync_engine.py + models/env_branches.py: ruff format - roboco/api/schemas/project.py: trailing-newline format Backend gate green (13013 passed / 439 skipped), mypy clean, ruff clean. Panel gate green (typecheck/lint/522 tests). * [env-bran] fix: add env_chain to _ReleaseContext in e2e smoke (CI red) The release-executor promotion_chain change made _ReleaseContext.env_chain required. I fixed the three unit/release test files but missed the construction in tests/e2e_smoke/test_background_engines.py:98 — my local gate ran 'mypy roboco/' (excludes tests/) and I skipped 'make e2e-smoke', so CI's mypy-on-tests + the e2e runtime job caught it instead of me. Verified locally with the CI-equivalent gates: uv run mypy roboco/ tests/ -> 1170 files, clean ROBOCO_E2E_SMOKE=1 uv run pytest tests/e2e_smoke -> 50 passed, 1 skipped * [env-bran] fix: extract _ensure_prod_fetched to clear xenon rank C (CI red) _production_assess grew past xenon --max-absolute B (rank C) when the env-branches prod-tip fetch added an if/try/except branch. Extracted the fetch-with-fallback into _ensure_prod_fetched (degan+fetch paths), moved _run_git to the module-level import. Local make quality green (all gates incl xenon/vulture/deptry/import-linter/foundation-check). --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
231 lines
6.9 KiB
Python
231 lines
6.9 KiB
Python
"""env_branches — shim, ladder pairs, head/prod resolution, normalization.
|
|
|
|
Pure domain helpers (pydantic-only); no DB. The read-time shim synthesizes a
|
|
degenerate single-branch ladder from ``default_branch`` when ``environments``
|
|
is null, so every consumer behaves identically until a real ladder is declared.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from roboco.models.env_branches import (
|
|
EnvRung,
|
|
effective_environments,
|
|
head_branch,
|
|
ladder_pairs,
|
|
normalize_environments,
|
|
prod_branch,
|
|
promotion_chain,
|
|
)
|
|
|
|
|
|
class _Proj:
|
|
"""Duck-typed project row (matches Project + ProjectTable surface)."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
default_branch: str = "master",
|
|
environments: list[dict[str, str]] | None = None,
|
|
) -> None:
|
|
self.default_branch = default_branch
|
|
self.environments = environments
|
|
|
|
|
|
# --- effective_environments shim ------------------------------------------
|
|
|
|
|
|
def test_null_environments_synthesizes_degenerate_ladder() -> None:
|
|
proj = _Proj(default_branch="slave")
|
|
rungs = effective_environments(proj)
|
|
assert [(r.name, r.branch) for r in rungs] == [("head", "slave"), ("prod", "slave")]
|
|
|
|
|
|
def test_empty_environments_synthesizes_degenerate_ladder() -> None:
|
|
proj = _Proj(default_branch="master", environments=[])
|
|
rungs = effective_environments(proj)
|
|
assert [(r.name, r.branch) for r in rungs] == [
|
|
("head", "master"),
|
|
("prod", "master"),
|
|
]
|
|
|
|
|
|
def test_missing_default_branch_falls_back_to_master() -> None:
|
|
proj = _Proj(default_branch="") # falsy default_branch
|
|
assert head_branch(proj) == "master"
|
|
assert prod_branch(proj) == "master"
|
|
|
|
|
|
def test_set_environments_returned_as_is_preserving_order() -> None:
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "qa", "branch": "qa"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
rungs = effective_environments(proj)
|
|
assert [r.branch for r in rungs] == ["dev", "qa", "master"]
|
|
assert all(isinstance(r, EnvRung) for r in rungs)
|
|
|
|
|
|
# --- head_branch / prod_branch --------------------------------------------
|
|
|
|
|
|
def test_head_and_prod_single_rung() -> None:
|
|
proj = _Proj(environments=[{"name": "prod", "branch": "master"}])
|
|
assert head_branch(proj) == "master"
|
|
assert prod_branch(proj) == "master"
|
|
|
|
|
|
def test_head_and_prod_two_rungs() -> None:
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "slave"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
assert head_branch(proj) == "slave"
|
|
assert prod_branch(proj) == "master"
|
|
|
|
|
|
def test_head_and_prod_four_rungs() -> None:
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "qa", "branch": "qa"},
|
|
{"name": "stag", "branch": "stag"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
assert head_branch(proj) == "dev"
|
|
assert prod_branch(proj) == "master"
|
|
|
|
|
|
# --- ladder_pairs (prod -> head cascade) -----------------------------------
|
|
|
|
|
|
def test_ladder_pairs_empty_for_single_rung() -> None:
|
|
assert (
|
|
ladder_pairs(_Proj(environments=[{"name": "prod", "branch": "master"}])) == []
|
|
)
|
|
|
|
|
|
def test_ladder_pairs_two_rungs() -> None:
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
pairs = ladder_pairs(proj)
|
|
assert [(u.branch, lower.branch) for u, lower in pairs] == [("master", "dev")]
|
|
|
|
|
|
def test_ladder_pairs_four_rungs_top_down() -> None:
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "qa", "branch": "qa"},
|
|
{"name": "stag", "branch": "stag"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
# [(prod, stag), (stag, qa), (qa, head)] — merge upper into lower.
|
|
pairs = ladder_pairs(proj)
|
|
assert [(u.branch, lower.branch) for u, lower in pairs] == [
|
|
("master", "stag"),
|
|
("stag", "qa"),
|
|
("qa", "dev"),
|
|
]
|
|
|
|
|
|
def test_ladder_pairs_lower_rung_is_never_prod() -> None:
|
|
"""The cascade's target is never prod by construction — only CEO merges prod."""
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "qa", "branch": "qa"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
prod_name = prod_branch(proj)
|
|
for _upper, lower in ladder_pairs(proj):
|
|
assert lower.branch != prod_name
|
|
|
|
|
|
# --- promotion_chain (full-chain release promotion) ----------------------
|
|
|
|
|
|
def test_promotion_chain_empty_for_degenerate_ladder() -> None:
|
|
"""head==prod => nothing to promote (no-op release promotion)."""
|
|
assert promotion_chain(_Proj(default_branch="master")) == []
|
|
assert (
|
|
promotion_chain(_Proj(environments=[{"name": "prod", "branch": "master"}]))
|
|
== []
|
|
)
|
|
|
|
|
|
def test_promotion_chain_two_rungs() -> None:
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
assert promotion_chain(proj) == ["dev"]
|
|
|
|
|
|
def test_promotion_chain_four_rungs_head_first_excluding_prod() -> None:
|
|
proj = _Proj(
|
|
environments=[
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "qa", "branch": "qa"},
|
|
{"name": "stag", "branch": "stag"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
)
|
|
assert promotion_chain(proj) == ["dev", "qa", "stag"]
|
|
|
|
|
|
# --- normalize_environments ------------------------------------------------
|
|
|
|
|
|
def test_normalize_none_returns_none() -> None:
|
|
assert normalize_environments(None) is None
|
|
assert normalize_environments([]) is None
|
|
|
|
|
|
def test_normalize_strips_and_preserves_order() -> None:
|
|
out = normalize_environments(
|
|
[{"name": " head ", "branch": " dev "}, {"name": "prod", "branch": "master"}]
|
|
)
|
|
assert out == [
|
|
{"name": "head", "branch": "dev"},
|
|
{"name": "prod", "branch": "master"},
|
|
]
|
|
|
|
|
|
def test_normalize_rejects_empty_name() -> None:
|
|
with pytest.raises(ValueError, match="non-empty name"):
|
|
normalize_environments([{"name": "", "branch": "dev"}])
|
|
|
|
|
|
def test_normalize_rejects_empty_branch() -> None:
|
|
with pytest.raises(ValueError, match="non-empty name"):
|
|
normalize_environments(
|
|
[{"name": "head", "branch": " "}]
|
|
) # branch trimmed to empty
|
|
|
|
|
|
def test_normalize_rejects_duplicate_branch() -> None:
|
|
with pytest.raises(ValueError, match="duplicate environment branch"):
|
|
normalize_environments(
|
|
[{"name": "head", "branch": "dev"}, {"name": "prod", "branch": "dev"}]
|
|
)
|
|
|
|
|
|
def test_normalize_accepts_envrung_models() -> None:
|
|
out = normalize_environments([EnvRung(name="head", branch="dev")])
|
|
assert out == [{"name": "head", "branch": "dev"}]
|