Important
This project is under development. All source code and features on the main branch are for the purpose of testing or evaluation and not production ready.
Module for handling File Transfer Protocol.
FTPUtils(ip, username, password)
ip: ip_address - correct IPv4/6 ip address
username, password: str
return_dirs(catalog: str) -> typing.List - returning list of directories in root or given catalog
is_directory_on_ftp(directory: str) -> bool - checking if directory exists
copy_files_to_ftp(source_directory: str, destination_directory: str) -> bool - copying files from given directory to new directory on FTP
In code
from mfd_ftp import server
from mfd_connect import RPyCConnection
conn = RPyCConnection("0.0.0.0")
p = server.start_server_as_process(ip='0.0.0.0', port=21, directory=Path('/'),
username='test_user', password='xxxx')
p = server.start_remote_server_as_process(connection=conn, ip='0.0.0.0', port=21, directory=Path('/'),
username='test_user', password='xxxx')Standalone
python -m mfd_ftp.server --ip {ip} --port {port} --directory {directory} --username {username}
--password {password}
In code
from mfd_ftp import client
from mfd_connect import RPyCConnection
conn = RPyCConnection("0.0.0.0")
p = client.start_client_as_process(ip='0.0.0.0', port=21, task='send',
source='source_file', destination='transferred_file', username='test_user', password='xxxx')
p = client.start_remote_client_as_process(connection=conn, ip='0.0.0.0', port=21, task='send',
source='source_file', destination='transferred_file', username='test_user', password='xxxx')Standalone
python -m mfd_ftp.client --ip 0.0.0.0 --port 21 --task send --source original_filename
--destination transffered_filename --username username --password password
def __init__(
self,
connection: "Connection",
ip: Union[IPv4Address, IPv6Address],
port: int,
username: str,
password: str,
task: str,
source: str,
destination: str,
python_executable: Optional[str] = None,
timeout: Optional[int] = None,
) -> None:
"""FTP client traffic.
:param connection: FTP Connection
:param ip: Address of the FTP server.
:param port: Number of a port on which server is running.
:param username: Username for the server to login.
:param password: Password for the user.
:param task: Task for the client, can either be 'send' or 'receive'.
:param source: File to either receive or send.
:param destination: Destination for the received or sent file.
:param python_executable: Executable python
:param timeout: Timeout in seconds for operations, like connection attempt, or transfer.
:return: None
""" def __init__(
self,
connection: "Connection",
ip: Union[IPv4Address, IPv6Address],
port: int,
directory: str,
username: str,
password: str,
python_executable: Optional[str] = None,
permissions: Optional[str] = None,
) -> None:
"""FTP server traffic.
:param connection: FTP Connection
:param ip: Address of the FTP server.
:param port: Number of a port on which server is running.
:param directory: Path to directory which server should share.
:param username: Username for the server to login.
:param password: Password for the user.
:param python_executable: Executable python
:param permissions: User permissions, indicating which operations are allowed.
:return: None
"""In the server_traffic and client_traffic: python_executable parameter requires a python executable path(of the device under test) if connection object is anything else other than [RPyCConnection, LocalConnection].
FTP-Server/ClientTraffic have been implemented with Traffic interface. It can be used with Traffic Manager and Stream in mfd-traffic-manager as one of supported traffics. Details are available in mfd-traffic-manager documentation.
All classes for specific traffics should inherit from Traffic base class interface. It provides some abstract methods, all of them need to be implemented in child classes:
start() -> None- start traffic indefinitelystop() -> None- stop trafficrun(duration: int) -> None- run traffic for a specified number of seconds (usually for server, process should be running until stop)validate(validation_criteria: Optional[Dict[Callable, Dict[str, Any]]]) -> bool- validate traffic by passed criteria
Example usage for FTP:
from mfd_connect import RPyCConnection
from mfd_ftp import FTPClientTraffic, FTPServerTraffic
from mfd_traffic_manager import TrafficManager, Stream
def validate_files_transmitted(results: str, files_transmitted: int) -> bool:
"""Validate Files transmitted.
:param results: Results from FTP Process
:param files_transmitted: expected number of files transmitted
:return: status of validation
"""
ftp_files_transmitted = results.count("File was sent")
if ftp_files_transmitted != files_transmitted:
print("Number of files transmitted are not as expected")
return False
return True
server_conn = RPyCConnection("10.10.10.10")
client_conn = RPyCConnection("10.10.10.20")
ftp_server = FTPServerTraffic(
connection=server_conn, ip="10.10.10.10", port=21, directory="/", username="user", password="xxxx"
)
ftp_client = FTPClientTraffic(
connection=client_conn,
ip="10.10.10.10",
port=21,
username="user",
password="xxxx",
task="send",
source="source.txt",
destination="destination.txt",
)
manager = TrafficManager()
stream = Stream(clients=[ftp_client], server=ftp_server, name="Stream1")
manager.add_stream(stream)
manager.start(name="Stream1")
manager.stop(name="Stream1")
validation_criteria = {validate_files_transmitted: {"files_transmitted": 1}}
print(manager.validate(name="Stream1", clients_validation_criteria=validation_criteria))
# Output: TrueIf you encounter any bugs or have suggestions for improvements, you're welcome to contribute directly or open an issue here.