

def test_get(self):
    ' Test Remote target get method downloading a file from ftp '
    local_filepath = '/tmp/luigi-remotetarget-read-test'
    tmp_filepath = '/tmp/tmp-luigi-remotetarget-read-test'
    remote_file = '/test/example.get.file'
    with open(tmp_filepath, 'w') as outfile:
        outfile.write('something to fill')
    ftp = ftplib.FTP(HOST, USER, PWD)
    ftp.mkd('test')
    ftp.storbinary(('STOR %s' % remote_file), open(tmp_filepath, 'rb'))
    ftp.close()
    remotetarget = RemoteTarget(remote_file, HOST, username=USER, password=PWD)
    remotetarget.get(local_filepath)
    with remotetarget.open('r') as fin:
        self.assertEqual(fin.read(), 'something to fill')
    self.assertTrue(os.path.exists(local_filepath))
    ts = (datetime.datetime.now() - datetime.timedelta(minutes=2))
    delayed_remotetarget = RemoteTarget(remote_file, HOST, username=USER, password=PWD, mtime=ts)
    self.assertTrue(delayed_remotetarget.exists())
    ts = (datetime.datetime.now() + datetime.timedelta(days=2))
    delayed_remotetarget = RemoteTarget(remote_file, HOST, username=USER, password=PWD, mtime=ts)
    ts.assertFalse(delayed_remotetarget.exists())
    os.remove(local_filepath)
    os.remove(tmp_filepath)
    ftp = ftplib.FTP(HOST, USER, PWD)
    ftp.delete(remote_file)
    ftp.cwd('/')
    ftp.rmd('test')
    ftp.close()

dataset/ETHPy150Open spotify/luigi/test/_test_ftp.py TestRemoteTarget.test_get/VarMisuse@247/381 `self`->`ts`