Document FTP: write Yaml tests, have special helper module.
bzr revid: p_christ@hol.gr-20100701224249-fmble80t8nlg2azq
This commit is contained in:
@@ -2,65 +2,212 @@
|
||||
In order to test the document_ftp functionality
|
||||
-
|
||||
I open the 8021 port and see for ftp presence there
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_plain_ftp()
|
||||
assert ftp.sock and (ftp.lastresp == '220'), ftp.lastresp
|
||||
-
|
||||
I read the list of databases at port 8021 and confirm our db is
|
||||
there
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_login(cr, uid, self)
|
||||
assert cr.dbname in ftp.nlst("/")
|
||||
-
|
||||
I try to locate the default "Documents" folder in the db.
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_login(cr, uid, self)
|
||||
ftp.cwd('/Documents')
|
||||
-
|
||||
I create a "test.txt" file at the server (directly). The file
|
||||
should have the "abcd" content
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
fdata = StringIO('abcd')
|
||||
ftp.storbinary('STOR test.txt', fdata)
|
||||
-
|
||||
I look for the "test.txt" file at the server
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
assert ftp.nlst("test.txt") == ['test.txt']
|
||||
-
|
||||
I check that the content of "test.txt" is "abcd"
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
assert te.get_ftp_fulldata(ftp, "test.txt") == 'abcd'
|
||||
-
|
||||
I append the string 'defgh' into "test.txt"
|
||||
-
|
||||
I check that the content of "text.txt" is 'abcdefgh'
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
fdata = StringIO('defgh')
|
||||
ftp.storbinary('APPE test.txt', fdata)
|
||||
-
|
||||
I check that the content of "text.txt" is 'abcddefgh'
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
assert te.get_ftp_fulldata(ftp, "test.txt") == 'abcddefgh'
|
||||
-
|
||||
I try to cd into an non-existing folder 'Not-This'
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_login(cr, uid, self)
|
||||
try:
|
||||
ftp.cwd('/Not-This')
|
||||
assert False, "We should't be able to change here"
|
||||
except OSError, err:
|
||||
assert err.errno == 2, err.errno
|
||||
-
|
||||
I create a "test2.txt" file through FTP.
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
fdata = StringIO('abcd')
|
||||
ftp.storbinary('STOR test2.txt', fdata)
|
||||
-
|
||||
I look for the "test2.txt" file at the server
|
||||
-
|
||||
I look for the "test2.txt" file through FTP.
|
||||
!python {model: ir.attachment }: |
|
||||
ids = self.search(cr, uid, [('name', '=', 'test2.txt')])
|
||||
assert ids
|
||||
-
|
||||
I delete the "test2.txt" file using FTP.
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
ftp.delete('test2.txt')
|
||||
-
|
||||
I check at the server that test2.txt is deleted
|
||||
-
|
||||
!python {model: ir.attachment }: |
|
||||
ids = self.search(cr, uid, [('name', '=', 'test2.txt')])
|
||||
assert not ids
|
||||
-
|
||||
I create a test2.txt file again.
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
fdata = StringIO('abcd')
|
||||
ftp.storbinary('STOR test2.txt', fdata)
|
||||
-
|
||||
I delete the test2.txt from the server (RPC).
|
||||
-
|
||||
!delete { model: ir.attachment, id:, search: "[('name','=','test2.txt')]" }
|
||||
-
|
||||
I check through FTP that test2.txt does not appear.
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
assert ftp.nlst("test2.txt") == []
|
||||
-
|
||||
I create a "test-name.txt" file
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
fdata = StringIO('abcd')
|
||||
ftp.storbinary('STOR test-name.txt', fdata)
|
||||
-
|
||||
I rename the "test-name.txt" file through ftp.
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
ftp.rename("test-name.txt", "test-renamed.txt")
|
||||
-
|
||||
I check that test-name.txt has been renamed.
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
assert ftp.nlst("test-name.txt") == []
|
||||
assert ftp.nlst("test-renamed.txt") == ['test-renamed.txt']
|
||||
-
|
||||
I create a new folder 'Test-Folder2' through FTP
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
ftp.mkd("Test-Folder2")
|
||||
-
|
||||
I create a file 'test3.txt' at the 'Test-Folder2'
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents/Test-Folder2')
|
||||
fdata = StringIO('abcd')
|
||||
ftp.storbinary('STOR test3.txt', fdata)
|
||||
-
|
||||
I try to retrieve test3.txt
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents/Test-Folder2')
|
||||
assert ftp.nlst("test3.txt") == ['test3.txt']
|
||||
-
|
||||
I create a new folder, 'Test-Folder3', through FTP
|
||||
I try to move test3.txt to 'Test-Folder3'
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents')
|
||||
ftp.mkd("Test-Folder2")
|
||||
# TODO move
|
||||
-
|
||||
I remove the 'Test-Folder3'
|
||||
-
|
||||
I check that test3.txt is removed.
|
||||
-
|
||||
I create 1000 files through FTP
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
from cStringIO import StringIO
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents/Test-Folder2')
|
||||
fdata = StringIO('abcd')
|
||||
# TODO speed
|
||||
for i in range(0, 1000):
|
||||
fdata.seek(0)
|
||||
ftp.storbinary('STOR test-name%s.txt' %i, fdata)
|
||||
-
|
||||
I list the 1000 files, check speed
|
||||
-
|
||||
!python {model: ir.attachment}: |
|
||||
from document_ftp import test_easyftp as te
|
||||
ftp = te.get_ftp_folder(cr, uid, self, 'Documents/Test-Folder2')
|
||||
# TODO speed
|
||||
assert len(ftp.nlst() > 1000)
|
||||
-
|
||||
I read the 1000 files, check speed
|
||||
# TODO
|
||||
-
|
||||
I move the 1000 files to 'Test-Folder2'
|
||||
# TODO
|
||||
-
|
||||
@@ -0,0 +1,66 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
##############################################################################
|
||||
#
|
||||
# OpenERP, Open Source Management Solution
|
||||
# Copyright (C) 2004-2010 Tiny SPRL (<http://tiny.be>).
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as
|
||||
# published by the Free Software Foundation, either version 3 of the
|
||||
# License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
#
|
||||
##############################################################################
|
||||
|
||||
""" This is a testing module, which exports some functions for the YAML tests.
|
||||
Instead of repeating the same FTP code all over, we prefer to have
|
||||
it in this file
|
||||
"""
|
||||
|
||||
from ftplib import FTP
|
||||
from tools.misc import detect_ip_addr
|
||||
from tools import config
|
||||
|
||||
def get_plain_ftp():
|
||||
ftp = FTP()
|
||||
host = config.get('ftp_server_host', '127.0.0.1')
|
||||
port = config.get('ftp_server_port','8021')
|
||||
ftp.connect(host,port)
|
||||
return ftp
|
||||
|
||||
def get_ftp_login(cr, uid, ormobj):
|
||||
ftp = get_plain_ftp()
|
||||
user = ormobj.pool.get('res.users').read(cr, uid, uid)
|
||||
ftp.login(user.get('login',''),user.get('password',''))
|
||||
ftp.cwd("/" + cr.dbname)
|
||||
return ftp
|
||||
|
||||
def get_ftp_anonymous(cr):
|
||||
ftp = get_plain_ftp()
|
||||
ftp.login('anonymous', 'the-test')
|
||||
ftp.cwd("/")
|
||||
return ftp
|
||||
|
||||
def get_ftp_folder(cr, uid, ormobj, foldername):
|
||||
ftp = get_ftp_login(cr, uid, ormobj)
|
||||
ftp.cwd("/" + cr.dbname+"/"+foldername)
|
||||
return ftp
|
||||
|
||||
def get_ftp_fulldata(ftp, fname, limit=8192):
|
||||
from functools import partial
|
||||
data = []
|
||||
def ffp(data, ndata):
|
||||
if len(data)+ len(ndata) > limit:
|
||||
raise IndexError('Data over the limit')
|
||||
data.append(ndata)
|
||||
ftp.retrbinary('RETR %s' % fname, partial(ffp,data))
|
||||
return ''.join(data)
|
||||
|
||||
#eof
|
||||
Reference in New Issue
Block a user