Merge pull request #1689 from Emantor/topic/managedfile-symlink

Managedfile: replace ssh usage for localhost

Author: Emantor

labgrid/util/managedfile.py

@@ -68,18 +68,23 @@ def sync_to_resource(self, symlink=None):
                     f"{self.rpath}{os.path.basename(self.local_path)}"
                 )
         else:
-            conn = sshmanager.open("localhost")
             self.rpath = os.path.dirname(self.local_path) + "/"
 
         if symlink is not None:
             self.logger.info("Linking")
-            try:
-                conn.run_check(f"test ! -e {symlink} -o -L {symlink}")
-            except ExecutionError:
-                raise ManagedFileError(f"Path {symlink} exists but is not a symlink.")
-            # use short options to be compatible with busybox
-            # --symbolic --force --no-dereference
-            conn.run_check(f"ln -sfn {self.rpath}{os.path.basename(self.local_path)} {symlink}")
+            if isinstance(self.resource, NetworkResource):
+                try:
+                    conn.run_check(f"test ! -e {symlink} -o -L {symlink}")
+                except ExecutionError:
+                    raise ManagedFileError(f"Path {symlink} exists but is not a symlink.")
+                # use short options to be compatible with busybox
+                # --symbolic --force --no-dereference
+                conn.run_check(f"ln -sfn {self.rpath}{os.path.basename(self.local_path)} {symlink}")
+            else:
+                if os.path.exists(symlink) and not os.path.islink(symlink):
+                    raise ManagedFileError(f"Path {symlink} exists but is not a symlink.")
+                os.symlink(f"{self.rpath}{os.path.basename(self.local_path)}", symlink)
+
 
     def _on_nfs(self, conn):
         if self._on_nfs_cached is not None:
@@ -131,6 +136,9 @@ def get_remote_path(self):
             str: path to the file on the remote host
         """
         if isinstance(self.resource, NetworkResource):
+            if self.rpath is None:
+                raise ManagedFileError("sync_to_resource() needs to be called before the remote-path can be retrieved")
+
             return f"{self.rpath}{os.path.basename(self.local_path)}"
 
         return self.local_path

tests/test_util.py

@@ -1,4 +1,5 @@
 import os.path
+import re
 import subprocess
 import socket
 import atexit
@@ -14,7 +15,7 @@
 from labgrid.util.helper import get_free_port
 from labgrid.util.ssh import ForwardError, SSHConnection, sshmanager
 from labgrid.util.proxy import proxymanager
-from labgrid.util.managedfile import ManagedFile
+from labgrid.util.managedfile import ManagedFile, ManagedFileError
 from labgrid.driver.exception import ExecutionError
 from labgrid.resource.serialport import NetworkSerialPort
 from labgrid.resource.common import Resource, NetworkResource
@@ -352,6 +353,47 @@ def test_local_managedfile(target, tmpdir):
     assert hash == mf.get_hash()
     assert str(t) == mf.get_remote_path()
 
+def test_network_managedfile_no_sync(target, tmpdir):
+    res = NetworkResource(target, "test", "localhost")
+    t = tmpdir.join("test")
+    t.write(
+"""
+Test
+"""
+    )
+    mf = ManagedFile(t, res, detect_nfs=False)
+
+    expected = re.escape("sync_to_resource() needs to be called before the remote-path can be retrieved")
+    with pytest.raises(ManagedFileError, match=expected) as e:
+        mf.get_remote_path()
+
+def test_local_managedfile_symlink(target, tmpdir):
+    res = Resource(target, "test")
+    t = tmpdir.join("test")
+    t.write(
+"""
+Test
+"""
+    )
+    mf = ManagedFile(t, res, detect_nfs=False)
+    mf.sync_to_resource(symlink=tmpdir.join("link"))
+
+    assert str(t) == mf.get_remote_path()
+    assert os.path.islink(tmpdir.join("link"))
+
+@pytest.mark.localsshmanager
+def test_remote_managedfile_symlink(target, tmpdir):
+    res = NetworkResource(target, "test", "localhost")
+    t = tmpdir.join("test")
+    t.write(
+"""
+Test
+"""
+    )
+    mf = ManagedFile(t, res, detect_nfs=False)
+    mf.sync_to_resource(symlink=tmpdir.join("link"))
+
+    assert os.path.islink(tmpdir.join("link"))
 
 def test_find_dict():
     dict_a = {"a": {"a.a": {"a.a.a": "a.a.a_val"}}, "b": "b_val"}