6
6
from test .support import warnings_helper
7
7
from test .support .script_helper import assert_python_ok
8
8
9
- # Skip these tests if there is no posix module.
10
- posix = import_helper .import_module ('posix' )
11
-
12
9
import errno
13
10
import sys
14
11
import signal
22
19
import textwrap
23
20
from contextlib import contextmanager
24
21
22
+ try :
23
+ import posix
24
+ except ImportError :
25
+ import nt as posix
26
+
25
27
try :
26
28
import pwd
27
29
except ImportError :
@@ -231,6 +233,9 @@ def test_register_at_fork(self):
231
233
with self .assertRaises (TypeError , msg = "Invalid arg was allowed" ):
232
234
# Ensure a combination of valid and invalid is an error.
233
235
os .register_at_fork (before = None , after_in_parent = lambda : 3 )
236
+ with self .assertRaises (TypeError , msg = "At least one argument is required" ):
237
+ # when no arg is passed
238
+ os .register_at_fork ()
234
239
with self .assertRaises (TypeError , msg = "Invalid arg was allowed" ):
235
240
# Ensure a combination of valid and invalid is an error.
236
241
os .register_at_fork (before = lambda : None , after_in_child = '' )
@@ -630,13 +635,11 @@ def test_fstat(self):
630
635
finally :
631
636
fp .close ()
632
637
633
- # TODO: RUSTPYTHON: AssertionError: DeprecationWarning not triggered by stat
634
- @unittest .expectedFailure
635
638
def test_stat (self ):
636
639
self .assertTrue (posix .stat (os_helper .TESTFN ))
637
640
self .assertTrue (posix .stat (os .fsencode (os_helper .TESTFN )))
638
641
639
- self .assertWarnsRegex ( DeprecationWarning ,
642
+ self .assertRaisesRegex ( TypeError ,
640
643
'should be string, bytes, os.PathLike or integer, not' ,
641
644
posix .stat , bytearray (os .fsencode (os_helper .TESTFN )))
6
6D4E
42
645
self .assertRaisesRegex (TypeError ,
@@ -788,7 +791,7 @@ def check_stat(uid, gid):
788
791
self .assertRaises (TypeError , chown_func , first_param , uid , t (gid ))
789
792
check_stat (uid , gid )
790
793
791
- @os_helper . skip_unless_working_chmod
794
+ @unittest . skipUnless ( hasattr ( os , "chown" ), "requires os.chown()" )
792
795
@unittest .skipIf (support .is_emscripten , "getgid() is a stub" )
793
796
def test_chown (self ):
794
797
# raise an OSError if the file does not exist
@@ -841,15 +844,10 @@ def test_listdir_bytes(self):
841
844
# the returned strings are of type bytes.
842
845
self .assertIn (os .fsencode (os_helper .TESTFN ), posix .listdir (b'.' ))
843
846
844
- # TODO: RUSTPYTHON: AssertionError: DeprecationWarning not triggered
845
- @unittest .expectedFailure
846
847
def test_listdir_bytes_like (self ):
847
848
for cls in bytearray , memoryview :
848
- with self .assertWarns (DeprecationWarning ):
849
- names = posix .listdir (cls (b'.' ))
850
- self .assertIn (os .fsencode (os_helper .TESTFN ), names )
851
- for name in names :
852
- self .assertIs (type (name ), bytes )
849
+ with self .assertRaises (TypeError ):
850
+ posix .listdir (cls (b'.' ))
853
851
854
852
@unittest .skipUnless (posix .listdir in os .supports_fd ,
855
853
"test needs fd support for posix.listdir()" )
@@ -937,6 +935,124 @@ def test_utime(self):
937
935
posix .utime (os_helper .TESTFN , (int (now ), int (now )))
938
936
posix .utime (os_helper .TESTFN , (now , now ))
939
937
938
+ def check_chmod (self , chmod_func , target , ** kwargs ):
939
+ mode = os .stat (target ).st_mode
940
+ try :
941
+ new_mode = mode & ~ (stat .S_IWOTH | stat .S_IWGRP | stat .S_IWUSR )
942
+ chmod_func (target , new_mode , ** kwargs )
943
+ self .assertEqual (os .stat (target ).st_mode , new_mode )
944
+ if stat .S_ISREG (mode ):
945
+ try :
946
+ with open (target , 'wb+' ):
947
+ pass
948
+ except PermissionError :
949
+ pass
950
+ new_mode = mode | (stat .S_IWOTH | stat .S_IWGRP | stat .S_IWUSR )
951
+ chmod_func (target , new_mode , ** kwargs )
952
+ self .assertEqual (os .stat (target ).st_mode , new_mode )
953
+ if stat .S_ISREG (mode ):
954
+ with open (target , 'wb+' ):
955
+ pass
956
+ finally :
957
+ posix .chmod (target , mode )
958
+
959
+ @os_helper .skip_unless_working_chmod
960
+ def test_chmod_file (self ):
961
+ self .check_chmod (posix .chmod , os_helper .TESTFN )
962
+
963
+ def tempdir (self ):
964
+ target = os_helper .TESTFN + 'd'
965
+ posix .mkdir (target )
966
+ self .addCleanup (posix .rmdir , target )
967
+ return target
968
+
969
+ @os_helper .skip_unless_working_chmod
970
+ def test_chmod_dir (self ):
971
+ target = self .tempdir ()
972
+ self .check_chmod (posix .chmod , target )
973
+
974
+ @unittest .skipUnless (hasattr (posix , 'lchmod' ), 'test needs os.lchmod()' )
975
+ def test_lchmod_file (self ):
976
+ self .check_chmod (posix .lchmod , os_helper .TESTFN )
977
+ self .check_chmod (posix .chmod , os_helper .TESTFN , follow_symlinks = False )
978
+
979
+ @unittest .skipUnless (hasattr (posix , 'lchmod' ), 'test needs os.lchmod()' )
980
+ def test_lchmod_dir (self ):
981
+ target = self .tempdir ()
982
+ self .check_chmod (posix .lchmod , target )
983
+ self .check_chmod (posix .chmod , target , follow_symlinks = False )
984
+
985
+ def check_chmod_link (self , chmod_func , target , link , ** kwargs ):
986
+ target_mode = os .stat (target ).st_mode
987
+ link_mode = os .lstat (link ).st_mode
988
+ try :
989
+ new_mode = target_mode & ~ (stat .S_IWOTH | stat .S_IWGRP | stat .S_IWUSR )
990
+ chmod_func (link , new_mode , ** kwargs )
991
+ self .assertEqual (os .stat (target ).st_mode , new_mode )
992
+ self .assertEqual (os .lstat (link ).st_mode , link_mode )
993
+ new_mode = target_mode | (stat .S_IWOTH | stat .S_IWGRP | stat .S_IWUSR )
994
+ chmod_func (link , new_mode , ** kwargs )
995
+ self .assertEqual (os .stat (target ).st_mode , new_mode )
996
+ self .assertEqual (os .lstat (link ).st_mode , link_mode )
997
+ finally :
998
+ posix .chmod (target , target_mode )
999
+
1000
+ def check_lchmod_link (self , chmod_func , target , link , ** kwargs ):
1001
+ target_mode = os .stat (target ).st_mode
1002
+ link_mode = os .lstat (link ).st_mode
1003
+ new_mode = link_mode & ~ (stat .S_IWOTH | stat .S_IWGRP | stat .S_IWUSR )
1004
+ chmod_func (link , new_mode , ** kwargs )
1005
+ self .assertEqual (os .stat (target ).st_mode , target_mode )
1006
+ self .assertEqual (os .lstat (link ).st_mode , new_mode )
1007
+ new_mode = link_mode | (stat .S_IWOTH | stat .S_IWGRP | stat .S_IWUSR )
1008
+ chmod_func (link , new_mode , ** kwargs )
1009
+ self .assertEqual (os .stat (target ).st_mode , target_mode )
1010
+ self .assertEqual (os .lstat (link ).st_mode , new_mode )
1011
+
1012
+ @os_helper .skip_unless_symlink
1013
+ def test_chmod_file_symlink (self ):
1014
+ target = os_helper .TESTFN
1015
+ link = os_helper .TESTFN + '-link'
1016
+ os .symlink (target , link )
1017
+ self .addCleanup (posix .unlink , link )
1018
+ if os .name == 'nt' :
1019
+ self .check_lchmod_link (posix .chmod , target , link )
1020
+ else :
1021
+ self .check_chmod_link (posix .chmod , target , link )
1022
+ self .check_chmod_link (posix .chmod , target , link , follow_symlinks = True )
1023
+
1024
+ @os_helper .skip_unless_symlink
1025
+ def test_chmod_dir_symlink (self ):
1026
+ target = self .tempdir ()
1027
+ link = os_helper .TESTFN + '-link'
1028
+ os .symlink (target , link , target_is_directory = True )
1029
+ self .addCleanup (posix .unlink , link )
1030
+ if os .name == 'nt' :
1031
+ self .check_lchmod_link (posix .chmod , target , link )
1032
+ else :
1033
+ self .check_chmod_link (posix .chmod , target , link )
1034
+ self .check_chmod_link (posix .chmod , target , link , follow_symlinks = True )
1035
+
1036
+ @unittest .skipUnless (hasattr (posix , 'lchmod' ), 'test needs os.lchmod()' )
1037
+ @os_helper .skip_unless_symlink
1038
+ def test_lchmod_file_symlink (self ):
1039
+ target = os_helper .TESTFN
1040
+ link = os_helper .TESTFN + '-link'
1041
+ os .symlink (target , link )
1042
+ self .addCleanup (posix .unlink , link )
1043
+ self .check_lchmod_link (posix .chmod , target , link , follow_symlinks = False )
1044
+ self .check_lchmod_link (posix .lchmod , target , link )
1045
+
1046
+ @unittest .skipUnless (hasattr (posix , 'lchmod' ), 'test needs os.lchmod()' )
1047
+ @os_helper .skip_unless_symlink
1048
+ def test_lchmod_dir_symlink (self ):
1049
+ target = self .tempdir ()
1050
+ link = os_helper .TESTFN + '-link'
1051
+ os .symlink (target , link )
1052
+ self .addCleanup (posix .unlink , link )
1053
+ self .check_lchmod_link (posix .chmod , target , link , follow_symlinks = False )
1054
+ self .check_lchmod_link (posix .lchmod , target , link )
1055
+
940
1056
def _test_chflags_regular_file (self , chflags_func , target_file , ** kwargs ):
941
1057
st = os .stat (target_file )
942
1058
self .assertTrue (hasattr (st , 'st_flags' ))
@@ -1016,16 +1132,17 @@ def test_environ(self):
1016
1132
def test_putenv (self ):
1017
1133
with self .assertRaises (ValueError ):
1018
1134
os .putenv ('FRUIT\0 VEGETABLE' , 'cabbage' )
1019
- with self .assertRaises (ValueError ):
1020
- os .putenv (b'FRUIT\0 VEGETABLE' , b'cabbage' )
1021
1135
with self .assertRaises (ValueError ):
1022
1136
os .putenv ('FRUIT' , 'orange\0 VEGETABLE=cabbage' )
1023
- with self .assertRaises (ValueError ):
1024
- os .putenv (b'FRUIT' , b'orange\0 VEGETABLE=cabbage' )
1025
1137
with self .assertRaises (ValueError ):
1026
1138
os .putenv ('FRUIT=ORANGE' , 'lemon' )
1027
- with self .assertRaises (ValueError ):
1028
- os .putenv (b'FRUIT=ORANGE' , b'lemon' )
1139
+ if os .name == 'posix' :
1140
+ with self .assertRaises (ValueError ):
1141
+ os .putenv (b'FRUIT\0 VEGETABLE' , b'cabbage' )
1142
+ with self .assertRaises (ValueError ):
1143
+ os .putenv (b'FRUIT' , b'orange\0 VEGETABLE=cabbage' )
1144
+ with self .assertRaises (ValueError ):
1145
+ os .putenv (b'FRUIT=ORANGE' , b'lemon' )
1029
1146
1030
1147
@unittest .skipUnless (hasattr (posix , 'getcwd' ), 'test needs posix.getcwd()' )
1031
1148
def test_getcwd_long_pathnames (self ):
@@ -1209,12 +1326,22 @@ def test_sched_getaffinity(self):
1209
1326
@requires_sched_affinity
1210
1327
def test_sched_setaffinity (self ):
1211
1328
mask = posix .sched_getaffinity (0 )
1329
+ self .addCleanup (posix .sched_setaffinity , 0 , list (mask ))
1330
+
1212
1331
if len (mask ) > 1 :
1213
1332
# Empty masks are forbidden
1214
1333
mask .pop ()
1215
1334
posix .sched_setaffinity (0 , mask )
1216
1335
self .assertEqual (posix .sched_getaffinity (0 ), mask )
1217
- self .assertRaises (OSError , posix .sched_setaffinity , 0 , [])
1336
+
1337
+ try :
1338
+ posix .sched_setaffinity (0 , [])
1339
+ # gh-117061: On RHEL9, sched_setaffinity(0, []) does not fail
1340
+ except OSError :
1341
+ # sched_setaffinity() manual page documents EINVAL error
1342
+ # when the mask is empty.
1343
+ pass
1344
+
1218
1345
self .assertRaises (ValueError , posix .sched_setaffinity , 0 , [- 10 ])
1219
1346
self .assertRaises (ValueError , posix .sched_setaffinity , 0 , map (int , "0X" ))
1220
1347
self .assertRaises (OverflowError , posix .sched_setaffinity , 0 , [1 << 128 ])
@@ -1223,6 +1350,7 @@ def test_sched_setaffinity(self):
1223
1350
self .assertRaises (OSError , posix .sched_setaffinity , - 1 , mask )
1224
1351
1225
1352
@unittest .skipIf (support .is_wasi , "No dynamic linking on WASI" )
1353
+ @unittest .skipUnless (os .name == 'posix' , "POSIX-only test" )
1226
1354
def test_rtld_constants (self ):
1227
1355
# check presence of major RTLD_* constants
1228
1356
posix .RTLD_LAZY
@@ -1552,6 +1680,8 @@ def test_initgroups(self):
1552
1680
posix .initgroups (name , g )
1553
1681
self .assertIn (g , posix .getgroups ())
1554
1682
1683
+ # TODO: RUSTPYTHON: TypeError: Unexpected keyword argument setpgroup
1684
+ @unittest .expectedFailure
1555
1685
@unittest .skipUnless (hasattr (posix , 'setgroups' ),
1556
1686
"test needs posix.setgroups()" )
1557
1687
def test_setgroups (self ):
@@ -1659,12 +1789,6 @@ def test_resetids(self):
1659
1789
)
1660
1790
support .wait_process (pid , exitcode = 0 )
1661
1791
1662
- def test_resetids_wrong_type (self ):
1663
- with self .assertRaises (TypeError ):
1664
- self .spawn_func (sys .executable ,
1665
- [sys .executable , "-c" , "pass" ],
1666
- os .environ , resetids = None )
1667
-
1668
1792
# TODO: RUSTPYTHON: TypeError: Unexpected keyword argument setpgroup
1669
1793
@unittest .expectedFailure
1670
1794
def test_setpgroup (self ):
@@ -2216,6 +2340,53 @@ def test_utime(self):
2216
2340
os .utime ("path" , dir_fd = 0 )
2217
2341
2218
2342
2343
+ class NamespacesTests (unittest .TestCase ):
2344
+ """Tests for os.unshare() and os.setns()."""
2345
+
2346
+ @unittest .skipUnless (hasattr (os , 'unshare' ), 'needs os.unshare()' )
2347
+ @unittest .skipUnless (hasattr (os , 'setns' ), 'needs os.setns()' )
2348
+ @unittest .skipUnless (os .path .exists ('/proc/self/ns/uts' ), 'need /proc/self/ns/uts' )
2349
+ @support .requires_linux_version (3 , 0 , 0 )
2350
+ def test_unshare_setns (self ):
2351
+ code = """if 1:
2352
+ import errno
2353
+ import os
2354
+ import sys
2355
+ fd = os.open('/proc/self/ns/uts', os.O_RDONLY)
2356
+ try:
2357
+ original = os.readlink('/proc/self/ns/uts')
2358
+ try:
2359
+ os.unshare(os.CLONE_NEWUTS)
2360
+ except OSError as e:
2361
+ if e.errno == errno.ENOSPC:
2362
+ # skip test if limit is exceeded
2363
+ sys.exit()
2364
+ raise
2365
+ new = os.readlink('/proc/self/ns/uts')
2366
+ if original == new:
2367
+ raise Exception('os.unshare failed')
2368
+ os.setns(fd, os.CLONE_NEWUTS)
2369
+ restored = os.readlink('/proc/self/ns/uts')
2370
+ if original != restored:
2371
+ raise Exception('os.setns failed')
2372
+ except PermissionError:
2373
+ # The calling process did not have the required privileges
2374
+ # for this operation
2375
+ pass
2376
+ except OSError as e:
2377
+ # Skip the test on these errors:
2378
+ # - ENOSYS: syscall not available
2379
+ # - EINVAL: kernel was not configured with the CONFIG_UTS_NS option
2380
+ # - ENOMEM: not enough memory
2381
+ if e.errno not in (errno.ENOSYS, errno.EINVAL, errno.ENOMEM):
2382
+ raise
2383
+ finally:
2384
+ os.close(fd)
2385
+ """
2386
+
2387
+ assert_python_ok ("-c" , code )
2388
+
2389
+
2219
2390
def tearDownModule ():
2220
2391
support .reap_children ()
2221
2392
0 commit comments