directory: Allow to use an origin URL different from the root svn URL

When the SvnExportLoader must export multiple subsets of a subversion tree
using the same root svn URL, the same origin URL should not be used in order
to disambiguate what is archived.

So allow to specify the root svn URL in an optional svn_url parameter of the
loader while a different origin URL is provided as first positional parameter
of the loader.

Related to swh/infra/sysadm-environment#5263.

swh/loader/svn/directory.py

@@ -24,6 +24,9 @@ class SvnExportLoader(BaseDirectoryLoader):
     It is also possible to load a subset of the source tree by explicitly
     specifying the sub-paths to export in the ``svn_paths`` optional parameter.
 
+    If the origin URL should be different from the subversion URL, the latter
+    can be provided using the optional ``svn_url`` parameter.
+
     The output snapshot is of the form:
 
     .. code::
@@ -41,14 +44,23 @@ class SvnExportLoader(BaseDirectoryLoader):
 
     visit_type = "svn-export"
 
-    def __init__(self, *args, svn_paths: Optional[List[str]] = None, **kwargs):
+    def __init__(
+        self,
+        *args,
+        svn_paths: Optional[List[str]] = None,
+        svn_url: Optional[str] = None,
+        **kwargs,
+    ):
         self.svn_revision = kwargs.pop("ref")
         self.svn_paths = svn_paths
         super().__init__(*args, **kwargs)
+        self.svn_url = svn_url
+        if self.svn_url is None:
+            self.svn_url = self.origin.url
         self.svnrepo: Optional[SvnRepo] = None
 
     def prepare(self) -> None:
-        self.svnrepo = get_svn_repo(self.origin.url)
+        self.svnrepo = get_svn_repo(self.svn_url)
         super().prepare()
 
     def cleanup(self) -> None:
@@ -63,9 +75,10 @@ class SvnExportLoader(BaseDirectoryLoader):
             _, local_url = self.svnrepo.export_temporary(self.svn_revision)
             yield Path(local_url.decode())
         else:
+            assert self.svn_url is not None
             self.log.debug(
                 "Exporting from the svn source tree rooted at %s@%s the sub-paths: %s",
-                self.origin.url,
+                self.svn_url,
                 self.svn_revision,
                 ", ".join(self.svn_paths),
             )
@@ -73,7 +86,7 @@ class SvnExportLoader(BaseDirectoryLoader):
                 suffix="-" + datetime.now().isoformat()
             ) as tmp_dir:
                 for svn_path in self.svn_paths:
-                    svn_url = os.path.join(self.origin.url, svn_path.strip("/"))
+                    svn_url = os.path.join(self.svn_url, svn_path.strip("/"))
                     export_path = os.path.join(tmp_dir, svn_path.strip("/"))
                     os.makedirs("/".join(export_path.split("/")[:-1]), exist_ok=True)
                     self.svnrepo.export(

swh/loader/svn/tests/test_directory.py

@@ -5,6 +5,8 @@
 
 import os
 
+import pytest
+
 from swh.loader.core.nar import Nar
 from swh.loader.svn.directory import SvnExportLoader
 from swh.loader.svn.svn_repo import get_svn_repo
@@ -94,7 +96,14 @@ def test_loader_svn_directory(swh_storage, datadir, tmp_path):
     assert actual_result2 == {"status": "uneventful"}
 
 
-def test_loader_svn_directory_sub_paths(swh_storage, datadir, tmp_path):
+@pytest.mark.parametrize(
+    "use_custom_origin_url",
+    [False, True],
+    ids=["origin_url == svn_url", "origin_url != svn_url"],
+)
+def test_loader_svn_directory_sub_paths(
+    swh_storage, datadir, tmp_path, use_custom_origin_url
+):
     """Loading a subset of a svn tree with proper nar checksums should be eventful"""
     archive_name = "pkg-gourmet"
     archive_path = os.path.join(datadir, f"{archive_name}.tgz")
@@ -108,9 +117,17 @@ def test_loader_svn_directory_sub_paths(swh_storage, datadir, tmp_path):
         "sha256": "21e9553da2f8ae27d6b9ae87f509b0233fc6edbabc1099c31b90e1dec2cbb618"
     }
 
+    origin_url = (
+        f"{repo_url}?nar=sha256-{checksums['sha256']}"
+        if use_custom_origin_url
+        else repo_url
+    )
+    svn_url = repo_url if use_custom_origin_url else None
+
     loader = SvnExportLoader(
         swh_storage,
-        repo_url,
+        origin_url,
+        svn_url=svn_url,
         ref=svn_revision,
         svn_paths=svn_paths,
         checksum_layout=checksum_layout,
@@ -123,7 +140,7 @@ def test_loader_svn_directory_sub_paths(swh_storage, datadir, tmp_path):
 
     actual_visit = assert_last_visit_matches(
         swh_storage,
-        repo_url,
+        origin_url,
         status="full",
         type="svn-export",
     )
@@ -158,7 +175,8 @@ def test_loader_svn_directory_sub_paths(swh_storage, datadir, tmp_path):
     # Another run should be uneventful
     loader2 = SvnExportLoader(
         swh_storage,
-        repo_url,
+        origin_url,
+        svn_url=svn_url,
         ref=svn_revision,
         svn_paths=svn_paths,
         checksum_layout=checksum_layout,