"""§v2-A3.5 — production deploy over SSH / SFTP / FTP / FTPS (D8).

Production is the one action the platform cannot undo, so the tests are about the
guards around it as much as the deploy itself:

* a member who is not an org admin cannot deploy;
* the org slug has to be typed back;
* a changed host key is a hard failure;
* an SSRF target is refused before a connection is attempted;
* a secret never reaches a log, an event, a response or a database column;
* and the mechanics — release directories, the ``current`` symlink, keeping five
  releases, rollback — behave the way the design says they do.

The SSH and FTP transports are faked, so the suite needs no server. The deploy
*logic* is real.
"""

from __future__ import annotations

import os
import tarfile

import pytest
import pytest_asyncio
from httpx import AsyncClient

from app.application.ports.deploy_target import DeployCheck, DeployTargetSpec
from app.application.services.deploy_service import DeployService
from app.core.crypto import encrypt
from app.domain.delivery.models import DeployStatus
from app.domain.identity.models import PlatformRole
from app.infrastructure.adapters.deploy_ftp import FtpDeployer, FtpEntry
from app.infrastructure.adapters.deploy_ssh import (
    CURRENT_LINK,
    KEEP_RELEASES,
    CommandOutcome,
    SshDeployer,
)
from app.infrastructure.repositories.delivery import (
    ProductionDeployRepository,
    ReleaseRepository,
)
from app.infrastructure.repositories.identity import OrganizationRepository
from app.infrastructure.repositories.projects import (
    DeployTargetRepository,
    ProjectRepository,
)
from tests.conftest import auth_header, signup

SECRET = "ssh-password-that-must-never-leak"


# ---------------------------------------------------------------------------
# fake transports
# ---------------------------------------------------------------------------


class FakeSshTransport:
    """An in-memory SSH server: a filesystem plus a scripted command table."""

    def __init__(self, *, fingerprint: str = "SHA256:abc123", fail_on: str = "") -> None:
        self.commands: list[str] = []
        self.uploads: list[tuple[str, str]] = []
        self.fingerprint = fingerprint
        self.fail_on = fail_on
        self.closed = False
        #: The remote filesystem, as a set of paths.
        self.files: set[str] = set()

    async def connect(self, **kwargs) -> None:  # noqa: ANN003
        self.connect_kwargs = kwargs
        if self.fail_on == "connect":
            raise OSError("permission denied")

    def host_key_fingerprint(self) -> str:
        return self.fingerprint

    async def run(self, command: str, *, timeout: float = 600.0) -> CommandOutcome:
        self.commands.append(command)
        if self.fail_on and self.fail_on in command:
            return CommandOutcome(exit_code=1, output="failed on purpose")

        if "ls -1t" in command and "releases" in command:
            releases = sorted(
                {p.split("/releases/")[1].split("/")[0] for p in self.files if "/releases/" in p},
                reverse=True,
            )
            return CommandOutcome(exit_code=0, output="\n".join(releases))
        if "test -w" in command:
            return CommandOutcome(exit_code=0, output="")
        if "ln -sfn" in command:
            # Record the switch so a test can assert the pointer moved.
            for part in command.split():
                if part.startswith("releases/"):
                    self.files.add(f"/srv/app/{CURRENT_LINK}->{part}")
            return CommandOutcome(exit_code=0, output="")
        if "tail -n" in command:
            self.files = {
                p for p in self.files if not p.endswith(tuple(
                    f"release-{i}" for i in range(1, 20)
                ))
            }
            return CommandOutcome(exit_code=0, output="")
        return CommandOutcome(exit_code=0, output="")

    async def upload(self, *, local_path: str, remote_path: str) -> None:
        self.uploads.append((local_path, remote_path))
        self.files.add(remote_path)

    async def close(self) -> None:
        self.closed = True


class FakeFtpTransport:
    def __init__(self, *, fail_on: str = "") -> None:
        self.commands: list[str] = []
        self.uploaded: dict[str, bytes] = {}
        self.removed: list[str] = []
        self.fail_on = fail_on
        self.cwd_path = ""
        self.connected = False

    async def connect(self, **kwargs) -> None:  # noqa: ANN003
        self.connect_kwargs = kwargs
        if self.fail_on == "connect":
            raise OSError("530 login incorrect")
        self.connected = True

    async def cwd(self, path: str) -> None:
        if self.fail_on == "cwd":
            raise OSError("550 no such directory")
        self.cwd_path = path

    async def download(self, remote_path: str) -> bytes:
        return self.uploaded.get(remote_path, b"old-content")

    async def upload(self, remote_path: str, data: bytes) -> None:
        if self.fail_on == "upload":
            raise OSError("552 quota exceeded")
        self.uploaded[remote_path] = data

    async def remove(self, remote_path: str) -> None:
        self.removed.append(remote_path)
        self.uploaded.pop(remote_path, None)

    async def list_files(self) -> list[FtpEntry]:
        return [FtpEntry(path=name, size=len(data)) for name, data in self.uploaded.items()]

    async def close(self) -> None:
        self.connected = False


class FakeDeployTargetTester:
    """Dispatches to injected deployers, like the real one but with no network."""

    def __init__(self, *, ssh: SshDeployer, ftp: FtpDeployer) -> None:
        self.ssh = ssh
        self.ftp = ftp
        self.calls: list[DeployTargetSpec] = []

    async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck:
        self.calls.append(target)
        if target.kind in ("ssh", "sftp"):
            return await self.ssh.test_connection(target=target)
        return await self.ftp.test_connection(target=target)

    def deployer_for(self, kind: str):
        return self.ssh if kind in ("ssh", "sftp") else self.ftp


# ---------------------------------------------------------------------------
# fixtures
# ---------------------------------------------------------------------------


@pytest_asyncio.fixture(autouse=True)
async def _platform_seed(seeded):
    return seeded


def _artifact(tmp_path, *, build_dir: str = "") -> str:
    """A release archive shaped like the pipeline produces."""
    source = tmp_path / "artifact-src"
    target = source / build_dir if build_dir else source
    target.mkdir(parents=True, exist_ok=True)
    (target / "index.html").write_text("<h1>hello</h1>")
    (target / "app.js").write_text("console.log('hi')")
    if build_dir:
        (source / "README.md").write_text("not deployed")

    archive = tmp_path / "release.tar.gz"
    with tarfile.open(archive, "w:gz") as tar:
        # Add the children rather than the directory itself: a bare "." member
        # is the archive root, which some extractors try to mkdir.
        for child in sorted(source.iterdir()):
            tar.add(child, arcname=child.name)
    return str(archive)


@pytest_asyncio.fixture
async def tenant(client: AsyncClient, session_factory, tmp_path):
    """An org + app + live release + a configured SSH deploy target."""
    token = (
        await signup(
            client,
            "deploy-owner@example.com",
            name="مدیر استقرار",
            platform_role=PlatformRole.OWNER,
        )
    )["access_token"]
    headers = auth_header(token)
    org = (
        await client.post("/api/organizations", json={"name": "شرکت استقرار"}, headers=headers)
    ).json()
    org_id, slug = org["id"], org["slug"]
    project_id = (
        await client.post(
            f"/api/organizations/{org_id}/projects",
            json={"name": "اپ استقرار"},
            headers=headers,
        )
    ).json()["id"]

    artifact = _artifact(tmp_path, build_dir="dist")

    async with session_factory() as session:
        from app.domain.delivery.models import PipelineRun, Release
        from app.infrastructure.repositories.delivery import PipelineRunRepository

        # A release points at the pipeline run that produced it, so the run has
        # to exist: the foreign key is real, not decorative.
        run = await PipelineRunRepository(session).queue(
            PipelineRun(
                organization_id=org_id,
                project_id=project_id,
                sha="abc1234def",
                status="succeeded",
            )
        )
        release = await ReleaseRepository(session).create(
            Release(
                organization_id=org_id,
                project_id=project_id,
                run_id=run.id,
                sha="abc1234def",
                artifact_path=artifact,
                is_demo_live=True,
            )
        )
        await DeployTargetRepository(session).upsert(
            values={
                "organization_id": org_id,
                "project_id": project_id,
                "kind": "ssh",
                "host": "deploy.example.com",
                "port": 22,
                "username": "deployer",
                "auth_type": "password",
                "secret_encrypted": encrypt(SECRET),
                "remote_path": "/srv/app",
                "post_deploy_command": "systemctl reload app",
                "is_active": True,
            }
        )
        await session.commit()

    return {
        "org_id": org_id,
        "slug": slug,
        "project_id": project_id,
        "headers": headers,
        "release_id": release.id,
        "artifact": artifact,
    }


def _service(session, *, ssh: FakeSshTransport, ftp: FakeFtpTransport | None = None) -> tuple[
    DeployService, FakeDeployTargetTester
]:
    ssh_deployer = SshDeployer(
        transport_factory=lambda: ssh, artifacts_root="/tmp/artifacts", allow_private_hosts=False
    )
    ftp_deployer = FtpDeployer(
        transport_factory=lambda: ftp or FakeFtpTransport(),
        artifacts_root="/tmp/artifacts",
        allow_private_hosts=False,
    )
    tester = FakeDeployTargetTester(ssh=ssh_deployer, ftp=ftp_deployer)
    service = DeployService(
        projects=ProjectRepository(session),
        organizations=OrganizationRepository(session),
        targets=DeployTargetRepository(session),
        deploys=ProductionDeployRepository(session),
        releases=ReleaseRepository(session),
        runtimes=None,  # not needed for the deploy paths under test
        tester=tester,
        allow_private_hosts=False,
    )
    return service, tester


# ---------------------------------------------------------------------------
# the SSH deploy mechanics
# ---------------------------------------------------------------------------


async def test_ssh_deploy_uploads_extracts_and_switches(tmp_path):
    transport = FakeSshTransport()
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)

    ok, release_dir, message = await deployer.deploy(
        target=DeployTargetSpec(
            kind="ssh", host="deploy.example.com", port=22, username="deployer",
            secret="pw", remote_path="/srv/app",
        ),
        artifact_path=_artifact(tmp_path, build_dir="dist"),
        sha="abc1234def",
        post_deploy_command="systemctl reload app",
    )

    assert ok is True, message
    assert release_dir.startswith("/srv/app/releases/")
    assert transport.uploads, "the artifact must be uploaded"
    joined = "\n".join(transport.commands)
    assert "tar -xzf" in joined, "the artifact must be extracted on the target"
    assert "systemctl reload app" in joined, "the post-deploy command must run"
    assert "ln -sfn" in joined and CURRENT_LINK in joined, "the symlink must be switched"
    assert "tail -n" in joined, "old releases must be pruned"


async def test_ssh_deploy_prunes_to_five_releases(tmp_path):
    transport = FakeSshTransport()
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
    await deployer.deploy(
        target=DeployTargetSpec(
            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
        ),
        artifact_path=_artifact(tmp_path),
        sha="abc1234",
    )
    prune = next(c for c in transport.commands if "tail -n" in c)
    assert f"tail -n +{KEEP_RELEASES + 1}" in prune, (
        "keep the newest five and delete the rest"
    )


async def test_a_failed_post_deploy_command_does_not_switch_the_symlink(tmp_path):
    """The previous release must keep serving after a failed deploy."""
    transport = FakeSshTransport(fail_on="systemctl")
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)

    ok, _release, message = await deployer.deploy(
        target=DeployTargetSpec(
            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
        ),
        artifact_path=_artifact(tmp_path),
        sha="abc1234",
        post_deploy_command="systemctl reload app",
    )

    assert ok is False
    assert "پس از استقرار" in message
    assert not any("ln -sfn" in c for c in transport.commands), (
        "the switch must not happen when the hook failed"
    )


async def test_ssh_rollback_points_current_at_the_previous_release(tmp_path):
    transport = FakeSshTransport()
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
    # Two releases exist on the target, newest first.
    transport.files = {
        "/srv/app/releases/20260101T000000-aaaaaaa/index.html",
        "/srv/app/releases/20260102T000000-bbbbbbb/index.html",
    }

    ok, target_name, message = await deployer.rollback(
        target=DeployTargetSpec(
            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
        ),
        post_deploy_command="systemctl reload app",
    )

    assert ok is True, message
    assert target_name == "20260101T000000-aaaaaaa"
    assert any("ln -sfn" in c for c in transport.commands)


async def test_ssh_rollback_refuses_when_there_is_only_one_release(tmp_path):
    transport = FakeSshTransport()
    transport.files = {"/srv/app/releases/only-one/index.html"}
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)

    ok, _target, message = await deployer.rollback(
        target=DeployTargetSpec(
            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
        )
    )
    assert ok is False
    assert "قبلی" in message


# ---------------------------------------------------------------------------
# host key pinning (TOFU)
# ---------------------------------------------------------------------------


async def test_a_changed_host_key_is_a_hard_failure():
    transport = FakeSshTransport(fingerprint="SHA256:CHANGED")
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)

    check = await deployer.test_connection(
        target=DeployTargetSpec(
            kind="ssh", host="h", port=22, username="u", secret="pw",
            remote_path="/srv/app", known_fingerprint="SHA256:ORIGINAL",
        )
    )

    assert check.ok is False
    assert "کلید میزبان" in check.message
    assert check.fingerprint == "SHA256:CHANGED"


async def test_the_fingerprint_is_returned_for_pinning_on_first_contact():
    transport = FakeSshTransport(fingerprint="SHA256:FIRST")
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)

    check = await deployer.test_connection(
        target=DeployTargetSpec(
            kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app"
        )
    )
    assert check.ok is True
    assert check.fingerprint == "SHA256:FIRST"


async def test_a_matching_pinned_key_is_accepted():
    transport = FakeSshTransport(fingerprint="SHA256:SAME")
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True)
    check = await deployer.test_connection(
        target=DeployTargetSpec(
            kind="ssh", host="h", port=22, username="u", secret="pw",
            remote_path="/srv/app", known_fingerprint="SHA256:SAME",
        )
    )
    assert check.ok is True


# ---------------------------------------------------------------------------
# SSRF
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "host",
    ["127.0.0.1", "169.254.169.254", "10.0.0.5", "localhost"],
)
async def test_an_internal_target_is_refused_before_connecting(host: str):
    transport = FakeSshTransport()
    deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=False)

    check = await deployer.test_connection(
        target=DeployTargetSpec(
            kind="ssh", host=host, port=22, username="u", secret="pw", remote_path="/srv/app"
        )
    )

    assert check.ok is False
    assert transport.commands == [], "no connection may be attempted"
    assert not transport.uploads


async def test_an_internal_ftp_target_is_refused():
    transport = FakeFtpTransport()
    deployer = FtpDeployer(transport_factory=lambda: transport, allow_private_hosts=False)
    check = await deployer.test_connection(
        target=DeployTargetSpec(
            kind="ftp", host="127.0.0.1", port=21, username="u", secret="pw",
            remote_path="/var/www",
        )
    )
    assert check.ok is False
    assert transport.connected is False


# ---------------------------------------------------------------------------
# FTP
# ---------------------------------------------------------------------------


async def test_ftp_deploy_backs_up_before_overwriting(tmp_path):
    transport = FakeFtpTransport()
    transport.uploaded = {"index.html": b"OLD", "app.js": b"OLD"}
    deployer = FtpDeployer(
        transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts")
    )

    ok, backup, message = await deployer.deploy(
        target=DeployTargetSpec(
            kind="ftp", host="ftp.example.com", port=21, username="u", secret="pw",
            remote_path="/var/www",
        ),
        artifact_path=_artifact(tmp_path, build_dir="dist"),
        sha="abc1234",
        organization_id="org-1",
        project_id="proj-1",
    )

    assert ok is True, message
    assert backup and os.path.isfile(backup), "a backup must exist before anything is overwritten"
    with tarfile.open(backup, "r:gz") as tar:
        names = set(tar.getnames())
    assert {"index.html", "app.js"} <= names, "the backup must contain what it overwrites"


async def test_ftp_deploy_uploads_only_the_build_output(tmp_path):
    """A static deploy must not publish the repository, only the build."""
    transport = FakeFtpTransport()
    deployer = FtpDeployer(
        transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts")
    )
    await deployer.deploy(
        target=DeployTargetSpec(
            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
        ),
        artifact_path=_artifact(tmp_path, build_dir="dist"),
        sha="abc1234",
    )

    assert "index.html" in transport.uploaded
    assert "README.md" not in transport.uploaded, "the source tree must not be published"


async def test_ftp_rollback_re_uploads_the_backup(tmp_path):
    transport = FakeFtpTransport()
    transport.uploaded = {"index.html": b"OLD"}
    deployer = FtpDeployer(
        transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts")
    )
    _ok, backup, _message = await deployer.deploy(
        target=DeployTargetSpec(
            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
        ),
        artifact_path=_artifact(tmp_path, build_dir="dist"),
        sha="abc1234",
        organization_id="org-1",
        project_id="proj-1",
    )

    transport.uploaded = {"index.html": b"BROKEN"}
    ok, message = await deployer.rollback(
        target=DeployTargetSpec(
            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
        ),
        backup_path=backup,
    )

    assert ok is True, message
    assert transport.uploaded["index.html"] == b"OLD"


async def test_ftp_rollback_without_a_backup_fails_cleanly():
    transport = FakeFtpTransport()
    deployer = FtpDeployer(transport_factory=lambda: transport)
    ok, message = await deployer.rollback(
        target=DeployTargetSpec(
            kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www"
        ),
        backup_path="",
    )
    assert ok is False
    assert message


# ---------------------------------------------------------------------------
# the service guards
# ---------------------------------------------------------------------------


async def test_the_confirm_slug_must_match(tenant, session_factory):
    transport = FakeSshTransport()
    async with session_factory() as session:
        service, _tester = _service(session, ssh=transport)
        with pytest.raises(Exception) as excinfo:
            await service.deploy_production(
                organization_id=tenant["org_id"],
                project_id=tenant["project_id"],
                actor_user_id="u1",
                confirm="not-the-slug",
            )
        assert "تأییدیه" in str(excinfo.value)
    assert transport.commands == [], "nothing may be attempted on a bad confirmation"


async def test_a_correct_confirmation_deploys_and_records_the_row(tenant, session_factory):
    transport = FakeSshTransport()
    async with session_factory() as session:
        service, _tester = _service(session, ssh=transport)
        deploy = await service.deploy_production(
            organization_id=tenant["org_id"],
            project_id=tenant["project_id"],
            actor_user_id="u1",
            confirm=tenant["slug"],
        )
        await session.commit()

    assert deploy.status == DeployStatus.SUCCEEDED.value, deploy.message
    assert deploy.kind == "ssh"
    async with session_factory() as session:
        rows = await ProductionDeployRepository(session).list_for_project(
            project_id=tenant["project_id"]
        )
        assert len(rows) == 1
        assert rows[0].release_id == tenant["release_id"]


async def test_a_second_deploy_is_refused_while_one_is_running(tenant, session_factory):
    from app.domain.delivery.models import ProductionDeploy

    transport = FakeSshTransport()
    async with session_factory() as session:
        repo = ProductionDeployRepository(session)
        await repo.create(
            ProductionDeploy(
                organization_id=tenant["org_id"],
                project_id=tenant["project_id"],
                kind="ssh",
                host="h",
                status=DeployStatus.RUNNING.value,
            )
        )
        await session.commit()

        service, _tester = _service(session, ssh=transport)
        with pytest.raises(Exception) as excinfo:
            await service.deploy_production(
                organization_id=tenant["org_id"],
                project_id=tenant["project_id"],
                actor_user_id="u1",
                confirm=tenant["slug"],
            )
        assert "در حال انجام" in str(excinfo.value)


async def test_a_secret_never_reaches_the_log_or_the_row(tenant, session_factory):
    """The one thing that must never happen: a credential in a stored field."""
    transport = FakeSshTransport()
    async with session_factory() as session:
        service, _tester = _service(session, ssh=transport)
        deploy = await service.deploy_production(
            organization_id=tenant["org_id"],
            project_id=tenant["project_id"],
            actor_user_id="u1",
            confirm=tenant["slug"],
        )
        await session.commit()

    assert SECRET not in deploy.log_tail
    assert SECRET not in deploy.message
    for _local, remote in transport.uploads:
        assert SECRET not in remote
    for command in transport.commands:
        assert SECRET not in command, "a credential must never reach a remote command line"


async def test_the_ftp_rollback_without_a_backup_is_refused_not_guessed(
    tenant, session_factory
):
    """FTP has no atomic switch: without the backup there is nothing to undo."""
    async with session_factory() as session:
        target = await DeployTargetRepository(session).get_for_project(tenant["project_id"])
        assert target is not None
        target.kind = "ftp"
        await session.commit()

    transport = FakeFtpTransport()
    async with session_factory() as session:
        service, _tester = _service(session, ssh=FakeSshTransport(), ftp=transport)
        deploy = await service.deploy_production(
            organization_id=tenant["org_id"],
            project_id=tenant["project_id"],
            actor_user_id="u1",
            confirm=tenant["slug"],
        )
        await session.commit()
        assert deploy.status == DeployStatus.SUCCEEDED.value, deploy.message

        # Clear the recorded backup to simulate a lost archive.
        row = await ProductionDeployRepository(session).get(deploy.id)
        assert row is not None
        deploy_row = await session.get(
            __import__(
                "app.infrastructure.db.models_delivery", fromlist=["ProductionDeployRow"]
            ).ProductionDeployRow,
            deploy.id,
        )
        assert deploy_row is not None
        deploy_row.backup_path = ""
        await session.commit()

        rolled = await service.rollback_production(
            organization_id=tenant["org_id"],
            project_id=tenant["project_id"],
            actor_user_id="u1",
            confirm=tenant["slug"],
            deploy_id=deploy.id,
        )
        await session.commit()

    assert rolled.status == DeployStatus.FAILED.value
    assert "پشتیبان" in rolled.message


# ---------------------------------------------------------------------------
# HTTP surface
# ---------------------------------------------------------------------------


async def test_a_member_cannot_deploy_to_production(client: AsyncClient, tenant):
    member_token = (await signup(client, "deploy-member@example.com"))["access_token"]
    member_headers = auth_header(member_token)
    # Join the org as a viewer so membership succeeds but the capability does not.
    invite = await client.post(
        f"/api/organizations/{tenant['org_id']}/invites",
        json={"email": "deploy-member@example.com", "role": "viewer"},
        headers=tenant["headers"],
    )
    if invite.status_code in (200, 201):
        token = invite.json().get("token") or invite.json().get("invite_url", "").split("=")[-1]
        if token:
            await client.post(
                "/api/invites/accept",
                json={"token": token, "password": "S3cure-passw0rd!"},
                headers=member_headers,
            )

    response = await client.post(
        f"/api/organizations/{tenant['org_id']}/delivery/production/deploy",
        json={
            "project_id": tenant["project_id"],
            "confirm": tenant["slug"],
        },
        headers=member_headers,
    )
    # Either not a member (404, the tenancy check) or a member without the
    # capability (403). Never a success.
    assert response.status_code in (403, 404), response.text


async def test_the_confirm_slug_is_enforced_over_http(client: AsyncClient, tenant):
    response = await client.post(
        f"/api/organizations/{tenant['org_id']}/delivery/production/deploy",
        json={"project_id": tenant["project_id"], "confirm": "wrong-slug"},
        headers=tenant["headers"],
    )
    assert response.status_code == 422, response.text
    assert "تأییدیه" in response.json()["error"]["message"]


async def test_the_target_summary_never_returns_the_secret(client: AsyncClient, tenant):
    response = await client.get(
        f"/api/organizations/{tenant['org_id']}/delivery/production",
        params={"project_id": tenant["project_id"]},
        headers=tenant["headers"],
    )

    assert response.status_code == 200, response.text
    body = response.json()
    assert body["target"]["has_secret"] is True
    assert SECRET not in response.text
    assert "secret" not in set(body["target"]), "the raw secret column must not be exposed"


async def test_deploying_without_a_target_is_refused(client: AsyncClient, tenant, session_factory):
    async with session_factory() as session:
        target = await DeployTargetRepository(session).get_for_project(tenant["project_id"])
        assert target is not None
        target.is_active = False
        await session.commit()

    response = await client.post(
        f"/api/organizations/{tenant['org_id']}/delivery/production/deploy",
        json={"project_id": tenant["project_id"], "confirm": tenant["slug"]},
        headers=tenant["headers"],
    )
    assert response.status_code == 422, response.text
    assert "مقصد" in response.json()["error"]["message"]