# Path: tests/base_test.py
# class SSH2TestCase(unittest.TestCase):
# 
#     @classmethod
#     def setUpClass(cls):
#         _mask = int('0600') if version_info <= (2,) else 0o600
#         os.chmod(PKEY_FILENAME, _mask)
#         cls.server = OpenSSHServer()
#         cls.server.start_server()
# 
#     @classmethod
#     def tearDownClass(cls):
#         cls.server.stop()
#         del cls.server
# 
#     def setUp(self):
#         self.host = '127.0.0.1'
#         self.port = 2222
#         self.cmd = 'echo me'
#         self.resp = u'me'
#         self.user_key = PKEY_FILENAME
#         self.user_pub_key = PUB_FILE
#         self.user = pwd.getpwuid(os.geteuid()).pw_name
#         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#         sock.connect((self.host, self.port))
#         self.sock = sock
#         self.session = Session()
#         self.session.handshake(self.sock)
# 
#     def tearDown(self):
#         self.session.disconnect()
#         self.sock.close()
# 
#     def _auth(self):
#         return self.session.userauth_publickey_fromfile(
#             self.user, self.user_key)


import os
import platform
import stat
import shutil
from sys import version_info
from unittest import skipUnless
from .base_test import SSH2TestCase
from ssh2.session import Session
from ssh2.error_codes import LIBSSH2_ERROR_EAGAIN
from ssh2.utils import wait_socket
from ssh2.sftp_handle import SFTPHandle, SFTPAttributes
from ssh2.exceptions import SFTPProtocolError, BufferTooSmallError
from ssh2.sftp import LIBSSH2_FXF_CREAT, LIBSSH2_FXF_WRITE, \
    LIBSSH2_SFTP_S_IRUSR, LIBSSH2_SFTP_S_IRGRP, LIBSSH2_SFTP_S_IWUSR, \
    LIBSSH2_SFTP_S_IROTH, LIBSSH2_SFTP_S_IXUSR, SFTP, \
    LIBSSH2_SFTP_S_IWUSR, LIBSSH2_SFTP_S_IWGRP, LIBSSH2_SFTP_S_IWOTH, \
    LIBSSH2_SFTP_ATTR_PERMISSIONS
            os.unlink(remote_filename)

    def test_sftp_setstat(self):
        self.assertEqual(self._auth(), 0)
        sftp = self.session.sftp_init()
        self.assertTrue(sftp is not None)
        test_data = b"data"
        remote_filename = os.sep.join([os.path.dirname(__file__),
                                       "remote_test_file"])
        rdonly_perms = LIBSSH2_SFTP_S_IRUSR |  LIBSSH2_SFTP_S_IRGRP | \
            LIBSSH2_SFTP_S_IROTH
        rw_perms = LIBSSH2_SFTP_S_IRUSR |  LIBSSH2_SFTP_S_IRGRP | \
            LIBSSH2_SFTP_S_IROTH | LIBSSH2_SFTP_S_IWUSR |  \
            LIBSSH2_SFTP_S_IWGRP | LIBSSH2_SFTP_S_IWOTH
        with open(remote_filename, 'wb') as fh:
            fh.write(test_data)
        _mask = 0o666
        os.chmod(remote_filename, _mask)
        attrs = sftp.stat(remote_filename)
        attrs.permissions = LIBSSH2_SFTP_S_IRUSR
        attrs.flags = LIBSSH2_SFTP_ATTR_PERMISSIONS
        try:
            self.assertEqual(sftp.setstat(remote_filename, attrs), 0)
            attrs = sftp.stat(remote_filename)
            expected = '0100400L' if version_info.major <= 2 else '0o100400'
            self.assertEqual(oct(attrs.permissions), expected)
            attrs.permissions = rdonly_perms
            attrs.flags = LIBSSH2_SFTP_ATTR_PERMISSIONS
            self.assertEqual(sftp.setstat(remote_filename, attrs), 0)
            attrs = sftp.stat(remote_filename)

# Path: tests/base_test.py
# class SSH2TestCase(unittest.TestCase):
# 
#     @classmethod
#     def setUpClass(cls):
#         _mask = int('0600') if version_info <= (2,) else 0o600
#         os.chmod(PKEY_FILENAME, _mask)
#         cls.server = OpenSSHServer()
#         cls.server.start_server()
# 
#     @classmethod
#     def tearDownClass(cls):
#         cls.server.stop()
#         del cls.server
# 
#     def setUp(self):
#         self.host = '127.0.0.1'
#         self.port = 2222
#         self.cmd = 'echo me'
#         self.resp = u'me'
#         self.user_key = PKEY_FILENAME
#         self.user_pub_key = PUB_FILE
#         self.user = pwd.getpwuid(os.geteuid()).pw_name
#         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#         sock.connect((self.host, self.port))
#         self.sock = sock
#         self.session = Session()
#         self.session.handshake(self.sock)
# 
#     def tearDown(self):
#         self.session.disconnect()
#         self.sock.close()
# 
#     def _auth(self):
#         return self.session.userauth_publickey_fromfile(
#             self.user, self.user_key)


# Path: tests/test_sftp.py
import os
import platform
import stat
import shutil
from sys import version_info
from unittest import skipUnless
from .base_test import SSH2TestCase
from ssh2.session import Session
from ssh2.error_codes import LIBSSH2_ERROR_EAGAIN
from ssh2.utils import wait_socket
from ssh2.sftp_handle import SFTPHandle, SFTPAttributes
from ssh2.exceptions import SFTPProtocolError, BufferTooSmallError
from ssh2.sftp import LIBSSH2_FXF_CREAT, LIBSSH2_FXF_WRITE, \
    LIBSSH2_SFTP_S_IRUSR, LIBSSH2_SFTP_S_IRGRP, LIBSSH2_SFTP_S_IWUSR, \
    LIBSSH2_SFTP_S_IROTH, LIBSSH2_SFTP_S_IXUSR, SFTP, \
    LIBSSH2_SFTP_S_IWUSR, LIBSSH2_SFTP_S_IWGRP, LIBSSH2_SFTP_S_IWOTH, \
    LIBSSH2_SFTP_ATTR_PERMISSIONS

            os.unlink(remote_filename)

    def test_sftp_setstat(self):
        self.assertEqual(self._auth(), 0)
        sftp = self.session.sftp_init()
        self.assertTrue(sftp is not None)
        test_data = b"data"
        remote_filename = os.sep.join([os.path.dirname(__file__),
                                       "remote_test_file"])
        rdonly_perms = LIBSSH2_SFTP_S_IRUSR |  LIBSSH2_SFTP_S_IRGRP | \
            LIBSSH2_SFTP_S_IROTH
        rw_perms = LIBSSH2_SFTP_S_IRUSR |  LIBSSH2_SFTP_S_IRGRP | \
            LIBSSH2_SFTP_S_IROTH | LIBSSH2_SFTP_S_IWUSR |  \
            LIBSSH2_SFTP_S_IWGRP | LIBSSH2_SFTP_S_IWOTH
        with open(remote_filename, 'wb') as fh:
            fh.write(test_data)
        _mask = 0o666
        os.chmod(remote_filename, _mask)
        attrs = sftp.stat(remote_filename)
        attrs.permissions = LIBSSH2_SFTP_S_IRUSR
        attrs.flags = LIBSSH2_SFTP_ATTR_PERMISSIONS
        try:
            self.assertEqual(sftp.setstat(remote_filename, attrs), 0)
            attrs = sftp.stat(remote_filename)
            expected = '0100400L' if version_info.major <= 2 else '0o100400'
            self.assertEqual(oct(attrs.permissions), expected)
            attrs.permissions = rdonly_perms
            attrs.flags = LIBSSH2_SFTP_ATTR_PERMISSIONS
            self.assertEqual(sftp.setstat(remote_filename, attrs), 0)
            attrs = sftp.stat(remote_filename)

            expected = '0100444L' if version_info.major <= 2 else '0o100444'