Add --force-platform to conda-lock install command.

conda_lock/conda_lock.py

@@ -182,8 +182,6 @@ def extract_input_hash(lockfile_contents: str) -> Optional[str]:
 
 
 def _do_validate_platform(platform: str) -> Tuple[bool, str]:
-    from ensureconda.resolve import platform_subdir
-
     determined_subdir = platform_subdir()
     return platform == determined_subdir, determined_subdir
 
@@ -985,6 +983,7 @@ def _render_lockfile_for_install(
     filename: pathlib.Path,
     include_dev_dependencies: bool = True,
     extras: Optional[AbstractSet[str]] = None,
+    force_platform: Optional[str] = None,
 ) -> Iterator[pathlib.Path]:
     """
     Render lock content into a temporary, explicit lockfile for the current platform
@@ -1006,7 +1005,8 @@ def _render_lockfile_for_install(
 
     lock_content = parse_conda_lock_file(pathlib.Path(filename))
 
-    platform = platform_subdir()
+    platform = force_platform or platform_subdir()
+
     if platform not in lock_content.metadata.platforms:
         suggested_platforms_section = "platforms:\n- "
         suggested_platforms_section += "\n- ".join(
@@ -1506,6 +1506,11 @@ def lock(
     default=[],
     help="include extra dependencies from the lockfile (where applicable)",
 )
+@click.option(
+    "--force-platform",
+    help="Force using the given platform when installing from the lockfile, instead of the native platform.",
+    default=platform_subdir,
+)
 @click.argument("lock-file", default=DEFAULT_INSTALL_OPT_LOCK_FILE, type=click.Path())
 @click.pass_context
 def click_install(
@@ -1523,6 +1528,7 @@ def click_install(
     log_level: TLogLevel,
     dev: bool,
     extras: List[str],
+    force_platform: str,
 ) -> None:
     # bail out if we do not encounter the lockfile
     lock_file = pathlib.Path(lock_file)
@@ -1545,6 +1551,7 @@ def click_install(
         validate_platform=validate_platform,
         dev=dev,
         extras=extras,
+        force_platform=force_platform,
     )
 
 
@@ -1561,6 +1568,7 @@ def install(
     validate_platform: bool = DEFAULT_INSTALL_OPT_VALIDATE_PLATFORM,
     dev: bool = DEFAULT_INSTALL_OPT_DEV,
     extras: Optional[List[str]] = None,
+    force_platform: Optional[str] = None,
 ) -> None:
     if extras is None:
         extras = []
@@ -1580,7 +1588,10 @@ def install(
                 error.args[0] + " Disable validation with `--no-validate-platform`."
             )
     with _render_lockfile_for_install(
-        lock_file, include_dev_dependencies=dev, extras=set(extras)
+        lock_file,
+        include_dev_dependencies=dev,
+        extras=set(extras),
+        force_platform=force_platform,
     ) as lockfile:
         if _auth is not None:
             with _add_auth(read_file(lockfile), _auth) as lockfile_with_auth:

tests/test_conda_lock.py

@@ -25,6 +25,7 @@
 
 from click.testing import CliRunner
 from click.testing import Result as CliResult
+from ensureconda.resolve import platform_subdir
 from flaky import flaky
 from freezegun import freeze_time
 
@@ -2060,16 +2061,16 @@ def test_solve_arch_multiple_categories():
         assert numpy_deps[0].categories == {"test", "dev"}
 
 
-def _check_package_installed(package: str, prefix: str):
-    import glob
-
-    files = list(glob.glob(f"{prefix}/conda-meta/{package}-*.json"))
+def _check_package_installed(package: str, prefix: str, subdir: Optional[str] = None):
+    files = list(glob(f"{prefix}/conda-meta/{package}-*.json"))
     assert len(files) >= 1
     # TODO: validate that all the files are in there
     for fn in files:
         data = json.load(open(fn))
         for expected_file in data["files"]:
             assert (Path(prefix) / Path(expected_file)).exists()
+        if subdir is not None:
+            assert data["subdir"] == subdir
     return True
 
 
@@ -2503,8 +2504,6 @@ def test_virtual_packages(
 
     platform = "linux-64"
 
-    from click.testing import CliRunner
-
     for lockfile in glob(f"conda-{platform}.*"):
         os.unlink(lockfile)
 
@@ -2651,6 +2650,54 @@ def test_fake_conda_env(conda_exe: str, conda_lock_yaml: Path):
             assert platform == path.parent.name
 
 
+def test_forced_platform(
+    conda_exe: str,
+    tmp_path: Path,
+    conda_lock_yaml: Path,
+    capsys: "pytest.CaptureFixture[str]",
+):
+    if is_micromamba(conda_exe):
+        pytest.skip(reason="micromamba tests are failing")
+    if platform_subdir() != "osx-arm64":
+        pytest.skip("Test is only relevant for macOS on ARM64")
+
+    root_prefix = tmp_path / "root_prefix"
+    root_prefix.mkdir(exist_ok=True)
+    prefix = root_prefix / "test_env"
+
+    package = "bzip2"
+    platform = "osx-64"
+    assert platform != platform_subdir()
+
+    install_args = [
+        "install",
+        "--conda",
+        conda_exe,
+        "--prefix",
+        str(prefix),
+        "--force-platform",
+        platform,
+        str(conda_lock_yaml),
+    ]
+    with capsys.disabled():
+        runner = CliRunner(mix_stderr=False)
+        result = runner.invoke(main, install_args, catch_exceptions=False)
+
+    print(result.stdout, file=sys.stdout)
+    print(result.stderr, file=sys.stderr)
+    if Path(conda_lock_yaml).exists():
+        logging.debug(
+            "lockfile contents: \n\n=======\n%s\n\n==========",
+            Path(conda_lock_yaml).read_text(),
+        )
+
+    assert _check_package_installed(
+        package=package,
+        prefix=str(prefix),
+        subdir=platform,
+    ), f"Package {package} does not exist in {prefix} environment"
+
+
 @pytest.mark.parametrize("placeholder", ["$QUETZ_API_KEY", "${QUETZ_API_KEY}"])
 @flaky
 def test_private_lock(
@@ -2668,7 +2715,6 @@ def test_private_lock(
         )
         logging.info("using micromamba version %s", res.stdout)
         pytest.xfail("micromamba doesn't support our quetz server urls properly")
-    from ensureconda.resolve import platform_subdir
 
     monkeypatch.setenv("QUETZ_API_KEY", quetz_server.api_key)
     monkeypatch.chdir(tmp_path)
@@ -2683,10 +2729,8 @@ def test_private_lock(
     (tmp_path / "environment.yml").write_text(content)
 
     with capsys.disabled():
-        from click.testing import CliRunner, Result
-
         runner = CliRunner(mix_stderr=False)
-        result: Result = runner.invoke(
+        result: CliResult = runner.invoke(
             main,
             [
                 "lock",
@@ -2703,7 +2747,7 @@ def run_install():
             env_name = uuid.uuid4().hex
             env_prefix = tmp_path / env_name
 
-            result: Result = runner.invoke(
+            result: CliResult = runner.invoke(
                 main,
                 [
                     "install",
@@ -2778,10 +2822,8 @@ def test_lookup(
     monkeypatch.chdir(cwd)
     lookup_filename = str((cwd / lookup_source).absolute())
     with capsys.disabled():
-        from click.testing import CliRunner, Result
-
         runner = CliRunner(mix_stderr=False)
-        result: Result = runner.invoke(
+        result: CliResult = runner.invoke(
             main,
             ["lock", "--pypi_to_conda_lookup_file", lookup_filename],
             catch_exceptions=False,
@@ -2859,8 +2901,6 @@ def test_get_pkgs_dirs_mocked_output(info_file: str, expected: Optional[List[Pat
 def test_cli_version(capsys: "pytest.CaptureFixture[str]"):
     """It should correctly report its version."""
 
-    from click.testing import CliRunner
-
     with capsys.disabled():
         runner = CliRunner(mix_stderr=False)
         result = runner.invoke(