33
44import pytest
55import re
6- import tempfile
7- import logging
86
9- from ..testgres import ExecUtilException
10- from ..testgres import InvalidOperationException
117from ..testgres import LocalOperations
128
13- from .helpers .run_conditions import RunConditions
14-
159
1610class TestLocalOperations :
1711
1812@pytest .fixture (scope = "function" ,autouse = True )
1913def setup (self ):
2014self .operations = LocalOperations ()
2115
22- def test_mkdtemp__default (self ):
23- path = self .operations .mkdtemp ()
24- logging .info ("Path is [{0}]." .format (path ))
25- assert os .path .exists (path )
26- os .rmdir (path )
27- assert not os .path .exists (path )
28-
29- def test_mkdtemp__custom (self ):
30- C_TEMPLATE = "abcdef"
31- path = self .operations .mkdtemp (C_TEMPLATE )
32- logging .info ("Path is [{0}]." .format (path ))
33- assert os .path .exists (path )
34- assert C_TEMPLATE in os .path .basename (path )
35- os .rmdir (path )
36- assert not os .path .exists (path )
37-
38- def test_exec_command_success (self ):
39- """
40- Test exec_command for successful command execution.
41- """
42- RunConditions .skip_if_windows ()
43-
44- cmd = "python3 --version"
45- response = self .operations .exec_command (cmd ,wait_exit = True ,shell = True )
46-
47- assert b'Python 3.' in response
48-
49- def test_exec_command_failure (self ):
50- """
51- Test exec_command for command execution failure.
52- """
53- RunConditions .skip_if_windows ()
54-
55- cmd = "nonexistent_command"
56- while True :
57- try :
58- self .operations .exec_command (cmd ,wait_exit = True ,shell = True )
59- except ExecUtilException as e :
60- assert type (e .exit_code )== int # noqa: E721
61- assert e .exit_code == 127
62-
63- assert type (e .message )== str # noqa: E721
64- assert type (e .error )== bytes # noqa: E721
65-
66- assert e .message .startswith ("Utility exited with non-zero code (127). Error:" )
67- assert "nonexistent_command" in e .message
68- assert "not found" in e .message
69- assert b"nonexistent_command" in e .error
70- assert b"not found" in e .error
71- break
72- raise Exception ("We wait an exception!" )
73-
74- def test_exec_command_failure__expect_error (self ):
75- """
76- Test exec_command for command execution failure.
77- """
78- RunConditions .skip_if_windows ()
79-
80- cmd = "nonexistent_command"
81-
82- exit_status ,result ,error = self .operations .exec_command (cmd ,verbose = True ,wait_exit = True ,shell = True ,expect_error = True )
83-
84- assert exit_status == 127
85- assert result == b''
86- assert type (error )== bytes # noqa: E721
87- assert b"nonexistent_command" in error
88- assert b"not found" in error
89-
90- def test_listdir (self ):
91- """
92- Test listdir for listing directory contents.
93- """
94- path = "/etc"
95- files = self .operations .listdir (path )
96- assert isinstance (files ,list )
97- for f in files :
98- assert f is not None
99- assert type (f )== str # noqa: E721
100-
101- def test_read__text (self ):
102- """
103- Test LocalOperations::read for text data.
104- """
105- filename = __file__ # current file
106-
107- with open (filename ,'r' )as file :# open in a text mode
108- response0 = file .read ()
109-
110- assert type (response0 )== str # noqa: E721
111-
112- response1 = self .operations .read (filename )
113- assert type (response1 )== str # noqa: E721
114- assert response1 == response0
115-
116- response2 = self .operations .read (filename ,encoding = None ,binary = False )
117- assert type (response2 )== str # noqa: E721
118- assert response2 == response0
119-
120- response3 = self .operations .read (filename ,encoding = "" )
121- assert type (response3 )== str # noqa: E721
122- assert response3 == response0
123-
124- response4 = self .operations .read (filename ,encoding = "UTF-8" )
125- assert type (response4 )== str # noqa: E721
126- assert response4 == response0
127-
128- def test_read__binary (self ):
129- """
130- Test LocalOperations::read for binary data.
131- """
132- filename = __file__ # current file
133-
134- with open (filename ,'rb' )as file :# open in a binary mode
135- response0 = file .read ()
136-
137- assert type (response0 )== bytes # noqa: E721
138-
139- response1 = self .operations .read (filename ,binary = True )
140- assert type (response1 )== bytes # noqa: E721
141- assert response1 == response0
142-
143- def test_read__binary_and_encoding (self ):
144- """
145- Test LocalOperations::read for binary data and encoding.
146- """
147- filename = __file__ # current file
148-
149- with pytest .raises (
150- InvalidOperationException ,
151- match = re .escape ("Enconding is not allowed for read binary operation" )):
152- self .operations .read (filename ,encoding = "" ,binary = True )
153-
15416def test_read__unknown_file (self ):
15517"""
15618 Test LocalOperations::read with unknown file.
@@ -159,40 +21,6 @@ def test_read__unknown_file(self):
15921with pytest .raises (FileNotFoundError ,match = re .escape ("[Errno 2] No such file or directory: '/dummy'" )):
16022self .operations .read ("/dummy" )
16123
162- def test_read_binary__spec (self ):
163- """
164- Test LocalOperations::read_binary.
165- """
166- filename = __file__ # current file
167-
168- with open (filename ,'rb' )as file :# open in a binary mode
169- response0 = file .read ()
170-
171- assert type (response0 )== bytes # noqa: E721
172-
173- response1 = self .operations .read_binary (filename ,0 )
174- assert type (response1 )== bytes # noqa: E721
175- assert response1 == response0
176-
177- response2 = self .operations .read_binary (filename ,1 )
178- assert type (response2 )== bytes # noqa: E721
179- assert len (response2 )< len (response1 )
180- assert len (response2 )+ 1 == len (response1 )
181- assert response2 == response1 [1 :]
182-
183- response3 = self .operations .read_binary (filename ,len (response1 ))
184- assert type (response3 )== bytes # noqa: E721
185- assert len (response3 )== 0
186-
187- response4 = self .operations .read_binary (filename ,len (response2 ))
188- assert type (response4 )== bytes # noqa: E721
189- assert len (response4 )== 1
190- assert response4 [0 ]== response1 [len (response1 )- 1 ]
191-
192- response5 = self .operations .read_binary (filename ,len (response1 )+ 1 )
193- assert type (response5 )== bytes # noqa: E721
194- assert len (response5 )== 0
195-
19624def test_read_binary__spec__unk_file (self ):
19725"""
19826 Test LocalOperations::read_binary with unknown file.
@@ -203,29 +31,6 @@ def test_read_binary__spec__unk_file(self):
20331match = re .escape ("[Errno 2] No such file or directory: '/dummy'" )):
20432self .operations .read_binary ("/dummy" ,0 )
20533
206- def test_read_binary__spec__negative_offset (self ):
207- """
208- Test LocalOperations::read_binary with negative offset.
209- """
210-
211- with pytest .raises (
212- ValueError ,
213- match = re .escape ("Negative 'offset' is not supported." )):
214- self .operations .read_binary (__file__ ,- 1 )
215-
216- def test_get_file_size (self ):
217- """
218- Test LocalOperations::get_file_size.
219- """
220- filename = __file__ # current file
221-
222- sz0 = os .path .getsize (filename )
223- assert type (sz0 )== int # noqa: E721
224-
225- sz1 = self .operations .get_file_size (filename )
226- assert type (sz1 )== int # noqa: E721
227- assert sz1 == sz0
228-
22934def test_get_file_size__unk_file (self ):
23035"""
23136 Test LocalOperations::get_file_size.
@@ -234,70 +39,6 @@ def test_get_file_size__unk_file(self):
23439with pytest .raises (FileNotFoundError ,match = re .escape ("[Errno 2] No such file or directory: '/dummy'" )):
23540self .operations .get_file_size ("/dummy" )
23641
237- def test_isfile_true (self ):
238- """
239- Test isfile for an existing file.
240- """
241- filename = __file__
242-
243- response = self .operations .isfile (filename )
244-
245- assert response is True
246-
247- def test_isfile_false__not_exist (self ):
248- """
249- Test isfile for a non-existing file.
250- """
251- filename = os .path .join (os .path .dirname (__file__ ),"nonexistent_file.txt" )
252-
253- response = self .operations .isfile (filename )
254-
255- assert response is False
256-
257- def test_isfile_false__directory (self ):
258- """
259- Test isfile for a firectory.
260- """
261- name = os .path .dirname (__file__ )
262-
263- assert self .operations .isdir (name )
264-
265- response = self .operations .isfile (name )
266-
267- assert response is False
268-
269- def test_isdir_true (self ):
270- """
271- Test isdir for an existing directory.
272- """
273- name = os .path .dirname (__file__ )
274-
275- response = self .operations .isdir (name )
276-
277- assert response is True
278-
279- def test_isdir_false__not_exist (self ):
280- """
281- Test isdir for a non-existing directory.
282- """
283- name = os .path .join (os .path .dirname (__file__ ),"it_is_nonexistent_directory" )
284-
285- response = self .operations .isdir (name )
286-
287- assert response is False
288-
289- def test_isdir_false__file (self ):
290- """
291- Test isdir for a file.
292- """
293- name = __file__
294-
295- assert self .operations .isfile (name )
296-
297- response = self .operations .isdir (name )
298-
299- assert response is False
300-
30142def test_cwd (self ):
30243"""
30344 Test cwd.
@@ -314,71 +55,3 @@ def test_cwd(self):
31455
31556# Comp result
31657assert v == expectedValue
317-
318- class tagWriteData001 :
319- def __init__ (self ,sign ,source ,cp_rw ,cp_truncate ,cp_binary ,cp_data ,result ):
320- self .sign = sign
321- self .source = source
322- self .call_param__rw = cp_rw
323- self .call_param__truncate = cp_truncate
324- self .call_param__binary = cp_binary
325- self .call_param__data = cp_data
326- self .result = result
327-
328- sm_write_data001 = [
329- tagWriteData001 ("A001" ,"1234567890" ,False ,False ,False ,"ABC" ,"1234567890ABC" ),
330- tagWriteData001 ("A002" ,b"1234567890" ,False ,False ,True ,b"ABC" ,b"1234567890ABC" ),
331-
332- tagWriteData001 ("B001" ,"1234567890" ,False ,True ,False ,"ABC" ,"ABC" ),
333- tagWriteData001 ("B002" ,"1234567890" ,False ,True ,False ,"ABC1234567890" ,"ABC1234567890" ),
334- tagWriteData001 ("B003" ,b"1234567890" ,False ,True ,True ,b"ABC" ,b"ABC" ),
335- tagWriteData001 ("B004" ,b"1234567890" ,False ,True ,True ,b"ABC1234567890" ,b"ABC1234567890" ),
336-
337- tagWriteData001 ("C001" ,"1234567890" ,True ,False ,False ,"ABC" ,"1234567890ABC" ),
338- tagWriteData001 ("C002" ,b"1234567890" ,True ,False ,True ,b"ABC" ,b"1234567890ABC" ),
339-
340- tagWriteData001 ("D001" ,"1234567890" ,True ,True ,False ,"ABC" ,"ABC" ),
341- tagWriteData001 ("D002" ,"1234567890" ,True ,True ,False ,"ABC1234567890" ,"ABC1234567890" ),
342- tagWriteData001 ("D003" ,b"1234567890" ,True ,True ,True ,b"ABC" ,b"ABC" ),
343- tagWriteData001 ("D004" ,b"1234567890" ,True ,True ,True ,b"ABC1234567890" ,b"ABC1234567890" ),
344-
345- tagWriteData001 ("E001" ,"\000 1234567890\000 " ,False ,False ,False ,"\000 ABC\000 " ,"\000 1234567890\000 \000 ABC\000 " ),
346- tagWriteData001 ("E002" ,b"\000 1234567890\000 " ,False ,False ,True ,b"\000 ABC\000 " ,b"\000 1234567890\000 \000 ABC\000 " ),
347-
348- tagWriteData001 ("F001" ,"a\n b\n " ,False ,False ,False , ["c" ,"d" ],"a\n b\n c\n d\n " ),
349- tagWriteData001 ("F002" ,b"a\n b\n " ,False ,False ,True , [b"c" ,b"d" ],b"a\n b\n c\n d\n " ),
350-
351- tagWriteData001 ("G001" ,"a\n b\n " ,False ,False ,False , ["c\n \n " ,"d\n " ],"a\n b\n c\n d\n " ),
352- tagWriteData001 ("G002" ,b"a\n b\n " ,False ,False ,True , [b"c\n \n " ,b"d\n " ],b"a\n b\n c\n d\n " ),
353- ]
354-
355- @pytest .fixture (
356- params = sm_write_data001 ,
357- ids = [x .sign for x in sm_write_data001 ],
358- )
359- def write_data001 (self ,request ):
360- assert isinstance (request ,pytest .FixtureRequest )
361- assert type (request .param )== __class__ .tagWriteData001 # noqa: E721
362- return request .param
363-
364- def test_write (self ,write_data001 ):
365- assert type (write_data001 )== __class__ .tagWriteData001 # noqa: E721
366-
367- mode = "w+b" if write_data001 .call_param__binary else "w+"
368-
369- with tempfile .NamedTemporaryFile (mode = mode ,delete = True )as tmp_file :
370- tmp_file .write (write_data001 .source )
371- tmp_file .flush ()
372-
373- self .operations .write (
374- tmp_file .name ,
375- write_data001 .call_param__data ,
376- read_and_write = write_data001 .call_param__rw ,
377- truncate = write_data001 .call_param__truncate ,
378- binary = write_data001 .call_param__binary )
379-
380- tmp_file .seek (0 )
381-
382- s = tmp_file .read ()
383-
384- assert s == write_data001 .result