Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@unittest.skipUnless(hasattr(psutil.Process, "environ"),
"environ not available")
@unittest.skipUnless(POSIX, "posix only")
def test_weird_environ(self):
# environment variables can contain values without an equals sign
code = textwrap.dedent("""
#include
#include
char * const argv[] = {"cat", 0};
char * const envp[] = {"A=1", "X", "C=3", 0};
int main(void) {
/* Close stderr on exec so parent can wait for the execve to
* finish. */
if (fcntl(2, F_SETFD, FD_CLOEXEC) != 0)
return 0;
return execve("/bin/cat", argv, envp);
}
@unittest.skipUnless(LINUX and RLIMIT_SUPPORT,
"only available on Linux >= 2.6.36")
def test_rlimit_get(self):
import resource
p = psutil.Process(os.getpid())
names = [x for x in dir(psutil) if x.startswith('RLIMIT')]
assert names, names
for name in names:
value = getattr(psutil, name)
self.assertGreaterEqual(value, 0)
if name in dir(resource):
self.assertEqual(value, getattr(resource, name))
self.assertEqual(p.rlimit(value), resource.getrlimit(value))
else:
ret = p.rlimit(value)
self.assertEqual(len(ret), 2)
self.assertGreaterEqual(ret[0], -1)
def rlimit(self, ret, proc):
self.assertEqual(len(ret), 2)
self.assertGreaterEqual(ret[0], -1)
self.assertGreaterEqual(ret[1], -1)
def environ(self, ret, proc):
self.assertIsInstance(ret, dict)
# ===================================================================
# --- Limited user tests
# ===================================================================
@unittest.skipUnless(POSIX, "UNIX only")
@unittest.skipUnless(hasattr(os, 'getuid') and os.getuid() == 0,
"super user privileges are required")
class LimitedUserTestCase(TestProcess):
"""Repeat the previous tests by using a limited user.
Executed only on UNIX and only if the user who run the test script
is root.
"""
# the uid/gid the test suite runs under
if hasattr(os, 'getuid'):
PROCESS_UID = os.getuid()
PROCESS_GID = os.getgid()
def __init__(self, *args, **kwargs):
TestProcess.__init__(self, *args, **kwargs)
# re-define all existent test methods in order to
# ignore AccessDenied exceptions
for attr in [x for x in dir(self) if x.startswith('test')]:
@unittest.skipUnless(CCM_IS_DSE, 'DSE version unavailable')
def test_dse_67(self):
self._test_basic(Version('6.7.0'))
@unittest.skipUnless(hasattr(FakeClient, "says2squad"), "FakeClient.says2squad not available in this version of B3")
def test_squad_chat(self):
# WHEN
self.joe.says2squad("hi")
# THEN
self.assertEqual(1, self.count_chatlog_lines())
self.assert_log_line(self.get_all_chatlog_lines_from_logfile()[0], "@1 [Joe] to SQUAD:\thi")
@unittest.skipUnless(tornado_connection is not None,
'tornado is not installed')
@patch.object(socket.socket, 'settimeout')
@patch.object(socket.socket, 'connect')
def test_tornado_connection_timeout(self, connect, settimeout):
connect.side_effect = mock_timeout
with self.assertRaises(exceptions.AMQPConnectionError):
params = pika.ConnectionParameters(socket_timeout=2.0)
tornado_connection.TornadoConnection(params)
settimeout.assert_called_with(2.0)
@unittest.skipUnless(is_jython(), "don't know how to support blob")
def test_execute_type_blob(self):
return super(SqliteXerialTest, self).test_execute_type_blob()
@unittest.skipUnless(_SCRIBE_PRESENT, "Scribe Modules Not Present")
def test_buffer_ok(self):
self.setup_mock_transport()
self.setup_mock_client()
self.set_handler(self.handler_buffer)
self.mock_transport.open()
self.mock_client.Log(_MESSAGES).AndReturn(scribe.ResultCode.OK)
self.mock_transport.close()
self.mox.ReplayAll()
logging.error(_TEST_MSG)
@unittest.skipUnless(sys.version_info >= (3, 6),
"typing and annotated assignment was introduced in "
"Python 3.6")
def test_function_typing(self):
source = canonical("""
def foo(x : int) ->str:
i : str = '3'
return i
""")
target = canonical("""
def foo(x: int) -> str:
i: str = '3'
return i
""")
self.assertAstEqualsSource(ast.parse(source), target)
SELECTOR = selectors.DefaultSelector
class SelectSelectorTestCase(BaseSelectorTestCase, unittest.TestCase):
SELECTOR = selectors.SelectSelector
@unittest.skipUnless(hasattr(selectors, 'PollSelector'),
"Test needs selectors.PollSelector")
class PollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn, unittest.TestCase):
SELECTOR = getattr(selectors, 'PollSelector', None)
@unittest.skipUnless(hasattr(selectors, 'EpollSelector'),
"Test needs selectors.EpollSelector")
class EpollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn, unittest.TestCase):
SELECTOR = getattr(selectors, 'EpollSelector', None)
@unittest.skipUnless(hasattr(selectors, 'KqueueSelector'),
"Test needs selectors.KqueueSelector)")
class KqueueSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn, unittest.TestCase):
SELECTOR = getattr(selectors, 'KqueueSelector', None)
@unittest.skipUnless(hasattr(selectors, 'DevpollSelector'),
"Test needs selectors.DevpollSelector")
class DevpollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn, unittest.TestCase):