import os.path
import attr
from ..factory import target_factory
from ..step import step
from .common import Driver
from ..util.managedfile import ManagedFile
[docs]@attr.s(eq=False)
class BaseProviderDriver(Driver):
[docs] def __attrs_post_init__(self):
super().__attrs_post_init__()
[docs] @Driver.check_bound
def get_export_vars(self):
return {
"host": self.provider.host,
"internal": self.provider.internal,
"external": self.provider.external,
}
[docs] @Driver.check_active
@step(args=['filename'], result=True)
def stage(self, filename):
symlink = os.path.join(self.provider.internal, os.path.basename(filename))
assert symlink.startswith(self.provider.internal)
mf = ManagedFile(filename, self.provider)
mf.sync_to_resource(symlink=symlink)
return self.provider.external + symlink[len(self.provider.internal):]
[docs]@target_factory.reg_driver
@attr.s(eq=False)
class TFTPProviderDriver(BaseProviderDriver):
bindings = {
"provider": {"TFTPProvider", "RemoteTFTPProvider"},
}
[docs]@target_factory.reg_driver
@attr.s(eq=False)
class NFSPProviderDriver(BaseProviderDriver):
bindings = {
"provider": {"NFSProvider", "RemoteNFSProvider"},
}
[docs]@target_factory.reg_driver
@attr.s(eq=False)
class HTTPProviderDriver(BaseProviderDriver):
bindings = {
"provider": {"HTTPProvider", "RemoteHTTPProvider"},
}