Download translated resources directly

Author: eemeli

pontoon/base/tests/views/test_download.py

@@ -1,88 +1,143 @@
-from os import makedirs
-from tempfile import TemporaryDirectory
-from unittest.mock import patch
+from io import BytesIO
+from textwrap import dedent
+from zipfile import ZipFile
 
 import pytest
 
-from django.conf import settings
 from django.test import RequestFactory
 
-from pontoon.base.models import Project, Resource
+from pontoon.base.models import Project
 from pontoon.base.views import download_translations
-from pontoon.sync.tests.test_checkouts import MockVersionControl
-from pontoon.sync.tests.utils import build_file_tree
 from pontoon.test.factories import (
+    EntityFactory,
     LocaleFactory,
     ProjectFactory,
-    RepositoryFactory,
     ResourceFactory,
+    SectionFactory,
     TranslatedResourceFactory,
+    TranslationFactory,
     UserFactory,
 )
 
 
 @pytest.mark.django_db
-@pytest.mark.parametrize("two_repos", [True, False])
-@pytest.mark.parametrize(
-    "repo_url,expected_location",
-    [
-        (
-            "https://github.com:gh-org/gh-repo.git",
-            "https://raw.githubusercontent.com/gh-org/gh-repo/HEAD/de-Test/a.ftl",
-        ),
-        (
-            "git@gitlab.com:gl-org/gl-repo.git",
-            "https://gitlab.com/gl-org/gl-repo/-/raw/HEAD/de-Test/a.ftl?inline=false",
-        ),
-        ("http://example.com/tgt-repo", "https://example.com/tgt-repo"),
-    ],
-)
-def test_download(two_repos, repo_url, expected_location):
-    with (
-        TemporaryDirectory() as root,
-        patch("pontoon.sync.core.checkout.get_repo", return_value=MockVersionControl()),
-    ):
-        settings.MEDIA_ROOT = root
-        locale = LocaleFactory.create(code="de-Test")
-        if two_repos:
-            repo_src = RepositoryFactory(
-                url="http://example.com/src-repo", source_repo=True
-            )
-            repo_tgt = RepositoryFactory(url=repo_url)
-            project = ProjectFactory.create(
-                name="test-dl",
-                locales=[locale],
-                repositories=[repo_src, repo_tgt],
-                visibility=Project.Visibility.PUBLIC,
-            )
-            src_root = repo_src.checkout_path
-            tgt_root = repo_tgt.checkout_path
-            makedirs(src_root)
-            build_file_tree(src_root, {"en-US": {"a.ftl": ""}})
-            makedirs(tgt_root)
-            build_file_tree(tgt_root, {"de-Test": {"a.ftl": ""}})
-        else:
-            repo = RepositoryFactory(url=repo_url)
-            project = ProjectFactory.create(
-                name="test-dl",
-                locales=[locale],
-                repositories=[repo],
-                visibility=Project.Visibility.PUBLIC,
-            )
-            repo_root = repo.checkout_path
-            makedirs(repo_root)
-            build_file_tree(
-                repo_root, {"en-US": {"a.ftl": ""}, "de-Test": {"a.ftl": ""}}
-            )
-        res = ResourceFactory.create(
-            project=project, path="a.ftl", format=Resource.Format.FLUENT
-        )
-        TranslatedResourceFactory.create(locale=locale, resource=res)
+def test_download_fluent():
+    locale = LocaleFactory.create(code="de-Test")
+    project = ProjectFactory.create(
+        name="test-dl",
+        locales=[locale],
+        visibility=Project.Visibility.PUBLIC,
+    )
+    res = ResourceFactory.create(
+        project=project, format="fluent", path="path/to/file.ftl"
+    )
+    TranslatedResourceFactory.create(locale=locale, resource=res)
+    section = SectionFactory.create(resource=res, key=[], comment="Group")
+    e1 = EntityFactory.create(resource=res, section=section, key=["e1"], value=["E1"])
+    e2 = EntityFactory.create(
+        resource=res,
+        section=section,
+        key=["e2"],
+        value=[],
+        properties={"attr": ["E2"]},
+    )
+    EntityFactory.create(resource=res, section=section, key=["e3"], value=["E3"])
+    TranslationFactory.create(
+        locale=locale, entity=e1, value=["T1"], active=True, approved=True
+    )
+    TranslationFactory.create(
+        locale=locale,
+        entity=e2,
+        value=[],
+        properties={"attr": ["T2"]},
+        active=True,
+        approved=True,
+    )
+
+    request = RequestFactory().get(
+        "/translations/?code=de-Test&slug=test-dl&part=path/to/file.ftl"
+    )
+    request.user = UserFactory()
+    response = download_translations(request)
+    assert response.status_code == 200
+    assert response["Content-Type"] == "application/zip"
+    assert (
+        response["Content-Disposition"]
+        == "attachment; filename=de-Test_test-dl_path_to_file.zip"
+    )
+    bytes_io = BytesIO(response.content)
+    with ZipFile(bytes_io, "r") as zipfile:
+        assert zipfile.namelist() == ["de-Test_test-dl_path_to_file.ftl"]
+        raw = zipfile.read("de-Test_test-dl_path_to_file.ftl")
+    assert raw.decode("utf-8") == dedent("""\
+        ## Group
+
+        e1 = T1
+        e2 =
+            .attr = T2
+        """)
+
+
+@pytest.mark.django_db
+def test_download_xliff():
+    locale = LocaleFactory.create(code="de-Test")
+    project = ProjectFactory.create(
+        name="test-dlx",
+        locales=[locale],
+        visibility=Project.Visibility.PUBLIC,
+    )
+    res = ResourceFactory.create(project=project, format="xliff", path="file.xlf")
+    TranslatedResourceFactory.create(locale=locale, resource=res)
+    section = SectionFactory.create(resource=res, key=["file.foo"], comment="Group")
+    e1 = EntityFactory.create(
+        resource=res, section=section, key=["file.foo", "e1"], value=["E1"]
+    )
+    e2 = EntityFactory.create(
+        resource=res, section=section, key=["file.foo", "e2"], value=["E2"]
+    )
+    EntityFactory.create(
+        resource=res, section=section, key=["file.foo", "e3"], value=["E3"]
+    )
+    TranslationFactory.create(
+        locale=locale, entity=e1, value=["T1"], active=True, approved=True
+    )
+    TranslationFactory.create(
+        locale=locale, entity=e2, value=["T2"], active=True, approved=True
+    )
 
-        request = RequestFactory().get(
-            "/translations/?code=de-Test&slug=test-dl&part=a.ftl"
-        )
-        request.user = UserFactory()
-        response = download_translations(request)
-        assert response.status_code == 302
-        assert response.get("Location") == expected_location
+    request = RequestFactory().get(
+        "/translations/?code=de-Test&slug=test-dlx&part=file.xlf"
+    )
+    request.user = UserFactory()
+    response = download_translations(request)
+    assert response.status_code == 200
+    assert response["Content-Type"] == "application/zip"
+    assert (
+        response["Content-Disposition"]
+        == "attachment; filename=de-Test_test-dlx_file.zip"
+    )
+    bytes_io = BytesIO(response.content)
+    with ZipFile(bytes_io, "r") as zipfile:
+        assert zipfile.namelist() == ["de-Test_test-dlx_file.xlf"]
+        raw = zipfile.read("de-Test_test-dlx_file.xlf")
+    assert raw.decode("utf-8") == dedent("""\
+        <?xml version="1.0" encoding="utf-8"?>
+        <xliff>
+          <file original="file.foo">
+            <!-- Group -->
+            <body>
+              <trans-unit id="e1">
+                <source>E1</source>
+                <target>T1</target>
+              </trans-unit>
+              <trans-unit id="e2">
+                <source>E2</source>
+                <target>T2</target>
+              </trans-unit>
+              <trans-unit id="e3">
+                <source>E3</source>
+              </trans-unit>
+            </body>
+          </file>
+        </xliff>
+        """)

pontoon/base/views.py

@@ -20,7 +20,6 @@
     HttpRequest,
     HttpResponse,
     HttpResponseForbidden,
-    HttpResponseRedirect,
     JsonResponse,
     StreamingHttpResponse,
 )
@@ -1008,9 +1007,12 @@ def perform_checks(request):
 
 @transaction.atomic
 def download_translations(request):
-    """Download translated resource from its backing repository."""
+    """Download translated resource."""
 
-    from pontoon.sync.utils import translations_target_url
+    from io import BytesIO
+    from zipfile import ZIP_DEFLATED, ZipFile
+
+    from pontoon.sync.utils import serialize_translated_resource
 
     try:
         slug = request.GET["slug"]
@@ -1020,16 +1022,21 @@ def download_translations(request):
         raise Http404
 
     project = get_object_or_404(Project.objects.visible_for(request.user), slug=slug)
+    resource = get_object_or_404(Resource, project=project, path=res_path)
     locale = get_object_or_404(Locale, code=code)
 
-    # FIXME This is a temporary hack, to be replaced by 04/2025 with proper downloads.
-    # Once fixed, we should remove SSH credentials from the web pod
-    # https://github.com/mozilla/webservices-infra/pull/9295
-    url = translations_target_url(project, locale, res_path)
-    if url and url.startswith("https://"):
-        return HttpResponseRedirect(url)
-    else:
-        raise Http404
+    filename = re.sub(r"[/\\]", "_", f"{locale.code}_{slug}_{res_path}")
+    bytes_io = BytesIO()
+    zipfile = ZipFile(bytes_io, "w", compression=ZIP_DEFLATED)
+    zipfile.writestr(filename, serialize_translated_resource(resource, locale))
+    zipfile.close()
+
+    response = HttpResponse()
+    response.content = bytes_io.getvalue()
+    response["Content-Type"] = "application/zip"
+    zip_name = re.sub(r"[^.]+$", "zip", filename)
+    response["Content-Disposition"] = f"attachment; filename={zip_name}"
+    return response
 
 
 @login_required(redirect_field_name="", login_url="/403")

pontoon/sync/utils.py

@@ -1,82 +1,53 @@
-import re
-
-from io import BytesIO
-from os.path import basename, exists, join, relpath
+from os.path import basename, join
 from tempfile import TemporaryDirectory
-from zipfile import ZipFile
+
+from moz.l10n.resource import serialize_resource
 
 from django.core.files import File
+from django.db.models import Q
 from django.utils import timezone
 
 from pontoon.base.badge_utils import badges_review_level, badges_translation_level
-from pontoon.base.models import ChangedEntityLocale, Locale, Project, Repository, User
+from pontoon.base.models import (
+    ChangedEntityLocale,
+    Locale,
+    Project,
+    Resource as DbResource,
+    Translation,
+    User,
+)
 from pontoon.messaging.notifications import send_badge_notification
-from pontoon.sync.core.checkout import checkout_repos
-from pontoon.sync.core.paths import UploadPaths, find_paths
+from pontoon.sync.core.paths import UploadPaths
 from pontoon.sync.core.stats import update_stats
 from pontoon.sync.core.translations_from_repo import find_db_updates, write_db_updates
-from pontoon.sync.core.translations_to_repo import update_changed_resources
-
-
-# FIXME This is a temporary hack, to be replaced by 04/2025 with proper downloads.
-# Once fixed, we should remove SSH credentials from the web pod
-# https://github.com/mozilla/webservices-infra/pull/9295
-def translations_target_url(
-    project: Project, locale: Locale, resource_path: str
-) -> str | None:
-    """The target repository URL for a resource, for direct download."""
-
-    if project.repositories.count() > 1:
-        # HACK: Let's assume that no config is used, and the target repo root is the right base.
-        target_repo: Repository = project.repositories.get(source_repo=False)
-        rel_path = f"{locale.code}/{resource_path}"
-    else:
-        checkouts = checkout_repos(project, shallow=True)
-        target_repo = checkouts.target.repo
-        paths = find_paths(project, checkouts)
-        target, _ = paths.target(resource_path)
-        if not target:
-            return None
-        abs_path = paths.format_target_path(target, locale.code)
-        rel_path = relpath(abs_path, checkouts.target.path).replace("\\", "/")
-
-    github = re.search(r"\bgithub\.com[:/]([^/]+)/([^/]+)\.git$", target_repo.url)
-    if github:
-        org, repo = github.groups()
-        ref = f"refs/heads/{target_repo.branch}" if target_repo.branch else "HEAD"
-        return f"https://raw.githubusercontent.com/{org}/{repo}/{ref}/{rel_path}"
-
-    gitlab = re.search(r"gitlab\.com[:/]([^/]+)/([^/]+)\.git$", target_repo.url)
-    if gitlab:
-        org, repo = gitlab.groups()
-        ref = target_repo.branch or "HEAD"
-        return f"https://gitlab.com/{org}/{repo}/-/raw/{ref}/{rel_path}?inline=false"
-
-    # Default to bare repo link
-    return re.sub(r"^.*?(://|@)", "https://", target_repo.url, count=1)
-
-
-# FIXME Currently not in use, to be refactored for proper download support
-def download_translations_zip(
-    project: Project, locale: Locale
-) -> tuple[bytes, str] | tuple[None, None]:
-    checkouts = checkout_repos(project, shallow=True)
-    paths = find_paths(project, checkouts)
-    db_changes = ChangedEntityLocale.objects.filter(
-        entity__resource__project=project, locale=locale
-    ).select_related("entity__resource", "locale")
-    update_changed_resources(project, paths, {}, [], db_changes, set(), timezone.now())
-
-    bytes_io = BytesIO()
-    zipfile = ZipFile(bytes_io, "w")
-    for _, tgt_path in paths.all():
-        filename = paths.format_target_path(tgt_path, locale.code)
-        if exists(filename):
-            arcname = relpath(filename, checkouts.target.path)
-            zipfile.write(filename, arcname)
-    zipfile.close()
+from pontoon.sync.core.translations_to_repo import (
+    build_moz_l10n_resource,
+    build_translated_resource,
+)
+
+
+def serialize_translated_resource(db_res: DbResource, locale: Locale) -> str:
+    res = build_moz_l10n_resource(db_res)
+    translations = {
+        tuple(tx.entity.key): tx
+        for tx in Translation.objects.filter(
+            entity__obsolete=False,
+            entity__resource=db_res,
+            locale=locale,
+            active=True,
+        )
+        .filter(
+            Q(approved=True)
+            | Q(pretranslated=True, warnings__isnull=True)
+            | Q(fuzzy=True)
+        )
+        .select_related("entity")
+        .iterator()
+    }
+    tr_res = build_translated_resource(locale, translations, res)
 
-    return bytes_io.getvalue(), f"{project.slug}.zip"
+    lc_plurals = locale.cldr_plurals_list()
+    return "".join(serialize_resource(tr_res, gettext_plurals=lc_plurals))
 
 
 def import_uploaded_file(