mfd-ftp 1.8.0

Module for handling operations with File Transfer Protocol on client and server sessions.

[!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.

MFD FTP

Module for handling File Transfer Protocol.

Usage

FTPUtils

FTPUtils(ip, username, password)
ip: ip_address - correct IPv4/6 ip address
username, password: str

Available methods

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

---

FTPServer

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}

FTP Client

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

FTP-Client-Traffic

    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
        """

FTP-Server-Traffic

    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].

Traffic Manager

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.

Traffic API

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 indefinitely
• stop() -> None - stop traffic
• run(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: True

Issue reporting

If you encounter any bugs or have suggestions for improvements, you're welcome to contribute directly or open an issue here.