always resolve symlinks for TemporaryDirectory
On macOS mktemp returns a temporary directory in a symlink. Nix has a bug where it won't accept path:// located in a symlink. This avoid this issue by always resolving symlinks as returned by TemporaryDirectory.
This commit is contained in:
@@ -19,9 +19,10 @@ def temporary_home(monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
|
|||||||
monkeypatch.chdir(str(path))
|
monkeypatch.chdir(str(path))
|
||||||
yield path
|
yield path
|
||||||
else:
|
else:
|
||||||
with tempfile.TemporaryDirectory(prefix="pytest-") as dirpath:
|
with tempfile.TemporaryDirectory(prefix="pytest-") as _dirpath:
|
||||||
|
dirpath = Path(_dirpath)
|
||||||
monkeypatch.setenv("HOME", str(dirpath))
|
monkeypatch.setenv("HOME", str(dirpath))
|
||||||
monkeypatch.setenv("XDG_CONFIG_HOME", str(Path(dirpath) / ".config"))
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(dirpath / ".config"))
|
||||||
monkeypatch.chdir(str(dirpath))
|
monkeypatch.chdir(str(dirpath))
|
||||||
log.debug("Temp HOME directory: %s", str(dirpath))
|
log.debug("Temp HOME directory: %s", str(dirpath))
|
||||||
yield Path(dirpath)
|
yield dirpath
|
||||||
|
|||||||
@@ -202,8 +202,8 @@ def generate_facts(
|
|||||||
prompt: Callable[[str, str], str] = prompt_func,
|
prompt: Callable[[str, str], str] = prompt_func,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
was_regenerated = False
|
was_regenerated = False
|
||||||
with TemporaryDirectory(prefix="facts-generate-") as tmp:
|
with TemporaryDirectory(prefix="facts-generate-") as _tmpdir:
|
||||||
tmpdir = Path(tmp)
|
tmpdir = Path(_tmpdir).resolve()
|
||||||
|
|
||||||
for machine in machines:
|
for machine in machines:
|
||||||
errors = 0
|
errors = 0
|
||||||
|
|||||||
@@ -19,8 +19,8 @@ def upload_secrets(machine: Machine) -> None:
|
|||||||
machine.info("Secrets already uploaded")
|
machine.info("Secrets already uploaded")
|
||||||
return
|
return
|
||||||
|
|
||||||
with TemporaryDirectory(prefix="facts-upload-") as tempdir:
|
with TemporaryDirectory(prefix="facts-upload-") as _tempdir:
|
||||||
local_secret_dir = Path(tempdir)
|
local_secret_dir = Path(_tempdir).resolve()
|
||||||
secret_facts_store.upload(local_secret_dir)
|
secret_facts_store.upload(local_secret_dir)
|
||||||
remote_secret_dir = Path(machine.secrets_upload_directory)
|
remote_secret_dir = Path(machine.secrets_upload_directory)
|
||||||
|
|
||||||
|
|||||||
@@ -100,8 +100,8 @@ def flash_machine(
|
|||||||
secret_facts_store: SecretStoreBase = secret_facts_module.SecretStore(
|
secret_facts_store: SecretStoreBase = secret_facts_module.SecretStore(
|
||||||
machine=machine
|
machine=machine
|
||||||
)
|
)
|
||||||
with TemporaryDirectory(prefix="disko-install-") as tmpdir_:
|
with TemporaryDirectory(prefix="disko-install-") as _tmpdir:
|
||||||
tmpdir = Path(tmpdir_)
|
tmpdir = Path(_tmpdir)
|
||||||
upload_dir = machine.secrets_upload_directory
|
upload_dir = machine.secrets_upload_directory
|
||||||
|
|
||||||
if upload_dir.startswith("/"):
|
if upload_dir.startswith("/"):
|
||||||
|
|||||||
@@ -61,8 +61,9 @@ def install_machine(opts: InstallOptions) -> None:
|
|||||||
generate_facts([machine])
|
generate_facts([machine])
|
||||||
generate_vars([machine])
|
generate_vars([machine])
|
||||||
|
|
||||||
with TemporaryDirectory(prefix="nixos-install-") as base_directory:
|
with TemporaryDirectory(prefix="nixos-install-") as _base_directory:
|
||||||
activation_secrets = Path(base_directory) / "activation_secrets"
|
base_directory = Path(_base_directory).resolve()
|
||||||
|
activation_secrets = base_directory / "activation_secrets"
|
||||||
upload_dir = activation_secrets / machine.secrets_upload_directory.lstrip("/")
|
upload_dir = activation_secrets / machine.secrets_upload_directory.lstrip("/")
|
||||||
upload_dir.mkdir(parents=True)
|
upload_dir.mkdir(parents=True)
|
||||||
machine.secret_facts_store.upload(upload_dir)
|
machine.secret_facts_store.upload(upload_dir)
|
||||||
@@ -70,7 +71,7 @@ def install_machine(opts: InstallOptions) -> None:
|
|||||||
upload_dir, phases=["activation", "users", "services"]
|
upload_dir, phases=["activation", "users", "services"]
|
||||||
)
|
)
|
||||||
|
|
||||||
partitioning_secrets = Path(base_directory) / "partitioning_secrets"
|
partitioning_secrets = base_directory / "partitioning_secrets"
|
||||||
partitioning_secrets.mkdir(parents=True)
|
partitioning_secrets.mkdir(parents=True)
|
||||||
machine.secret_vars_store.populate_dir(
|
machine.secret_vars_store.populate_dir(
|
||||||
partitioning_secrets, phases=["partitioning"]
|
partitioning_secrets, phases=["partitioning"]
|
||||||
|
|||||||
@@ -56,8 +56,8 @@ def morph_machine(
|
|||||||
).stdout.rstrip()
|
).stdout.rstrip()
|
||||||
archive_path = json.loads(archive_json)["path"]
|
archive_path = json.loads(archive_json)["path"]
|
||||||
|
|
||||||
with TemporaryDirectory(prefix="morph-") as temp_dir:
|
with TemporaryDirectory(prefix="morph-") as _temp_dir:
|
||||||
flakedir = Path(temp_dir) / "flake"
|
flakedir = Path(_temp_dir).resolve() / "flake"
|
||||||
|
|
||||||
flakedir.mkdir(parents=True, exist_ok=True)
|
flakedir.mkdir(parents=True, exist_ok=True)
|
||||||
run(["cp", "-r", archive_path + "/.", str(flakedir)])
|
run(["cp", "-r", archive_path + "/.", str(flakedir)])
|
||||||
|
|||||||
@@ -181,8 +181,8 @@ def execute_generator(
|
|||||||
raise ClanError(msg) from e
|
raise ClanError(msg) from e
|
||||||
|
|
||||||
env = os.environ.copy()
|
env = os.environ.copy()
|
||||||
with TemporaryDirectory(prefix="vars-") as tmp:
|
with TemporaryDirectory(prefix="vars-") as _tmpdir:
|
||||||
tmpdir = Path(tmp)
|
tmpdir = Path(_tmpdir).resolve()
|
||||||
tmpdir_in = tmpdir / "in"
|
tmpdir_in = tmpdir / "in"
|
||||||
tmpdir_prompts = tmpdir / "prompts"
|
tmpdir_prompts = tmpdir / "prompts"
|
||||||
tmpdir_out = tmpdir / "out"
|
tmpdir_out = tmpdir / "out"
|
||||||
|
|||||||
@@ -231,8 +231,8 @@ class SecretStore(StoreBase):
|
|||||||
if not self.needs_upload():
|
if not self.needs_upload():
|
||||||
log.info("Secrets already uploaded")
|
log.info("Secrets already uploaded")
|
||||||
return
|
return
|
||||||
with TemporaryDirectory(prefix="vars-upload-") as tempdir:
|
with TemporaryDirectory(prefix="vars-upload-") as _tempdir:
|
||||||
pass_dir = Path(tempdir)
|
pass_dir = Path(_tempdir).resolve()
|
||||||
self.populate_dir(pass_dir, phases)
|
self.populate_dir(pass_dir, phases)
|
||||||
upload_dir = Path(
|
upload_dir = Path(
|
||||||
self.machine.deployment["password-store"]["secretLocation"]
|
self.machine.deployment["password-store"]["secretLocation"]
|
||||||
|
|||||||
@@ -224,8 +224,8 @@ class SecretStore(StoreBase):
|
|||||||
if "partitioning" in phases:
|
if "partitioning" in phases:
|
||||||
msg = "Cannot upload partitioning secrets"
|
msg = "Cannot upload partitioning secrets"
|
||||||
raise NotImplementedError(msg)
|
raise NotImplementedError(msg)
|
||||||
with TemporaryDirectory(prefix="sops-upload-") as tempdir:
|
with TemporaryDirectory(prefix="sops-upload-") as _tempdir:
|
||||||
sops_upload_dir = Path(tempdir)
|
sops_upload_dir = Path(_tempdir).resolve()
|
||||||
self.populate_dir(sops_upload_dir, phases)
|
self.populate_dir(sops_upload_dir, phases)
|
||||||
upload(self.machine.target_host, sops_upload_dir, Path("/var/lib/sops-nix"))
|
upload(self.machine.target_host, sops_upload_dir, Path("/var/lib/sops-nix"))
|
||||||
|
|
||||||
|
|||||||
@@ -209,12 +209,13 @@ class ClanFlake:
|
|||||||
@pytest.fixture(scope="session")
|
@pytest.fixture(scope="session")
|
||||||
def minimal_flake_template() -> Iterator[ClanFlake]:
|
def minimal_flake_template() -> Iterator[ClanFlake]:
|
||||||
with (
|
with (
|
||||||
tempfile.TemporaryDirectory(prefix="flake-") as home,
|
tempfile.TemporaryDirectory(prefix="flake-") as _home,
|
||||||
pytest.MonkeyPatch.context() as mp,
|
pytest.MonkeyPatch.context() as mp,
|
||||||
):
|
):
|
||||||
mp.setenv("HOME", home)
|
home = Path(_home).resolve()
|
||||||
|
mp.setenv("HOME", str(home))
|
||||||
flake = ClanFlake(
|
flake = ClanFlake(
|
||||||
temporary_home=Path(home),
|
temporary_home=home,
|
||||||
flake_template=clan_templates(TemplateType.CLAN) / "minimal",
|
flake_template=clan_templates(TemplateType.CLAN) / "minimal",
|
||||||
)
|
)
|
||||||
flake.init_from_template()
|
flake.init_from_template()
|
||||||
|
|||||||
@@ -11,12 +11,13 @@ log = logging.getLogger(__name__)
|
|||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def temporary_home(monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
|
def temporary_home(monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
|
||||||
with tempfile.TemporaryDirectory(prefix="pytest-home-") as dirpath:
|
with tempfile.TemporaryDirectory(prefix="pytest-home-") as _dirpath:
|
||||||
|
dirpath = Path(_dirpath).resolve()
|
||||||
xdg_runtime_dir = os.getenv("XDG_RUNTIME_DIR")
|
xdg_runtime_dir = os.getenv("XDG_RUNTIME_DIR")
|
||||||
monkeypatch.setenv("HOME", str(dirpath))
|
monkeypatch.setenv("HOME", str(dirpath))
|
||||||
monkeypatch.setenv("XDG_CONFIG_HOME", str(Path(dirpath) / ".config"))
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(dirpath / ".config"))
|
||||||
|
|
||||||
runtime_dir = Path(dirpath) / "xdg-runtime-dir"
|
runtime_dir = dirpath / "xdg-runtime-dir"
|
||||||
runtime_dir.mkdir()
|
runtime_dir.mkdir()
|
||||||
runtime_dir.chmod(0o700)
|
runtime_dir.chmod(0o700)
|
||||||
|
|
||||||
@@ -34,10 +35,10 @@ def temporary_home(monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
|
|||||||
|
|
||||||
monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime_dir))
|
monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime_dir))
|
||||||
monkeypatch.chdir(str(dirpath))
|
monkeypatch.chdir(str(dirpath))
|
||||||
yield Path(dirpath)
|
yield dirpath
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def temp_dir() -> Iterator[Path]:
|
def temp_dir() -> Iterator[Path]:
|
||||||
with tempfile.TemporaryDirectory(prefix="pytest-") as dirpath:
|
with tempfile.TemporaryDirectory(prefix="pytest-") as _dirpath:
|
||||||
yield Path(dirpath)
|
yield Path(_dirpath).resolve()
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from tempfile import TemporaryDirectory
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from clan_cli.clan_uri import ClanURI
|
from clan_cli.clan_uri import ClanURI
|
||||||
@@ -36,14 +35,13 @@ def test_firefox_strip_uri() -> None:
|
|||||||
assert uri.get_url() == "git+https://git.clan.lol/clan/democlan.git"
|
assert uri.get_url() == "git+https://git.clan.lol/clan/democlan.git"
|
||||||
|
|
||||||
|
|
||||||
def test_local_uri() -> None:
|
def test_local_uri(temp_dir: Path) -> None:
|
||||||
with TemporaryDirectory(prefix="clan_test") as tempdir:
|
flake_nix = temp_dir / "flake.nix"
|
||||||
flake_nix = Path(tempdir) / "flake.nix"
|
flake_nix.write_text("outputs = _: {}")
|
||||||
flake_nix.write_text("outputs = _: {}")
|
|
||||||
|
|
||||||
# Create a ClanURI object from a local URI
|
# Create a ClanURI object from a local URI
|
||||||
uri = ClanURI.from_str(f"clan://file://{tempdir}")
|
uri = ClanURI.from_str(f"clan://file://{temp_dir}")
|
||||||
assert uri.flake.path == Path(tempdir)
|
assert uri.flake.path == temp_dir
|
||||||
|
|
||||||
|
|
||||||
def test_is_remote() -> None:
|
def test_is_remote() -> None:
|
||||||
@@ -79,25 +77,23 @@ def test_from_str_remote() -> None:
|
|||||||
assert uri.flake.identifier == "https://example.com"
|
assert uri.flake.identifier == "https://example.com"
|
||||||
|
|
||||||
|
|
||||||
def test_from_str_local() -> None:
|
def test_from_str_local(temp_dir: Path) -> None:
|
||||||
with TemporaryDirectory(prefix="clan_test") as tempdir:
|
flake_nix = temp_dir / "flake.nix"
|
||||||
flake_nix = Path(tempdir) / "flake.nix"
|
flake_nix.write_text("outputs = _: {}")
|
||||||
flake_nix.write_text("outputs = _: {}")
|
|
||||||
|
|
||||||
uri = ClanURI.from_str(url=tempdir, machine_name="myVM")
|
uri = ClanURI.from_str(url=str(temp_dir), machine_name="myVM")
|
||||||
assert uri.get_url().endswith(tempdir)
|
assert uri.get_url().endswith(str(temp_dir))
|
||||||
assert uri.machine_name == "myVM"
|
assert uri.machine_name == "myVM"
|
||||||
assert uri.flake.is_local
|
assert uri.flake.is_local
|
||||||
assert str(uri.flake).endswith(tempdir) # type: ignore
|
assert str(uri.flake).endswith(str(temp_dir))
|
||||||
|
|
||||||
|
|
||||||
def test_from_str_local_no_machine() -> None:
|
def test_from_str_local_no_machine(temp_dir: Path) -> None:
|
||||||
with TemporaryDirectory(prefix="clan_test") as tempdir:
|
flake_nix = temp_dir / "flake.nix"
|
||||||
flake_nix = Path(tempdir) / "flake.nix"
|
flake_nix.write_text("outputs = _: {}")
|
||||||
flake_nix.write_text("outputs = _: {}")
|
|
||||||
|
|
||||||
uri = ClanURI.from_str(tempdir)
|
uri = ClanURI.from_str(str(temp_dir))
|
||||||
assert uri.get_url().endswith(tempdir)
|
assert uri.get_url().endswith(str(temp_dir))
|
||||||
assert uri.machine_name == "defaultVM"
|
assert uri.machine_name == "defaultVM"
|
||||||
assert uri.flake.is_local
|
assert uri.flake.is_local
|
||||||
assert str(uri.flake).endswith(tempdir) # type: ignore
|
assert str(uri.flake).endswith(str(temp_dir))
|
||||||
|
|||||||
Reference in New Issue
Block a user