From 343fc364820ea4d8ea16597db4ae68aefa083c4f Mon Sep 17 00:00:00 2001 From: Yansong Zhang <916125788@qq.com> Date: Tue, 22 Sep 2026 19:37:58 +0800 Subject: [PATCH] test(api): move SaaS dev queue repair tests into API suite --- .github/scripts/saas-dev/README.md | 2 +- .../commands}/test_fix_workflow_queues.py | 254 +++++++++--------- 2 files changed, 135 insertions(+), 121 deletions(-) rename {.github/scripts/saas-dev => api/tests/unit_tests/commands}/test_fix_workflow_queues.py (61%) diff --git a/.github/scripts/saas-dev/README.md b/.github/scripts/saas-dev/README.md index 109dcad9167..fdc096eadea 100644 --- a/.github/scripts/saas-dev/README.md +++ b/.github/scripts/saas-dev/README.md @@ -121,7 +121,7 @@ script deliberately provides no automatic destructive rollback or queue purge. ## Local tests ```sh -python3 -m unittest discover -s .github/scripts/saas-dev -p 'test_*.py' -v +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 diff --git a/.github/scripts/saas-dev/test_fix_workflow_queues.py b/api/tests/unit_tests/commands/test_fix_workflow_queues.py similarity index 61% rename from .github/scripts/saas-dev/test_fix_workflow_queues.py rename to api/tests/unit_tests/commands/test_fix_workflow_queues.py index ec3fec85f77..26c12c15426 100644 --- a/.github/scripts/saas-dev/test_fix_workflow_queues.py +++ b/api/tests/unit_tests/commands/test_fix_workflow_queues.py @@ -1,14 +1,22 @@ +"""Unit tests for the opt-in SaaS-dev workflow queue repair script.""" + import copy import importlib.util -from pathlib import Path 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 -SPEC = importlib.util.spec_from_file_location( - "fix_workflow_queues", Path(__file__).with_name("fix_workflow_queues.py") -) +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) @@ -16,14 +24,12 @@ BASE = ",".join(repair.BASE_QUEUES) FULL = ",".join(repair.BASE_QUEUES + repair.WORKFLOW_QUEUES) -def compose(value=BASE, before_worker="", after_worker="", newline="\n"): +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" + "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" @@ -40,39 +46,56 @@ OTHER_WORKERS = ( ) +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): + 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 - ) + 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) - self.assertEqual( - after, before.replace(BASE.encode(), FULL.encode(), 1) - ) - self.assertEqual(old, list(repair.BASE_QUEUES)) - self.assertEqual( - new, list(repair.BASE_QUEUES + repair.WORKFLOW_QUEUES) - ) + 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): + 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)) - self.assertEqual(new, old + ["workflow_professional", "workflow_sandbox"]) - self.assertIn(",".join(new).encode(), after) + assert new == old + ["workflow_professional", "workflow_sandbox"] + assert ",".join(new).encode() in after - def test_idempotent_when_all_queues_already_exist(self): + def test_idempotent_when_all_queues_already_exist(self) -> None: before = compose(FULL) after, old, new = repair.plan_patch(before) - self.assertEqual(after, before) - self.assertEqual(old, new) + assert after == before + assert old == new - def test_rejects_duplicate_unknown_missing_or_interpolated_queues(self): + def test_rejects_duplicate_unknown_missing_or_interpolated_queues(self) -> None: for value in ( BASE + ",dataset", BASE + ",schedule_poller", @@ -83,18 +106,16 @@ class PatchPlanTests(unittest.TestCase): ): with ( self.subTest(value=value), - self.assertRaises(repair.UnsafeConfiguration), + pytest.raises(repair.UnsafeConfiguration), ): repair.plan_patch(compose(value)) - def test_missing_and_ambiguous_target_fields_fail_closed(self): + 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" environment:", b" environment: {}\n environment:"), original.replace(b" worker:", b" worker: {}\n worker:"), original.replace(b"services:", b"services: {}\nservices:"), original.replace( @@ -104,34 +125,33 @@ class PatchPlanTests(unittest.TestCase): ): with ( self.subTest(content=content), - self.assertRaises(repair.UnsafeConfiguration), + pytest.raises(repair.UnsafeConfiguration), ): repair.plan_patch(content) - def test_rejects_non_scalar_list_block_and_alias_values(self): + 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), - self.assertRaises(repair.UnsafeConfiguration), + pytest.raises(repair.UnsafeConfiguration), ): repair.plan_patch(compose(value)) - outside_alias = ( - b"x-queues: &queues " + BASE.encode() + b"\n" + compose("*queues") - ) - with self.assertRaises(repair.UnsafeConfiguration): + 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 self.assertRaises(repair.UnsafeConfiguration): + with pytest.raises(repair.UnsafeConfiguration): repair.plan_patch(worker_alias) class ResolvedValidationTests(unittest.TestCase): - def setUp(self): + @override + def setUp(self) -> None: self.before = { "services": { "worker": { @@ -140,17 +160,13 @@ class ResolvedValidationTests(unittest.TestCase): "CELERY_QUEUES": BASE, } }, - "priority_worker": { - "environment": { - "CELERY_QUEUES": "priority_dataset,priority_pipeline" - } - }, + "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): + def test_accepts_exact_single_field_change(self) -> None: repair.validate_resolved( self.before, self.after, @@ -158,7 +174,7 @@ class ResolvedValidationTests(unittest.TestCase): repair.BASE_QUEUES + repair.WORKFLOW_QUEUES, ) - def test_rejects_overrides_non_cloud_or_unrelated_changes(self): + def test_rejects_overrides_non_cloud_or_unrelated_changes(self) -> None: for field, value in ( ("CELERY_WORKER_QUEUES", "override"), ("DEPLOYMENT_EDITION", "COMMUNITY"), @@ -168,7 +184,7 @@ class ResolvedValidationTests(unittest.TestCase): before["services"]["worker"]["environment"][field] = value with ( self.subTest(field=field), - self.assertRaises(repair.UnsafeConfiguration), + pytest.raises(repair.UnsafeConfiguration), ): repair.validate_resolved( before, @@ -176,10 +192,8 @@ class ResolvedValidationTests(unittest.TestCase): repair.BASE_QUEUES, repair.BASE_QUEUES + repair.WORKFLOW_QUEUES, ) - self.after["services"]["priority_worker"]["environment"]["CELERY_QUEUES"] += ( - ",workflow_professional" - ) - with self.assertRaises(repair.UnsafeConfiguration): + self.after["services"]["priority_worker"]["environment"]["CELERY_QUEUES"] += ",workflow_professional" + with pytest.raises(repair.UnsafeConfiguration): repair.validate_resolved( self.before, self.after, @@ -189,7 +203,8 @@ class ResolvedValidationTests(unittest.TestCase): class ApplyTests(unittest.TestCase): - def setUp(self): + @override + def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.path = Path(self.temp.name) / "docker-compose.yaml" @@ -199,8 +214,11 @@ class ApplyTests(unittest.TestCase): self.path.chmod(0o640) def apply( - self, validate=lambda candidate: None, preflight=lambda: None, expected=None - ): + 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, @@ -210,79 +228,80 @@ class ApplyTests(unittest.TestCase): preflight, ) - def test_backup_content_permissions_and_exact_change(self): + def test_backup_content_permissions_and_exact_change(self) -> None: backup = self.apply() - self.assertEqual(backup.read_bytes(), self.before) - self.assertEqual(self.path.read_bytes(), self.after) - self.assertEqual(stat.S_IMODE(backup.stat().st_mode), 0o640) - self.assertEqual(stat.S_IMODE(self.path.stat().st_mode), 0o640) - self.assertEqual(list(self.path.parent.glob(".snp836-candidate-*")), []) + 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): + 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 _candidate: None, lambda: None, ) - self.assertIsNone(backup) - self.assertEqual(self.path.stat().st_ino, original_inode) - self.assertEqual(list(self.path.parent.glob("*backup*")), []) + 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): - with self.assertRaises(repair.UnsafeConfiguration): + 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 self.assertRaises(repair.UnsafeConfiguration): + with pytest.raises(repair.UnsafeConfiguration): self.apply() - self.assertEqual(list(self.path.parent.glob("*backup*")), []) + assert list(self.path.parent.glob("*backup*")) == [] - def test_validation_or_backlog_failure_does_not_write(self): - def refuse(*args): + def test_validation_or_backlog_failure_does_not_write(self) -> None: + def refuse(*_args: object) -> None: raise repair.UnsafeConfiguration("unsafe") - for arguments in ({"validate": refuse}, {"preflight": refuse}): + overrides: tuple[ApplyOverrides, ...] = ({"validate": refuse}, {"preflight": refuse}) + for arguments in overrides: with ( self.subTest(arguments=arguments), - self.assertRaises(repair.UnsafeConfiguration), + pytest.raises(repair.UnsafeConfiguration), ): self.apply(**arguments) - self.assertEqual(self.path.read_bytes(), self.before) - self.assertEqual(list(self.path.parent.glob("*backup*")), []) - self.assertEqual(list(self.path.parent.glob(".snp836-candidate-*")), []) + 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): + def test_concurrent_edit_after_validation_is_not_overwritten(self) -> None: concurrent = self.before + b"# operator edit\n" - with self.assertRaises(repair.UnsafeConfiguration): - self.apply(validate=lambda candidate: self.path.write_bytes(concurrent)) - self.assertEqual(self.path.read_bytes(), concurrent) + 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): + def test_symlink_is_rejected(self) -> None: alias = self.path.parent / "alias" alias.symlink_to(self.path) - with self.assertRaises(repair.UnsafeConfiguration): + with pytest.raises(repair.UnsafeConfiguration): repair.regular_file(alias) class BrokerPreflightTests(unittest.TestCase): - def setUp(self): + @override + def setUp(self) -> None: self.env = {"CELERY_BROKER_URL": "redis://fake-fixture:6379/1"} - self.config = { + self.config: ComposeConfig = { "services": { "api": {"environment": self.env.copy()}, "worker": {"environment": self.env.copy()}, } } - def test_matching_plain_broker_is_accepted(self): - repair.validate_broker_environments( - self.config, [self.env.copy(), self.env.copy(), 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): + 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"), @@ -290,44 +309,43 @@ class BrokerPreflightTests(unittest.TestCase): ("CELERY_USE_SENTINEL", "true"), ("BROKER_USE_SSL", "true"), ): - with ( - self.subTest(layer=layer, setting=setting), - self.assertRaises(repair.UnsafeConfiguration), - ): + 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 = runtime[0] if layer == "runtime" else config["services"][layer]["environment"] target[setting] = value - repair.validate_broker_environments(config, runtime) + with pytest.raises(repair.UnsafeConfiguration): + repair.validate_broker_environments(config, runtime) - def test_any_workflow_backlog_unacked_or_invalid_reply_blocks_apply(self): - empty = {queue: 0 for queue in repair.WORKFLOW_QUEUES} + 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) - self.assertEqual(repair.broker_check(self.config), 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), - self.assertRaises(repair.UnsafeConfiguration), + pytest.raises(repair.UnsafeConfiguration), ): repair.broker_check(self.config) - for invalid in ({}, {**empty, "unknown": 0}, {**empty, "unacked": False}): + 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 self.assertRaises(repair.UnsafeConfiguration): + with pytest.raises(repair.UnsafeConfiguration): repair.broker_check(self.config) - def test_runtime_requires_exact_topology_labels_and_real_worker_queues(self): + def test_runtime_requires_exact_topology_labels_and_real_worker_queues(self) -> None: self.config["services"]["worker"]["deploy"] = {"replicas": 2} - containers = [] + containers: list[Container] = [] for index, service in enumerate(("api", "worker", "worker")): containers.append( { @@ -341,9 +359,7 @@ class BrokerPreflightTests(unittest.TestCase): "Labels": { "com.docker.compose.service": service, "com.docker.compose.project": "langgenius", - "com.docker.compose.project.config_files": str( - repair.TARGET - ), + "com.docker.compose.project.config_files": str(repair.TARGET), }, }, } @@ -357,24 +373,22 @@ class BrokerPreflightTests(unittest.TestCase): 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 self.assertRaises(repair.UnsafeConfiguration): + 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 self.assertRaises(repair.UnsafeConfiguration): + with pytest.raises(repair.UnsafeConfiguration): repair.check_runtime_broker(self.config) - containers[0]["Config"]["Labels"]["com.docker.compose.project.config_files"] = ( - "/different/compose.yaml" - ) + 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 self.assertRaises(repair.UnsafeConfiguration): + with pytest.raises(repair.UnsafeConfiguration): repair.check_runtime_broker(self.config)