-
Notifications
You must be signed in to change notification settings - Fork 16
Open
Description
This issue describe a problem with the Jubilant Juju.scp function (wrapping juju's scp command), where it cannot locate local files.
How to reproduce
import os
import pathlib
import tempfile
import jubilant
from jubilant import Juju
from jubilant.statustypes import UnitStatus
MYSQL_APP_NAME = "mysql"
def _get_juju_keys() -> tuple[str, str]:
"""Get Juju public and private keys."""
config_dir = pathlib.Path("~/.local/share/juju")
if juju_data := os.getenv("JUJU_DATA"):
config_dir = pathlib.Path(juju_data)
return (
str(config_dir.expanduser().resolve() / "ssh" / "juju_id_rsa.pub"),
str(config_dir.expanduser().resolve() / "ssh" / "juju_id_rsa"),
)
def get_app_units(juju: Juju, app_name: str) -> dict[str, UnitStatus]:
"""Get the units for the given application."""
model_status = juju.status()
app_status = model_status.apps[app_name]
return app_status.units
def scp_unit_file(juju: Juju, source: str, target: str) -> None:
"""Copy a file from source to target."""
juju_keys = _get_juju_keys()
juju.scp(source, target, host_key_checks=False, scp_options=["-B", "-i", juju_keys[1]])
if __name__ == "__main__":
juju = Juju(model="testing")
juju.deploy(
MYSQL_APP_NAME,
app=MYSQL_APP_NAME,
base="ubuntu@22.04",
config={"profile": "testing"},
num_units=1,
)
juju.wait(
ready=lambda status: all((
jubilant.all_agents_idle(status),
jubilant.all_active(status, MYSQL_APP_NAME),
)),
timeout=20 * 60,
)
for unit_name in get_app_units(juju, MYSQL_APP_NAME):
with tempfile.NamedTemporaryFile(mode="w") as temp_file:
temp_file.write("data")
temp_file.flush()
scp_unit_file(
juju=juju,
source=f"{temp_file.name}",
target=f"{unit_name}:/tmp/file",
)Traceback:
juju scp --model testing --no-host-key-checks -- -B -i /root/.local/share/juju/ssh/juju_id_rsa /tmp/tmp12d6m2pr mysql/0:/tmp/file
ERROR exit status 255 (scp: stat local "/tmp/tmp12d6m2pr": No such file or directory)
Work-around
- def scp_unit_file(juju: Juju, source: str, target: str) -> None:
- """Copy a file from source to target."""
- juju_keys = _get_juju_keys()
- juju.scp(source, target, host_key_checks=False, scp_options=["-B", "-i", juju_keys[1]])
+ import subprocess
+
+ def get_unit_ip(juju: Juju, app_name: str, unit_name: str) -> str:
+ """Get the application unit IP."""
+ model_status = juju.status()
+ app_status = model_status.apps[app_name]
+ for name, status in app_status.units.items():
+ if name == unit_name:
+ return status.public_address
+
+ raise Exception("No application unit found")
+
+
+ def scp_file_to_unit(juju: Juju, app_name: str, unit_name: str, source_path: str) -> None:
+ """Copy a file from a given unit."""
+ juju_keys = _get_juju_keys()
+ unit_address = get_unit_ip(juju, app_name, unit_name)
+ unit_username = "ubuntu"
+
+ subprocess.check_output([
+ "scp",
+ "-B",
+ "-o",
+ "StrictHostKeyChecking=no",
+ "-i",
+ juju_keys[1],
+ f"{source_path}",
+ f"{unit_username}@{unit_address}:/tmp/file",
+ ])Versions
- Jubilant:
1.4.0(latest). - Python:
3.10 - Juju:
3.6.9(latest) - LXD:
5.21.4(latest) - Ubuntu:
24.04(latest LTS).
Metadata
Metadata
Assignees
Labels
No labels