diff --git a/.github/scripts/saas-dev/README.md b/.github/scripts/saas-dev/README.md deleted file mode 100644 index fdc096eadea..00000000000 --- a/.github/scripts/saas-dev/README.md +++ /dev/null @@ -1,128 +0,0 @@ -# SaaS dev workflow consumers (SNP-836) - -This is an **opt-in repair for the existing SaaS dev host**, not a general Cloud -worker configuration or a Staging/Production deployment change. It never starts, -stops, scales, or restarts containers and never enqueues, deletes, or consumes jobs. - -## Why - -The host's `/home/ubuntu/langgenius/docker-compose.yaml` explicitly sets the main -worker's `CELERY_QUEUES`. This overrides the image defaults and omits -`workflow_professional`, `workflow_team`, and `workflow_sandbox`. An allowed -Webhook can therefore return HTTP 200 while its workflow remains queued. - -The repair appends only these three queues to the existing main worker setting. -It preserves the existing ten queues, their order, comments, quoting, all other -file bytes, file owner/group/mode, the two worker replicas, and every other -service. It intentionally rejects a different or ambiguous baseline for review. - -Do **not** remove `CELERY_QUEUES`, enable all default queues, or change the shared -image entrypoint. Dedicated Cloud workers elsewhere must not start competing for -other queues. In the 2026-09-20 dev audit, `schedule_poller`, `schedule_executor`, -and `trigger_refresh_publisher` each already contained more than 51,000 messages. -Those queues and their data are outside this repair's scope. - -## Prerequisites and dry-run - -Requirements: Python 3.10+, PyYAML, Docker Compose v2, and access to the existing -`langgenius-api-1` and two main worker containers. PyYAML was already installed on -the audited host; no new package or credential is required. The existing API -image supplies its Redis client. Resolved Compose configuration and broker -credentials are processed only in memory, never printed or saved by the repair. - -From the repository root, run the script over SSH. Omitting `--apply` is a -dry-run: the target and services are unchanged. A temporary validation candidate -in the same directory is removed after validation. - -```sh -ssh -T -i /Users/yansongzhang/.ssh/id_rsa -o IdentitiesOnly=yes \ - ubuntu@ec2-100-30-236-187.compute-1.amazonaws.com \ - 'python3 - --target saas-dev' \ - < .github/scripts/saas-dev/fix_workflow_queues.py -``` - -Every dry-run and apply performs these fail-closed preflight checks: - -- The target is precisely `/home/ubuntu/langgenius/docker-compose.yaml`, a regular - file, not a symlink or hardlink. `services.worker.environment.CELERY_QUEUES` - occurs exactly once and is an explicit, single-line string. -- The resolved candidate differs only in that one environment value. The worker - is Cloud, and `CELERY_WORKER_QUEUES` does not override the intended setting. -- Exactly two running main worker containers match the existing Compose project - and file. Their real PID 1 `-Q` arguments match their environment and contain - only the reviewed original queues and the three approved workflow queues. -- Compose API/worker and running API/worker broker URLs match exactly in memory. - Redis prefixes, Sentinel, custom TLS, or a different topology fail closed; - review these explicitly instead of checking the wrong broker/namespace. -- All three workflow queues, including Redis priority suffix keys, have **zero** - messages. Both broker `unacked` and `unacked_index` must also be **zero**. - -If any guard fails, stop and inspect the cause. Do not purge queues, remove -messages, restart consumers, or weaken the guard to make the repair pass. A -nonzero `unacked` count may be legitimate in-flight work; wait for it to finish -and rerun the read-only check. The check is a snapshot, not a lock on producers; -recheck immediately before the planned worker deployment. - -## Apply and deploy - -Review the dry-run's old/new queue lists and `before_sha256`. Apply requires that -exact SHA256 plus the explicit target and apply flag: - -```sh -ssh -T -i /Users/yansongzhang/.ssh/id_rsa -o IdentitiesOnly=yes \ - ubuntu@ec2-100-30-236-187.compute-1.amazonaws.com \ - 'python3 - --target saas-dev --apply --expected-sha256 REVIEWED_SHA256' \ - < .github/scripts/saas-dev/fix_workflow_queues.py -``` - -The script validates first, verifies the original content again, writes an exact -same-directory backup with the original owner/group/mode, then atomically -replaces the target and verifies the resulting bytes. It prints the concrete -backup path, hashes, queue names, and `services_restarted: false`. Treat the -backup as sensitive: it contains the original Compose configuration and must not -be committed or attached to an issue. Record only its path and hash in the -deployment report. An already-patched file is a no-op without another backup. - -This repository's `deploy-saas.yml` executes the configured -`SSH_SCRIPT_SAAS_DEV`; the audited script pulls and recreates the normal API, -Web, main worker, priority worker, and plugin worker services from this existing -host Compose file. It does not replace that file, so the queue change persists -through subsequent `deploy/saas` deployments. Keep the repair and API changes on -the original PR and synchronize to `deploy/saas` through the normal reviewed -workflow. Do not change the deployment script variable, pipeline settings, -Staging/Production manifests, or unrelated service configurations for this fix. - -## Regression checklist - -1. After the normal deployment, inspect the two main worker PID 1 queue arguments: - the original ten queues plus exactly the three workflow queues must be present; - replica count and other workers' queue arguments must be unchanged. -2. Create a disposable, no-model Webhook → End workflow. Bind a denying IP policy: - a request must return 403 and produce no workflow run or trigger log. -3. Permit only the test source IP and send **one** Webhook request. HTTP 200 alone - is insufficient: verify the trigger log and workflow run reach `succeeded`, - with zero model tokens and no residual workflow queue message. -4. Restore a denying policy, verify 403 and unchanged run count; confirm the - existing Web/API/MCP access-control regressions remain correct. -5. Clean up only the disposable test resources through their normal lifecycle. - Do not clear shared queues, historical messages, or unrelated user data. -6. Confirm none of the excluded schedule/refresh queues gained a consumer. Their - lengths can grow naturally while the beat process continues producing tasks; - growth is not a reason to consume or purge them during this fix. - -If rolling back, first check for running/pending workflow work and obtain approval -for any service operation. Inspect the exact backup path printed by the apply; -verify that the live file still has the recorded `after_sha256` before restoring -it, so another person's intervening edits are not overwritten. Restore the -backup with owner/group/mode preserved, validate Compose without printing its -secret-bearing output, and use the normal worker deployment procedure. This -script deliberately provides no automatic destructive rollback or queue purge. - -## Local tests - -```sh -uv run --project api pytest api/tests/unit_tests/commands/test_fix_workflow_queues.py -``` - -Tests use temporary synthetic Compose files and mocked broker responses. They -do not contact dev, invoke Docker, or use real credentials. diff --git a/.github/scripts/saas-dev/fix_workflow_queues.py b/.github/scripts/saas-dev/fix_workflow_queues.py deleted file mode 100644 index 6f1fb763749..00000000000 --- a/.github/scripts/saas-dev/fix_workflow_queues.py +++ /dev/null @@ -1,445 +0,0 @@ -#!/usr/bin/env python3 -"""Opt-in, SaaS-dev-only repair for SNP-836; never starts a consumer. - -Requires Python 3.10+, PyYAML, Docker Compose v2, and the existing API container. -The host's PyYAML is used only to locate YAML scalar spans; the file is not dumped -or reformatted, and neither resolved configuration nor credentials are printed. -""" - -import argparse -import copy -import hashlib -import json -import os -from pathlib import Path -import re -import stat -import subprocess -import tempfile - -import yaml -from yaml.nodes import MappingNode, ScalarNode - -TARGET = Path("/home/ubuntu/langgenius/docker-compose.yaml") -BASE_QUEUES = ( - "dataset", - "mail", - "ops_trace", - "app_deletion", - "pipeline", - "workflow_storage", - "conversation", - "workflow_based_app_execution", - "evaluation", - "new_agent_beta", -) -WORKFLOW_QUEUES = ("workflow_professional", "workflow_team", "workflow_sandbox") - - -class UnsafeConfiguration(ValueError): - pass - - -def field(node, name): - if not isinstance(node, MappingNode): - raise UnsafeConfiguration("Expected an explicit YAML mapping") - matches = [ - (key, value) - for key, value in node.value - if isinstance(key, ScalarNode) and key.value == name - ] - if len(matches) != 1: - raise UnsafeConfiguration(f"Expected exactly one explicit {name} field") - key, value = matches[0] - if value.start_mark.index < key.end_mark.index: - raise UnsafeConfiguration(f"Aliased {name} field requires manual review") - return value - - -def plan_patch(before): - """Change only the unique, direct worker queue scalar; return UTF-8 bytes.""" - text = before.decode("utf-8") - try: - root = yaml.compose(text, Loader=yaml.SafeLoader) - except yaml.YAMLError: - raise UnsafeConfiguration( - "Invalid YAML; details suppressed to protect credentials" - ) from None - worker = field(field(root, "services"), "worker") - environment = field(worker, "environment") - queue = field(environment, "CELERY_QUEUES") - if not isinstance(queue, ScalarNode) or queue.tag != "tag:yaml.org,2002:str": - raise UnsafeConfiguration("CELERY_QUEUES must be an explicit string") - if ( - queue.style not in (None, "'", '"') - or queue.start_mark.line != queue.end_mark.line - ): - raise UnsafeConfiguration( - "Multiline or non-scalar queue syntax requires manual review" - ) - # Aliases can refer to a scalar outside this service. Never mutate its source. - if not ( - worker.start_mark.index - < queue.start_mark.index - < queue.end_mark.index - <= worker.end_mark.index - ): - raise UnsafeConfiguration("Queue aliases outside worker require manual review") - current = queue.value.split(",") - if any(not re.fullmatch(r"[a-z_]+", item) for item in current): - raise UnsafeConfiguration( - "Queue interpolation, whitespace, or unsupported syntax requires review" - ) - if len(current) != len(set(current)): - raise UnsafeConfiguration("Duplicate queue names require manual review") - if not set(BASE_QUEUES).issubset(current) or set(current) - set( - BASE_QUEUES + WORKFLOW_QUEUES - ): - raise UnsafeConfiguration( - "Worker queues differ from the reviewed SaaS dev baseline" - ) - missing = [item for item in WORKFLOW_QUEUES if item not in current] - if not missing: - return before, current, current - updated = current + missing - raw = text[queue.start_mark.index : queue.end_mark.index] - quoted = queue.style or "" - # Refuse anchors, tags, escapes and aliases instead of rewriting them. - if raw != quoted + queue.value + quoted: - raise UnsafeConfiguration("Decorated queue scalar requires manual review") - replacement = quoted + ",".join(updated) + quoted - after = ( - text[: queue.start_mark.index] + replacement + text[queue.end_mark.index :] - ).encode("utf-8") - if plan_patch(after)[0] != after: - raise UnsafeConfiguration("Candidate queue patch is not idempotent") - return after, current, updated - - -def sha256(content): - return hashlib.sha256(content).hexdigest() - - -def safe_command(command, *, input_text=None): - result = subprocess.run( - command, input=input_text, text=True, capture_output=True, timeout=45 - ) - if result.returncode: - raise UnsafeConfiguration( - "Read-only Docker validation failed; output suppressed to protect credentials" - ) - return result.stdout - - -def resolved_config(path): - return json.loads( - safe_command( - [ - "docker", - "compose", - "--project-directory", - str(TARGET.parent), - "-f", - str(path), - "config", - "--format", - "json", - ] - ) - ) - - -def validate_resolved(before, after, old_queues, new_queues): - """Catch inherited overrides, YAML ambiguity, and accidental unrelated edits.""" - environment = before["services"]["worker"]["environment"] - if environment.get("DEPLOYMENT_EDITION") != "CLOUD": - raise UnsafeConfiguration("Expected CLOUD SaaS dev worker") - if environment.get("CELERY_WORKER_QUEUES"): - raise UnsafeConfiguration("CELERY_WORKER_QUEUES overrides CELERY_QUEUES") - if environment.get("CELERY_QUEUES") != ",".join(old_queues): - raise UnsafeConfiguration( - "Resolved queues differ from the explicit queue scalar" - ) - expected = copy.deepcopy(before) - expected["services"]["worker"]["environment"]["CELERY_QUEUES"] = ",".join( - new_queues - ) - if after != expected: - raise UnsafeConfiguration( - "Resolved candidate changes more than worker CELERY_QUEUES" - ) - - -BROKER_CHECK = r""" -import json, os -from redis import Redis -try: - if os.environ.get("REDIS_KEY_PREFIX"): - raise ValueError("Unsupported broker namespace") - client = Redis.from_url(os.environ["CELERY_BROKER_URL"], socket_timeout=10) - counts = {} - for queue in ("workflow_professional", "workflow_team", "workflow_sandbox"): - name = queue.encode() - keys = {name, *client.scan_iter(match=name + b"\x06\x16*")} - counts[queue] = sum(client.llen(key) for key in keys) - counts["unacked"] = client.hlen("unacked") - counts["unacked_index"] = client.zcard("unacked_index") - print(json.dumps(counts)) -except Exception: - print("Broker check failed; details suppressed", file=__import__("sys").stderr) - raise SystemExit(1) -""" - - -def validate_broker_environments(config, runtime_environments): - """Compare credentials only in memory; unsupported broker layouts fail shut.""" - expected = config["services"]["worker"]["environment"].get("CELERY_BROKER_URL") - if not isinstance(expected, str) or not expected.startswith("redis://"): - raise UnsafeConfiguration("Only the reviewed direct Redis broker is supported") - environments = [ - config["services"][service]["environment"] for service in ("api", "worker") - ] - for environment in environments + list(runtime_environments): - if environment.get("CELERY_BROKER_URL") != expected: - raise UnsafeConfiguration( - "API/worker broker mismatch; refusing an unreliable backlog check" - ) - if environment.get("REDIS_KEY_PREFIX"): - raise UnsafeConfiguration( - "Redis key prefixes require a separate namespace-aware review" - ) - for setting in ("CELERY_USE_SENTINEL", "BROKER_USE_SSL"): - if str(environment.get(setting, "false")).lower() not in ( - "", - "false", - "none", - "0", - ): - raise UnsafeConfiguration( - "Redis Sentinel or custom TLS requires separate review" - ) - - -def check_runtime_broker(config): - workers = safe_command( - [ - "docker", - "ps", - "-q", - "--filter", - "label=com.docker.compose.project=langgenius", - "--filter", - "label=com.docker.compose.service=worker", - ] - ).split() - if ( - len(workers) != 2 - or config["services"]["worker"].get("deploy", {}).get("replicas") != 2 - ): - raise UnsafeConfiguration("Expected the reviewed two-worker SaaS dev topology") - containers = json.loads( - safe_command(["docker", "inspect", "langgenius-api-1", *workers]) - ) - environments = [] - for container in containers: - labels = container["Config"].get("Labels", {}) - service = labels.get("com.docker.compose.service") - if ( - not container["State"]["Running"] - or service not in ("api", "worker") - or labels.get("com.docker.compose.project") != "langgenius" - or labels.get("com.docker.compose.project.config_files") != str(TARGET) - ): - raise UnsafeConfiguration( - "Running containers do not match the reviewed Compose deployment" - ) - environment = dict(item.split("=", 1) for item in container["Config"]["Env"]) - environments.append(environment) - if service == "worker": - if environment.get("CELERY_WORKER_QUEUES"): - raise UnsafeConfiguration( - "Runtime CELERY_WORKER_QUEUES overrides the reviewed queue setting" - ) - argv = safe_command( - ["docker", "exec", container["Id"], "cat", "/proc/1/cmdline"] - ).split("\0") - if ( - argv.count("-Q") != 1 - or argv.index("-Q") + 1 >= len(argv) - or argv[argv.index("-Q") + 1] != environment.get("CELERY_QUEUES") - ): - raise UnsafeConfiguration( - "Runtime worker argv and queue environment differ" - ) - queues = environment["CELERY_QUEUES"].split(",") - if set(queues) - set(BASE_QUEUES + WORKFLOW_QUEUES) or not set( - BASE_QUEUES - ).issubset(queues): - raise UnsafeConfiguration( - "Runtime queues differ from the reviewed baseline" - ) - validate_broker_environments(config, environments) - - -def broker_check(config): - check_runtime_broker(config) - counts = json.loads( - safe_command( - [ - "docker", - "exec", - "-i", - "langgenius-api-1", - "/app/api/.venv/bin/python", - "-", - ], - input_text=BROKER_CHECK, - ) - ) - expected = set(WORKFLOW_QUEUES) | {"unacked", "unacked_index"} - if set(counts) != expected or any( - type(value) is not int or value != 0 for value in counts.values() - ): - raise UnsafeConfiguration( - "Workflow backlog or unacked tasks exist; do not start new consumers" - ) - return counts - - -def regular_file(path): - metadata = path.lstat() - if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: - raise UnsafeConfiguration( - "Target must be a regular non-symlink, non-hardlinked file" - ) - return metadata - - -def private_copy(path, content, metadata, prefix): - fd, name = tempfile.mkstemp(prefix=prefix, dir=path.parent) - copy_path = Path(name) - try: - with os.fdopen(fd, "wb") as handle: - os.fchown(handle.fileno(), metadata.st_uid, metadata.st_gid) - os.fchmod(handle.fileno(), stat.S_IMODE(metadata.st_mode)) - handle.write(content) - handle.flush() - os.fsync(handle.fileno()) - return copy_path - except BaseException: - copy_path.unlink(missing_ok=True) - raise - - -def apply_patch(path, before, after, expected_sha256, validate, preflight): - """No service operations. Keep an exact backup and fail closed on drift.""" - import fcntl - - if sha256(before) != expected_sha256: - raise UnsafeConfiguration("Expected SHA256 does not match the reviewed dry-run") - metadata = regular_file(path) - with path.open("rb") as original: - fcntl.flock(original.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - if path.read_bytes() != before: - raise UnsafeConfiguration("Compose changed since the dry-run") - candidate = private_copy(path, after, metadata, ".snp836-candidate-") - try: - validate(candidate) - preflight() - if before == after: - return None - backup = private_copy( - path, before, metadata, "docker-compose.yaml.snp836-backup-" - ) - if backup.read_bytes() != before: - raise UnsafeConfiguration("Backup verification failed") - if path.read_bytes() != before or path.stat().st_ino != metadata.st_ino: - raise UnsafeConfiguration( - "Compose changed before replacement; backup retained" - ) - os.replace(candidate, path) - if path.read_bytes() != after: - raise UnsafeConfiguration( - f"Post-write verification failed; restore backup {backup}" - ) - return backup - finally: - candidate.unlink(missing_ok=True) - - -def main(argv=None): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--target", required=True, choices=["saas-dev"]) - parser.add_argument( - "--apply", - action="store_true", - help="Write only the reviewed queue change; never restart", - ) - parser.add_argument( - "--expected-sha256", help="Required with --apply; copy from a fresh dry-run" - ) - args = parser.parse_args(argv) - if args.apply and not args.expected_sha256: - parser.error("--apply requires --expected-sha256") - try: - metadata = regular_file(TARGET) - before = TARGET.read_bytes() - after, old_queues, new_queues = plan_patch(before) - resolved_before = resolved_config(TARGET) - - def validate(candidate): - validate_resolved( - resolved_before, resolved_config(candidate), old_queues, new_queues - ) - - if args.apply: - backup = apply_patch( - TARGET, - before, - after, - args.expected_sha256, - validate, - lambda: broker_check(resolved_before), - ) - else: - # Compose reads secrets, so keep even a temporary candidate private. - candidate = private_copy(TARGET, after, metadata, ".snp836-candidate-") - try: - validate(candidate) - broker_check(resolved_before) - finally: - candidate.unlink(missing_ok=True) - backup = None - print( - json.dumps( - { - "target": "saas-dev", - "path": str(TARGET), - "mode": "apply" if args.apply else "dry-run", - "changed": before != after, - "before_sha256": sha256(before), - "after_sha256": sha256(after), - "queues_before": old_queues, - "queues_after": new_queues, - "backup": str(backup) if backup else None, - "services_restarted": False, - }, - indent=2, - ) - ) - except ( - UnsafeConfiguration, - OSError, - KeyError, - ValueError, - subprocess.SubprocessError, - ) as exc: - # Do not leak subprocess output, YAML snippets, broker URLs, or environment. - message = ( - str(exc) if isinstance(exc, UnsafeConfiguration) else type(exc).__name__ - ) - parser.exit(1, f"Refused: {message}\n") - - -if __name__ == "__main__": - main() diff --git a/api/tests/unit_tests/commands/test_fix_workflow_queues.py b/api/tests/unit_tests/commands/test_fix_workflow_queues.py deleted file mode 100644 index 26c12c15426..00000000000 --- a/api/tests/unit_tests/commands/test_fix_workflow_queues.py +++ /dev/null @@ -1,396 +0,0 @@ -"""Unit tests for the opt-in SaaS-dev workflow queue repair script.""" - -import copy -import importlib.util -import stat -import tempfile -import unittest -from collections.abc import Callable -from pathlib import Path -from typing import NotRequired, TypedDict, override -from unittest.mock import patch - -import pytest - -REPO_ROOT = Path(__file__).resolve().parents[4] -SCRIPT_PATH = REPO_ROOT / ".github" / "scripts" / "saas-dev" / "fix_workflow_queues.py" -SPEC = importlib.util.spec_from_file_location("fix_workflow_queues", SCRIPT_PATH) -assert SPEC is not None -assert SPEC.loader is not None -repair = importlib.util.module_from_spec(SPEC) -SPEC.loader.exec_module(repair) - -BASE = ",".join(repair.BASE_QUEUES) -FULL = ",".join(repair.BASE_QUEUES + repair.WORKFLOW_QUEUES) - - -def compose(value: str = BASE, before_worker: str = "", after_worker: str = "", newline: str = "\n") -> bytes: - return ( - ( - "# Keep this comment and unrelated settings\n" - "x-common: &common\n LOG_LEVEL: INFO\n" - "services:\n" + before_worker + " worker:\n image: langgenius/dify-api:deploy-saas\n" - " environment:\n <<: *common\n DEPLOYMENT_EDITION: CLOUD\n" - f" CELERY_QUEUES: {value} # deliberately explicit\n" - " OTHER_CONFIG: unchanged\n" + after_worker + "volumes:\n storage:\n" - ) - .replace("\n", newline) - .encode() - ) - - -OTHER_WORKERS = ( - " priority_worker:\n environment:\n CELERY_QUEUES: priority_dataset,priority_pipeline\n" - " worker-plugin_autoupgrade:\n environment:\n CELERY_QUEUES: plugin\n" - " worker_beat:\n environment:\n MODE: beat\n" -) - - -class ApplyOverrides(TypedDict, total=False): - validate: Callable[[Path], object] - preflight: Callable[[], object] - - -class ServiceConfig(TypedDict): - environment: dict[str, str] - deploy: NotRequired[dict[str, int]] - - -class ComposeConfig(TypedDict): - services: dict[str, ServiceConfig] - - -class ContainerConfig(TypedDict): - Env: list[str] - Labels: dict[str, str] - - -class Container(TypedDict): - Id: str - State: dict[str, bool] - Config: ContainerConfig - - -class PatchPlanTests(unittest.TestCase): - def test_preserves_every_other_byte_for_worker_at_different_positions(self) -> None: - for before_worker, after_worker in ((OTHER_WORKERS, ""), ("", OTHER_WORKERS)): - for quote in ("", "'", '"'): - for newline in ("\n", "\r\n"): - with self.subTest(before=bool(before_worker), quote=quote, newline=newline): - before = compose(quote + BASE + quote, before_worker, after_worker, newline) - after, old, new = repair.plan_patch(before) - assert after == before.replace(BASE.encode(), FULL.encode(), 1) - assert old == list(repair.BASE_QUEUES) - assert new == list(repair.BASE_QUEUES + repair.WORKFLOW_QUEUES) - - def test_only_appends_missing_names_and_keeps_existing_order(self) -> None: - current = ",".join(reversed(repair.BASE_QUEUES)) + ",workflow_team" - after, old, new = repair.plan_patch(compose(current)) - assert new == old + ["workflow_professional", "workflow_sandbox"] - assert ",".join(new).encode() in after - - def test_idempotent_when_all_queues_already_exist(self) -> None: - before = compose(FULL) - after, old, new = repair.plan_patch(before) - assert after == before - assert old == new - - def test_rejects_duplicate_unknown_missing_or_interpolated_queues(self) -> None: - for value in ( - BASE + ",dataset", - BASE + ",schedule_poller", - BASE.replace("dataset,", ""), - "${CELERY_QUEUES}", - BASE + ", workflow_team", - '"' + BASE + '\\n"', - ): - with ( - self.subTest(value=value), - pytest.raises(repair.UnsafeConfiguration), - ): - repair.plan_patch(compose(value)) - - def test_missing_and_ambiguous_target_fields_fail_closed(self) -> None: - original = compose() - for content in ( - original.replace(b" worker:", b" another_worker:"), - original.replace(b"CELERY_QUEUES:", b"OTHER_QUEUES:"), - original.replace(b" environment:", b" environment: {}\n environment:"), - original.replace(b" worker:", b" worker: {}\n worker:"), - original.replace(b"services:", b"services: {}\nservices:"), - original.replace( - b" CELERY_QUEUES:", - b" CELERY_QUEUES: other\n CELERY_QUEUES:", - ), - ): - with ( - self.subTest(content=content), - pytest.raises(repair.UnsafeConfiguration), - ): - repair.plan_patch(content) - - def test_rejects_non_scalar_list_block_and_alias_values(self) -> None: - values = ("[dataset,mail]", "true", "|\n " + BASE, ">\n " + BASE) - for value in values: - with ( - self.subTest(value=value), - pytest.raises(repair.UnsafeConfiguration), - ): - repair.plan_patch(compose(value)) - outside_alias = b"x-queues: &queues " + BASE.encode() + b"\n" + compose("*queues") - with pytest.raises(repair.UnsafeConfiguration): - repair.plan_patch(outside_alias) - worker_alias = ( - b"x-worker: &worker\n environment:\n CELERY_QUEUES: " - + BASE.encode() - + b"\nservices:\n worker: *worker\n" - ) - with pytest.raises(repair.UnsafeConfiguration): - repair.plan_patch(worker_alias) - - -class ResolvedValidationTests(unittest.TestCase): - @override - def setUp(self) -> None: - self.before = { - "services": { - "worker": { - "environment": { - "DEPLOYMENT_EDITION": "CLOUD", - "CELERY_QUEUES": BASE, - } - }, - "priority_worker": {"environment": {"CELERY_QUEUES": "priority_dataset,priority_pipeline"}}, - } - } - self.after = copy.deepcopy(self.before) - self.after["services"]["worker"]["environment"]["CELERY_QUEUES"] = FULL - - def test_accepts_exact_single_field_change(self) -> None: - repair.validate_resolved( - self.before, - self.after, - repair.BASE_QUEUES, - repair.BASE_QUEUES + repair.WORKFLOW_QUEUES, - ) - - def test_rejects_overrides_non_cloud_or_unrelated_changes(self) -> None: - for field, value in ( - ("CELERY_WORKER_QUEUES", "override"), - ("DEPLOYMENT_EDITION", "COMMUNITY"), - ("CELERY_QUEUES", "other"), - ): - before = copy.deepcopy(self.before) - before["services"]["worker"]["environment"][field] = value - with ( - self.subTest(field=field), - pytest.raises(repair.UnsafeConfiguration), - ): - repair.validate_resolved( - before, - self.after, - repair.BASE_QUEUES, - repair.BASE_QUEUES + repair.WORKFLOW_QUEUES, - ) - self.after["services"]["priority_worker"]["environment"]["CELERY_QUEUES"] += ",workflow_professional" - with pytest.raises(repair.UnsafeConfiguration): - repair.validate_resolved( - self.before, - self.after, - repair.BASE_QUEUES, - repair.BASE_QUEUES + repair.WORKFLOW_QUEUES, - ) - - -class ApplyTests(unittest.TestCase): - @override - def setUp(self) -> None: - self.temp = tempfile.TemporaryDirectory() - self.addCleanup(self.temp.cleanup) - self.path = Path(self.temp.name) / "docker-compose.yaml" - self.before = compose() - self.after = repair.plan_patch(self.before)[0] - self.path.write_bytes(self.before) - self.path.chmod(0o640) - - def apply( - self, - validate: Callable[[Path], object] = lambda _candidate: None, - preflight: Callable[[], object] = lambda: None, - expected: str | None = None, - ) -> Path | None: - return repair.apply_patch( - self.path, - self.before, - self.after, - expected or repair.sha256(self.before), - validate, - preflight, - ) - - def test_backup_content_permissions_and_exact_change(self) -> None: - backup = self.apply() - assert backup is not None - assert backup.read_bytes() == self.before - assert self.path.read_bytes() == self.after - assert stat.S_IMODE(backup.stat().st_mode) == 0o640 - assert stat.S_IMODE(self.path.stat().st_mode) == 0o640 - assert list(self.path.parent.glob(".snp836-candidate-*")) == [] - - def test_noop_does_not_write_or_create_backup(self) -> None: - original_inode = self.path.stat().st_ino - backup = repair.apply_patch( - self.path, - self.before, - self.before, - repair.sha256(self.before), - lambda _candidate: None, - lambda: None, - ) - assert backup is None - assert self.path.stat().st_ino == original_inode - assert list(self.path.parent.glob("*backup*")) == [] - - def test_hash_and_before_content_guards(self) -> None: - with pytest.raises(repair.UnsafeConfiguration): - self.apply(expected="0" * 64) - self.path.write_bytes(self.before + b"# concurrent update\n") - with pytest.raises(repair.UnsafeConfiguration): - self.apply() - assert list(self.path.parent.glob("*backup*")) == [] - - def test_validation_or_backlog_failure_does_not_write(self) -> None: - def refuse(*_args: object) -> None: - raise repair.UnsafeConfiguration("unsafe") - - overrides: tuple[ApplyOverrides, ...] = ({"validate": refuse}, {"preflight": refuse}) - for arguments in overrides: - with ( - self.subTest(arguments=arguments), - pytest.raises(repair.UnsafeConfiguration), - ): - self.apply(**arguments) - assert self.path.read_bytes() == self.before - assert list(self.path.parent.glob("*backup*")) == [] - assert list(self.path.parent.glob(".snp836-candidate-*")) == [] - - def test_concurrent_edit_after_validation_is_not_overwritten(self) -> None: - concurrent = self.before + b"# operator edit\n" - with pytest.raises(repair.UnsafeConfiguration): - self.apply(validate=lambda _candidate: self.path.write_bytes(concurrent)) - assert self.path.read_bytes() == concurrent - - def test_symlink_is_rejected(self) -> None: - alias = self.path.parent / "alias" - alias.symlink_to(self.path) - with pytest.raises(repair.UnsafeConfiguration): - repair.regular_file(alias) - - -class BrokerPreflightTests(unittest.TestCase): - @override - def setUp(self) -> None: - self.env = {"CELERY_BROKER_URL": "redis://fake-fixture:6379/1"} - self.config: ComposeConfig = { - "services": { - "api": {"environment": self.env.copy()}, - "worker": {"environment": self.env.copy()}, - } - } - - def test_matching_plain_broker_is_accepted(self) -> None: - repair.validate_broker_environments(self.config, [self.env.copy(), self.env.copy(), self.env.copy()]) - - def test_namespace_or_broker_mismatch_fails_at_every_layer(self) -> None: - for layer in ("api", "worker", "runtime"): - for setting, value in ( - ("REDIS_KEY_PREFIX", "isolated"), - ("CELERY_BROKER_URL", "redis://different-fixture:6379/2"), - ("CELERY_USE_SENTINEL", "true"), - ("BROKER_USE_SSL", "true"), - ): - with self.subTest(layer=layer, setting=setting): - config = copy.deepcopy(self.config) - runtime = [self.env.copy()] - target = runtime[0] if layer == "runtime" else config["services"][layer]["environment"] - target[setting] = value - with pytest.raises(repair.UnsafeConfiguration): - repair.validate_broker_environments(config, runtime) - - def test_any_workflow_backlog_unacked_or_invalid_reply_blocks_apply(self) -> None: - empty = dict.fromkeys(repair.WORKFLOW_QUEUES, 0) - empty.update(unacked=0, unacked_index=0) - with ( - patch.object(repair, "check_runtime_broker"), - patch.object(repair, "safe_command") as command, - ): - command.return_value = repair.json.dumps(empty) - assert repair.broker_check(self.config) == empty - for key in empty: - command.return_value = repair.json.dumps({**empty, key: 1}) - with ( - self.subTest(key=key), - pytest.raises(repair.UnsafeConfiguration), - ): - repair.broker_check(self.config) - invalid_replies: tuple[dict[str, int | bool], ...] = ( - {}, - {**empty, "unknown": 0}, - {**empty, "unacked": False}, - ) - for invalid in invalid_replies: - command.return_value = repair.json.dumps(invalid) - with pytest.raises(repair.UnsafeConfiguration): - repair.broker_check(self.config) - - def test_runtime_requires_exact_topology_labels_and_real_worker_queues(self) -> None: - self.config["services"]["worker"]["deploy"] = {"replicas": 2} - containers: list[Container] = [] - for index, service in enumerate(("api", "worker", "worker")): - containers.append( - { - "Id": f"fixture-{index}", - "State": {"Running": True}, - "Config": { - "Env": [ - "CELERY_BROKER_URL=" + self.env["CELERY_BROKER_URL"], - "CELERY_QUEUES=" + BASE, - ], - "Labels": { - "com.docker.compose.service": service, - "com.docker.compose.project": "langgenius", - "com.docker.compose.project.config_files": str(repair.TARGET), - }, - }, - } - ) - responses = [ - "fixture-1\nfixture-2\n", - repair.json.dumps(containers), - "celery\0worker\0-Q\0" + BASE + "\0", - "celery\0worker\0-Q\0" + BASE + "\0", - ] - with patch.object(repair, "safe_command", side_effect=responses): - repair.check_runtime_broker(self.config) - with patch.object(repair, "safe_command", return_value="one-worker-only\n"): - with pytest.raises(repair.UnsafeConfiguration): - repair.check_runtime_broker(self.config) - with patch.object( - repair, - "safe_command", - side_effect=responses[:2] + ["celery\0-Q\0wrong_queue\0"], - ): - with pytest.raises(repair.UnsafeConfiguration): - repair.check_runtime_broker(self.config) - containers[0]["Config"]["Labels"]["com.docker.compose.project.config_files"] = "/different/compose.yaml" - with patch.object( - repair, - "safe_command", - side_effect=[responses[0], repair.json.dumps(containers)], - ): - with pytest.raises(repair.UnsafeConfiguration): - repair.check_runtime_broker(self.config) - - -if __name__ == "__main__": - unittest.main()