Zato ftp test case

gistfile1.py

import unittest
import uuid

from zato.client import JSONClient
from fs.ftpfs import FTPFS

directory = '/home/ftp/ftpuser/'
username = 'ftpuser'
password = 'ftpuser'

def get_rand_string():
    return uuid.uuid4().hex

def run_service(filepath, directory, expected_contents):
    address = 'http://localhost:11223'
    path = '/ftp-service'
    auth = (username, password)

    client = JSONClient(address, path, auth)
    payload = {
        'connection' : 'My FTP',
        'file_path' : filepath,
        'directory' : directory,
        'content' : expected_contents}

    response = client.invoke(payload)
    return response.ok

class Tests(unittest.TestCase):
    def setUp(self):
        self.dirname = get_rand_string()
        self.filename = get_rand_string()
        self.expected_contents = get_rand_string()

        self.new_dir = directory + '%s' % (self.dirname)
        self.filepath = self.new_dir + '/%s' % (self.filename)

        self.ftp = FTPFS(host='127.0.0.1', user=username, passwd=password, timeout=10, port=21, dircache=True)

    def tearDown(self):
        self.ftp.remove(self.filepath)
        self.ftp.removedir(self.new_dir)

    def test_ftp(self):
        """ Check if directory is created. Check if file is created. Check if list of files has one item.
        Check if file content is valid.
        """
        self.assertEqual(run_service(self.filepath, self.new_dir, self.expected_contents), True)
        self.assertTrue(self.ftp.isdir(self.new_dir))
        self.assertTrue(self.ftp.isfile(self.filepath))
        self.assertEqual(len(self.ftp.listdir(self.new_dir)), 1)
        self.assertEqual(self.ftp.getcontents(self.filepath), self.expected_contents)

def main():
    unittest.main()

if __name__ == '__main__':
    main()