chore(infra): prod 환경 배포

.github/scripts/deploy_services.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""변경된 빌드 입력으로 배포할 서비스 목록을 계산한다."""
+
+import argparse
+import json
+from pathlib import Path
+import re
+import subprocess
+
+SERVICES = ("gateway", "identity", "application", "notification", "configuration", "admin", "observability")
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def select_services(paths, root=ROOT):
+    # 문서와 CI 설정을 제외하고 실제 빌드 입력의 영향 범위를 모은다.
+    selected = set()
+    for name in paths:
+        path = Path(name)
+        parts = path.parts
+        if not parts or path.suffix.lower() in {".md", ".rst"}:
+            continue
+        if parts[0] in {".github", ".codex", ".agents", "documents", "docs"} or "src/test" in path.as_posix():
+            continue
+        if len(parts) > 1 and parts[0] == "systems" and parts[1] in SERVICES:
+            selected.add(parts[1])
+        elif parts[0] in {"systems", "contracts", "packages", "tools", "scripts"} or name in {
+            "Dockerfile", ".dockerignore", "prepare-images.sh", "BUILD.bazel", "MODULE.bazel",
+            "MODULE.bazel.lock", ".bazelrc", ".bazelversion",
+        } or path.suffix in {".bzl", ".bazel"} or path.name.startswith(".bazel"):
+            return list(SERVICES)
+
+    # 다른 서비스 모듈을 참조하는 빌드 대상도 함께 갱신한다.
+    dependents = {service: set() for service in SERVICES}
+    for consumer in SERVICES:
+        directory = root / "systems" / consumer
+        for build_file in directory.rglob("*"):
+            if build_file.name != "BUILD.bazel" and build_file.suffix != ".bzl":
+                continue
+            for dependency in re.findall(r"//systems/([^/\s:\"]+)", build_file.read_text(encoding="utf-8")):
+                if dependency in dependents:
+                    dependents[dependency].add(consumer)
+    pending = list(selected)
+    while pending:
+        for consumer in dependents[pending.pop()] - selected:
+            selected.add(consumer)
+            pending.append(consumer)
+    return [service for service in SERVICES if service in selected]
+
+
+def main():
+    # push 전체 범위의 삭제·이동 전후 경로를 포함하고 수동 실행은 전체 배포한다.
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--before", default="")
+    parser.add_argument("--after", default="HEAD")
+    parser.add_argument("--all", action="store_true")
+    args = parser.parse_args()
+    if args.all or not args.before or set(args.before) == {"0"}:
+        services = list(SERVICES)
+    elif subprocess.run(
+        ["git", "cat-file", "-e", args.before + "^{commit}"], cwd=ROOT,
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    ).returncode:
+        # 강제 push 등으로 이전 커밋이 없으면 누락 방지를 위해 전체 배포한다.
+        services = list(SERVICES)
+    else:
+        paths = subprocess.check_output(
+            ["git", "diff", "--name-only", "--no-renames", "-z", args.before, args.after], cwd=ROOT,
+        ).decode("utf-8").split("\0")
+        services = select_services(filter(None, paths))
+    print("services=" + json.dumps(services, separators=(",", ":")))
+    print("service_args=" + " ".join(services))
+    print("has_services=" + str(bool(services)).lower())
+
+
+if __name__ == "__main__":
+    main()

.github/scripts/test_deploy_only.py

@@ -0,0 +1,131 @@
+"""수동 배포 워크플로의 실제 입력 처리 및 원격 명령 검증."""
+
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import textwrap
+import unittest
+
+
+WORKFLOW = Path(__file__).parents[1] / "workflows" / "deploy-only.yml"
+ALL = "gateway identity application notification configuration admin observability"
+
+
+def step_script(name):
+    section = WORKFLOW.read_text(encoding="utf-8").split(f"- name: {name}\n", 1)[1]
+    lines = section.split("run: |\n", 1)[1].splitlines()
+    body = []
+    for line in lines:
+        if line.strip() and not line.startswith("          "):
+            break
+        body.append(line)
+    return textwrap.dedent("\n".join(body))
+
+
+class DeployOnlyTest(unittest.TestCase):
+    def prepare(self, **changes):
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "output"
+            env = dict(os.environ, BRANCH="develop", SERVICE="identity", INPUT_IMAGE_TAG="", STAG_HOSTS='["stag.example.com"]', PROD_HOSTS='["prod.example.com"]', GITHUB_OUTPUT=str(output))
+            env.update(changes)
+            result = subprocess.run([sys.executable, "-c", step_script("서비스와 이미지 태그 검증")], env=env, capture_output=True, text=True)
+            values = dict(line.split("=", 1) for line in output.read_text(encoding="utf-8").splitlines()) if output.exists() else {}
+            return result, values
+
+    def test_branch_defaults_and_environment_hosts(self):
+        for branch, stage in [("develop", "stag"), ("main", "prod")]:
+            with self.subTest(branch=branch):
+                result, output = self.prepare(BRANCH=branch)
+                self.assertEqual(0, result.returncode, result.stderr)
+                self.assertEqual(stage, output["stage"])
+                self.assertEqual(stage, output["image_tag"])
+                self.assertEqual([f"{stage}.example.com"], json.loads(output["hosts"]))
+                self.assertEqual("identity", output["services"])
+
+    def test_all_and_explicit_tag(self):
+        result, output = self.prepare(SERVICE="all", INPUT_IMAGE_TAG="release-2026.10_19")
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual(ALL, output["services"])
+        self.assertEqual("release-2026.10_19", output["image_tag"])
+
+    def test_each_supported_service(self):
+        for service in ALL.split():
+            with self.subTest(service=service):
+                result, output = self.prepare(SERVICE=service)
+                self.assertEqual(0, result.returncode, result.stderr)
+                self.assertEqual(service, output["services"])
+
+    def test_invalid_inputs_rejected_before_outputs(self):
+        cases = [{"BRANCH": "feature/test"}, {"SERVICE": "identity notification"}, {"SERVICE": "identity;id"}]
+        cases += [{"INPUT_IMAGE_TAG": tag} for tag in ["bad tag", "x;id", "$(id)", "-bad", "a" * 129, "tag\nother"]]
+        cases += [{"STAG_HOSTS": hosts} for hosts in ['[]', '{}', '["host;id"]', '[42]']]
+        for changes in cases:
+            with self.subTest(changes=changes):
+                result, output = self.prepare(**changes)
+                self.assertNotEqual(0, result.returncode)
+                self.assertEqual({}, output)
+
+    def test_tag_length_boundary(self):
+        result, output = self.prepare(INPUT_IMAGE_TAG="a" * 128)
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual("a" * 128, output["image_tag"])
+
+    @unittest.skipUnless(shutil.which("bash"), "Bash 실행 환경 필요")
+    def test_remote_commands_only_deploy_selected_service(self):
+        script = step_script("기존 이미지 배포")
+        mocks = '''
+ssh() { bash -c "${@: -1}"; }
+docker() { printf 'DOCKER'; printf ' <%s>' "$@"; printf '\\n'; }
+export -f docker
+'''
+        for stage in ["stag", "prod"]:
+            with self.subTest(stage=stage):
+                env = dict(os.environ, GHCR_USERNAME="test", GHCR_TOKEN="fake", DEPLOY_USER="tester", HOST="mock", DEPLOY_PATH=".", IMAGE_TAG="existing-sha", SERVICES="identity", STAGE=stage)
+                result = subprocess.run([shutil.which("bash"), "-s"], input=mocks + script, env=env, capture_output=True, text=True)
+                self.assertEqual(0, result.returncode, result.stderr)
+                self.assertRegex(result.stdout, r"DOCKER <compose>[^\n]* <pull> <identity>")
+                self.assertRegex(result.stdout, r"DOCKER <compose>[^\n]* <up> <-d> <--no-deps> <identity>")
+                self.assertNotIn("<build>", result.stdout)
+                self.assertNotIn("<push>", result.stdout)
+                self.assertNotIn("<gateway>", result.stdout)
+                self.assertEqual(stage == "stag", "DOCKER <compose> <restart> <nginx>" in result.stdout)
+
+    @unittest.skipUnless(shutil.which("bash") and shutil.which("docker"), "Bash와 Docker Compose 필요")
+    def test_real_compose_preserves_dollar_environment_and_relative_mount(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "compose.yaml").write_text('name: deploy-fixture\nservices:\n  identity:\n    image: original:latest\n    environment:\n      SECRET: "literal$$TOKEN"\n    volumes:\n      - ./data:/data\n', encoding="utf-8")
+            baseline = subprocess.run([shutil.which("docker"), "compose", "-f", str(root / "compose.yaml"), "config", "--format", "json"], capture_output=True, text=True, check=True)
+            original_environment = json.loads(baseline.stdout)["services"]["identity"]["environment"]
+            mocks = '''
+ssh() { bash -c "${@: -1}"; }
+docker() {
+  if [ "$1" = login ]; then return 0; fi
+  local args=("$@")
+  if [[ " ${args[*]} " == *" pull "* ]]; then
+    "$DOCKER_EXE" "${args[@]:0:${#args[@]}-2}" config --format json
+  elif [[ " ${args[*]} " == *" up "* ]]; then
+    return 0
+  else
+    "$DOCKER_EXE" "$@"
+  fi
+}
+export -f docker
+'''
+            env = dict(os.environ, GHCR_USERNAME="test", GHCR_TOKEN="fake", DEPLOY_USER="tester", HOST="mock", DEPLOY_PATH=root.as_posix(), IMAGE_TAG="existing-sha", SERVICES="identity", STAGE="prod", DOCKER_EXE=Path(shutil.which("docker")).as_posix())
+            result = subprocess.run([shutil.which("bash"), "-s"], input=mocks + step_script("기존 이미지 배포"), env=env, capture_output=True, text=True)
+            self.assertEqual(0, result.returncode, result.stderr)
+            config = json.loads(result.stdout)
+            self.assertEqual("deploy-fixture", config["name"])
+            identity = config["services"]["identity"]
+            self.assertEqual("ghcr.io/entrydsm/entrydsm-identity:existing-sha", identity["image"])
+            self.assertEqual(original_environment, identity["environment"])
+            self.assertEqual((root / "data").resolve(), Path(identity["volumes"][0]["source"]).resolve())
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/scripts/test_deploy_services.py

@@ -0,0 +1,92 @@
+"""배포 대상 선정의 요구사항 기반 회귀 테스트."""
+
+import importlib.util
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+SCRIPT = Path(__file__).with_name("deploy_services.py")
+SPEC = importlib.util.spec_from_file_location("deploy_services", SCRIPT)
+deploy_services = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(deploy_services)
+
+ALL = ["gateway", "identity", "application", "notification", "configuration", "admin", "observability"]
+
+
+class DeployServicesTest(unittest.TestCase):
+    def select(self, *paths):
+        return deploy_services.select_services(paths)
+
+    def test_documentation_only_skips_deployment(self):
+        self.assertEqual([], self.select("README.md", "docs/auth.rst", "systems/identity/README.md"))
+
+    def test_unrelated_workflow_only_skips_deployment(self):
+        self.assertEqual([], self.select(".github/workflows/lint.yml", ".github/ISSUE_TEMPLATE/bug.yml"))
+
+    def test_no_changes_skips_deployment(self):
+        self.assertEqual([], self.select())
+
+    def test_identity_change_builds_only_identity(self):
+        self.assertEqual(["identity"], self.select("systems/identity/identity-application/src/main/kotlin/AuthService.kt"))
+
+    def test_multiple_services_are_selected_once(self):
+        self.assertEqual(["identity", "notification"], self.select(
+            "systems/notification/notification-domain/src/main/kotlin/Notification.kt",
+            "systems/identity/identity-domain/src/main/kotlin/User.kt",
+            "systems/identity/identity-domain/src/main/kotlin/Token.kt"))
+
+    def test_documentation_does_not_hide_code_change(self):
+        self.assertEqual(["identity"], self.select("README.md", "systems/identity/BUILD.bazel"))
+
+    def test_common_code_affects_all_services(self):
+        self.assertEqual(ALL, self.select("systems/common/src/main/kotlin/Shared.kt"))
+
+    def test_root_build_inputs_affect_all_services(self):
+        for path in ["Dockerfile", ".dockerignore", "MODULE.bazel", "MODULE.bazel.lock", ".bazelrc", ".bazelversion", "kotlin.bzl", "BUILD.bazel", "prepare-images.sh"]:
+            with self.subTest(path=path):
+                self.assertEqual(ALL, self.select(path))
+
+    def test_shared_input_directories_affect_all_services(self):
+        for path in ["contracts/example.proto", "packages/shared/BUILD.bazel", "tools/build.sh"]:
+            with self.subTest(path=path):
+                self.assertEqual(ALL, self.select(path))
+
+    def test_deleted_service_file_still_selects_service(self):
+        self.assertEqual(["identity"], self.select("systems/identity/deleted/Old.kt"))
+
+    def test_unrelated_directory_does_not_match_service_prefix(self):
+        self.assertNotEqual(["identity"], self.select("systems/identity-other/src/Main.kt"))
+
+    def test_transitive_consumers_are_included(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            for service, dependency in [("application", "configuration"), ("admin", "application")]:
+                target = root / "systems" / service / "BUILD.bazel"
+                target.parent.mkdir(parents=True)
+                target.write_text(f'alias(name="main", actual="//systems/{dependency}/domain:main")', encoding="utf-8")
+            self.assertEqual(["application", "configuration", "admin"], deploy_services.select_services(
+                ["systems/configuration/domain/src/main/kotlin/Config.kt"], root=root))
+
+    def test_initial_push_and_manual_deployment_select_all(self):
+        for arguments in [[], ["--all"], ["--before", "0" * 40]]:
+            with self.subTest(arguments=arguments):
+                result = subprocess.run([sys.executable, str(SCRIPT), *arguments], capture_output=True, text=True, check=True)
+                outputs = dict(line.split("=", 1) for line in result.stdout.splitlines())
+                self.assertEqual(ALL, json.loads(outputs["services"]))
+                self.assertEqual(" ".join(ALL), outputs["service_args"])
+                self.assertEqual("true", outputs["has_services"])
+
+    def test_unavailable_revision_deploys_all_instead_of_silently_skipping(self):
+        result = subprocess.run([sys.executable, str(SCRIPT), "--before", "not-a-valid-deploy-revision"], capture_output=True, text=True)
+        self.assertEqual(0, result.returncode)
+        outputs = dict(line.split("=", 1) for line in result.stdout.splitlines())
+        self.assertEqual(ALL, json.loads(outputs["services"]))
+        self.assertEqual("true", outputs["has_services"])
+
+
+if __name__ == "__main__":
+    unittest.main()