1:45 PM 11/12/2025 ���� JFIF    �� �        "" $(4,$&1'-=-157:::#+?D?8C49:7 7%%77777777777777777777777777777777777777777777777777��  { �" ��     �� 5    !1AQa"q�2��BR��#b�������  ��  ��   ? ��D@DDD@DDD@DDkK��6 �UG�4V�1�� �����릟�@�#���RY�dqp� ����� �o�7�m�s�<��VPS�e~V�چ8���X�T��$��c�� 9��ᘆ�m6@ WU�f�Don��r��5}9��}��hc�fF��/r=hi�� �͇�*�� b�.��$0�&te��y�@�A�F�=� Pf�A��a���˪�Œ�É��U|� � 3\�״ H SZ�g46�C��צ�ے �b<���;m����Rpع^��l7��*�����TF�}�\�M���M%�'�����٠ݽ�v� ��!-�����?�N!La��A+[`#���M����'�~oR�?��v^)��=��h����A��X�.���˃����^Ə��ܯsO"B�c>; �e�4��5�k��/CB��.  �J?��;�҈�������������������~�<�VZ�ꭼ2/)Í”jC���ע�V�G�!���!�F������\�� Kj�R�oc�h���:Þ I��1"2�q×°8��Р@ז���_C0�ր��A��lQ��@纼�!7��F�� �]�sZ B�62r�v�z~�K�7�c��5�.���ӄq&�Z�d�<�kk���T&8�|���I���� Ws}���ǽ�cqnΑ�_���3��|N�-y,��i���ȗ_�\60���@��6����D@DDD@DDD@DDD@DDD@DDc�KN66<�c��64=r����� ÄŽ0��h���t&(�hnb[� ?��^��\��â|�,�/h�\��R��5�? �0�!צ܉-����G����٬��Q�zA���1�����V��� �:R���`�$��ik��H����D4�����#dk����� h�}����7���w%�������*o8wG�LycuT�.���ܯ7��I��u^���)��/c�,s�Nq�ۺ�;�ך�YH2���.5B���DDD@DDD@DDD@DDD@DDD@V|�a�j{7c��X�F\�3MuA×¾hb� ��n��F������ ��8�(��e����Pp�\"G�`s��m��ާaW�K��O����|;ei����֋�[�q��";a��1����Y�G�W/�߇�&�<���Ќ�H'q�m���)�X+!���=�m�ۚ丷~6a^X�)���,�>#&6G���Y��{����"" """ """ """ """ ""��at\/�a�8 �yp%�lhl�n����)���i�t��B�������������?��modskinlienminh.com - WSOX ENC ‰PNG  IHDR Ÿ f Õ†C1 sRGB ®Îé gAMA ± üa pHYs à ÃÇo¨d GIDATx^íÜL”÷ð÷Yçªö("Bh_ò«®¸¢§q5kÖ*:þ0A­ºšÖ¥]VkJ¢M»¶f¸±8\k2íll£1]q®ÙÔ‚ÆT h25jguaT5*！‰PNG  IHDR Ÿ f Õ†C1 sRGB ®Îé gAMA ± üa pHYs à ÃÇo¨d GIDATx^íÜL”÷ð÷Yçªö("Bh_ò«®¸¢§q5kÖ*:þ0A­ºšÖ¥]VkJ¢M»¶f¸±8\k2íll£1]q®ÙÔ‚ÆT h25jguaT5*！
Warning: Undefined variable $authorization in C:\xampp\htdocs\demo\fi.php on line 57

Warning: Undefined variable $translation in C:\xampp\htdocs\demo\fi.php on line 118

Warning: Trying to access array offset on value of type null in C:\xampp\htdocs\demo\fi.php on line 119

Warning: file_get_contents(https://raw.githubusercontent.com/Den1xxx/Filemanager/master/languages/ru.json): Failed to open stream: HTTP request failed! HTTP/1.1 404 Not Found in C:\xampp\htdocs\demo\fi.php on line 120

Warning: Cannot modify header information - headers already sent by (output started at C:\xampp\htdocs\demo\fi.php:1) in C:\xampp\htdocs\demo\fi.php on line 247

Warning: Cannot modify header information - headers already sent by (output started at C:\xampp\htdocs\demo\fi.php:1) in C:\xampp\htdocs\demo\fi.php on line 248

Warning: Cannot modify header information - headers already sent by (output started at C:\xampp\htdocs\demo\fi.php:1) in C:\xampp\htdocs\demo\fi.php on line 249

Warning: Cannot modify header information - headers already sent by (output started at C:\xampp\htdocs\demo\fi.php:1) in C:\xampp\htdocs\demo\fi.php on line 250

Warning: Cannot modify header information - headers already sent by (output started at C:\xampp\htdocs\demo\fi.php:1) in C:\xampp\htdocs\demo\fi.php on line 251

Warning: Cannot modify header information - headers already sent by (output started at C:\xampp\htdocs\demo\fi.php:1) in C:\xampp\htdocs\demo\fi.php on line 252
""" Tests for kqueue wrapper. """ import errno import os import select import socket from test import support import time import unittest if not hasattr(select, "kqueue"): raise unittest.SkipTest("test works only on BSD") class TestKQueue(unittest.TestCase): def test_create_queue(self): kq = select.kqueue() self.assertTrue(kq.fileno() > 0, kq.fileno()) self.assertTrue(not kq.closed) kq.close() self.assertTrue(kq.closed) self.assertRaises(ValueError, kq.fileno) def test_create_event(self): from operator import lt, le, gt, ge fd = os.open(os.devnull, os.O_WRONLY) self.addCleanup(os.close, fd) ev = select.kevent(fd) other = select.kevent(1000) self.assertEqual(ev.ident, fd) self.assertEqual(ev.filter, select.KQ_FILTER_READ) self.assertEqual(ev.flags, select.KQ_EV_ADD) self.assertEqual(ev.fflags, 0) self.assertEqual(ev.data, 0) self.assertEqual(ev.udata, 0) self.assertEqual(ev, ev) self.assertNotEqual(ev, other) self.assertTrue(ev < other) self.assertTrue(other >= ev) for op in lt, le, gt, ge: self.assertRaises(TypeError, op, ev, None) self.assertRaises(TypeError, op, ev, 1) self.assertRaises(TypeError, op, ev, "ev") ev = select.kevent(fd, select.KQ_FILTER_WRITE) self.assertEqual(ev.ident, fd) self.assertEqual(ev.filter, select.KQ_FILTER_WRITE) self.assertEqual(ev.flags, select.KQ_EV_ADD) self.assertEqual(ev.fflags, 0) self.assertEqual(ev.data, 0) self.assertEqual(ev.udata, 0) self.assertEqual(ev, ev) self.assertNotEqual(ev, other) ev = select.kevent(fd, select.KQ_FILTER_WRITE, select.KQ_EV_ONESHOT) self.assertEqual(ev.ident, fd) self.assertEqual(ev.filter, select.KQ_FILTER_WRITE) self.assertEqual(ev.flags, select.KQ_EV_ONESHOT) self.assertEqual(ev.fflags, 0) self.assertEqual(ev.data, 0) self.assertEqual(ev.udata, 0) self.assertEqual(ev, ev) self.assertNotEqual(ev, other) ev = select.kevent(1, 2, 3, 4, 5, 6) self.assertEqual(ev.ident, 1) self.assertEqual(ev.filter, 2) self.assertEqual(ev.flags, 3) self.assertEqual(ev.fflags, 4) self.assertEqual(ev.data, 5) self.assertEqual(ev.udata, 6) self.assertEqual(ev, ev) self.assertNotEqual(ev, other) bignum = 0x7fff ev = select.kevent(bignum, 1, 2, 3, bignum - 1, bignum) self.assertEqual(ev.ident, bignum) self.assertEqual(ev.filter, 1) self.assertEqual(ev.flags, 2) self.assertEqual(ev.fflags, 3) self.assertEqual(ev.data, bignum - 1) self.assertEqual(ev.udata, bignum) self.assertEqual(ev, ev) self.assertNotEqual(ev, other) # Issue 11973 bignum = 0xffff ev = select.kevent(0, 1, bignum) self.assertEqual(ev.ident, 0) self.assertEqual(ev.filter, 1) self.assertEqual(ev.flags, bignum) self.assertEqual(ev.fflags, 0) self.assertEqual(ev.data, 0) self.assertEqual(ev.udata, 0) self.assertEqual(ev, ev) self.assertNotEqual(ev, other) # Issue 11973 bignum = 0xffffffff ev = select.kevent(0, 1, 2, bignum) self.assertEqual(ev.ident, 0) self.assertEqual(ev.filter, 1) self.assertEqual(ev.flags, 2) self.assertEqual(ev.fflags, bignum) self.assertEqual(ev.data, 0) self.assertEqual(ev.udata, 0) self.assertEqual(ev, ev) self.assertNotEqual(ev, other) def test_queue_event(self): serverSocket = socket.create_server(('127.0.0.1', 0)) client = socket.socket() client.setblocking(False) try: client.connect(('127.0.0.1', serverSocket.getsockname()[1])) except OSError as e: self.assertEqual(e.args[0], errno.EINPROGRESS) else: #raise AssertionError("Connect should have raised EINPROGRESS") pass # FreeBSD doesn't raise an exception here server, addr = serverSocket.accept() kq = select.kqueue() kq2 = select.kqueue.fromfd(kq.fileno()) ev = select.kevent(server.fileno(), select.KQ_FILTER_WRITE, select.KQ_EV_ADD | select.KQ_EV_ENABLE) kq.control([ev], 0) ev = select.kevent(server.fileno(), select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE) kq.control([ev], 0) ev = select.kevent(client.fileno(), select.KQ_FILTER_WRITE, select.KQ_EV_ADD | select.KQ_EV_ENABLE) kq2.control([ev], 0) ev = select.kevent(client.fileno(), select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE) kq2.control([ev], 0) events = kq.control(None, 4, 1) events = set((e.ident, e.filter) for e in events) self.assertEqual(events, set([ (client.fileno(), select.KQ_FILTER_WRITE), (server.fileno(), select.KQ_FILTER_WRITE)])) client.send(b"Hello!") server.send(b"world!!!") # We may need to call it several times for i in range(10): events = kq.control(None, 4, 1) if len(events) == 4: break time.sleep(1.0) else: self.fail('timeout waiting for event notifications') events = set((e.ident, e.filter) for e in events) self.assertEqual(events, set([ (client.fileno(), select.KQ_FILTER_WRITE), (client.fileno(), select.KQ_FILTER_READ), (server.fileno(), select.KQ_FILTER_WRITE), (server.fileno(), select.KQ_FILTER_READ)])) # Remove completely client, and server read part ev = select.kevent(client.fileno(), select.KQ_FILTER_WRITE, select.KQ_EV_DELETE) kq.control([ev], 0) ev = select.kevent(client.fileno(), select.KQ_FILTER_READ, select.KQ_EV_DELETE) kq.control([ev], 0) ev = select.kevent(server.fileno(), select.KQ_FILTER_READ, select.KQ_EV_DELETE) kq.control([ev], 0, 0) events = kq.control([], 4, 0.99) events = set((e.ident, e.filter) for e in events) self.assertEqual(events, set([ (server.fileno(), select.KQ_FILTER_WRITE)])) client.close() server.close() serverSocket.close() def testPair(self): kq = select.kqueue() a, b = socket.socketpair() a.send(b'foo') event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE) event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE) r = kq.control([event1, event2], 1, 1) self.assertTrue(r) self.assertFalse(r[0].flags & select.KQ_EV_ERROR) self.assertEqual(b.recv(r[0].data), b'foo') a.close() b.close() kq.close() def test_issue30058(self): # changelist must be an iterable kq = select.kqueue() a, b = socket.socketpair() ev = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE) kq.control([ev], 0) # not a list kq.control((ev,), 0) # __len__ is not consistent with __iter__ class BadList: def __len__(self): return 0 def __iter__(self): for i in range(100): yield ev kq.control(BadList(), 0) # doesn't have __len__ kq.control(iter([ev]), 0) a.close() b.close() kq.close() def test_close(self): open_file = open(__file__, "rb") self.addCleanup(open_file.close) fd = open_file.fileno() kqueue = select.kqueue() # test fileno() method and closed attribute self.assertIsInstance(kqueue.fileno(), int) self.assertFalse(kqueue.closed) # test close() kqueue.close() self.assertTrue(kqueue.closed) self.assertRaises(ValueError, kqueue.fileno) # close() can be called more than once kqueue.close() # operations must fail with ValueError("I/O operation on closed ...") self.assertRaises(ValueError, kqueue.control, None, 4) def test_fd_non_inheritable(self): kqueue = select.kqueue() self.addCleanup(kqueue.close) self.assertEqual(os.get_inheritable(kqueue.fileno()), False) @support.requires_fork() def test_fork(self): # gh-110395: kqueue objects must be closed after fork kqueue = select.kqueue() if (pid := os.fork()) == 0: try: self.assertTrue(kqueue.closed) with self.assertRaisesRegex(ValueError, "closed kqueue"): kqueue.fileno() except: os._exit(1) finally: os._exit(0) else: support.wait_process(pid, exitcode=0) self.assertFalse(kqueue.closed) # child done, we're still open. if __name__ == "__main__": unittest.main()