<commit_before>import unittest
from unittest import mock
from collections import namedtuple
import ctypes
import ctypes.util
import platform
import requests
import json
import vcr
import azure.storage.file as file
from azure.storage.file import models
import os


# USEFUL LINKS FOR THESE TESTS:
#     ERRNO http://man7.org/linux/man-pages/man3/errno.3.html
#     requests status codes:https://github.com/kennethreitz/requests/blob/5524472cc76ea00d64181505f1fbb7f93f11cc2b/requests/status_codes.py
#     http codes: https://en.wikipedia.org/wiki/List_of_HTTP_status_codes

# mock out finding the fuse library and loading it so we can load the driver.
# We don't actually use this bit anyway in testing.
# TODO: figure out how to have the mock only happen for find_library('fuse')
# mock should only occur for CDLL(_libfuse_path).  libfuse_path will be 'fake'
original_find_library = ctypes.util.find_library
original_cdll = ctypes.CDLL
def find_library_mock(value):
    if value == 'fuse':
        return 'fake'
    return original_find_library(value)

Contents = namedtuple('Contents', 'uid gid pid')

class fake_fuse_context():
    def __init__(self):
        pass
    contents = Contents(uid=123, gid=123, pid=123)

class fake_fuse_obj(object):
    def __init__(self):
        setattr(self, 'fuse_get_context', fake_fuse_context)
        setattr(self.fuse_get_context, 'restype', lambda: None)

def cdll_mock(*args, **kwargs):
    if args[0] is 'fake':
        return fake_fuse_obj()
    return original_cdll(*args, **kwargs)
with mock.patch.object(ctypes.util, 'find_library', side_effect=find_library_mock):
    with mock.patch('ctypes.CDLL',  side_effect=cdll_mock) as ma:
        with mock.patch.object(platform, 'system', side_effect=['Linux', 'Windows']):
            #cdll.side_effect = cdll_mock
            import azfilesfuse


class Test_azfilesfuse(unittest.TestCase):
    STORAGE_ACCOUNT_NAME='crwilcoxmsftplayground'
    STORAGE_ACCOUNT_SHARE='fusetests'
    STORAGE_ACCOUNT_SAS_TOKEN=None

    def setUp(self):
        # TODO: Verify Settings provided
        env_name = os.environ.get("azfilesfuse_test_accountname", None)
        if env_name is not None:
            self.STORAGE_ACCOUNT_NAME = env_name
        env_share = os.environ.get("azfilesfuse_test_accountshare", None)
        if env_share is not None:
            self.STORAGE_ACCOUNT_SHARE = env_share
        env_sas_token = os.environ.get("azfilesfuse_test_accountsastoken", None)
        if env_sas_token is not None:
            self.STORAGE_ACCOUNT_SAS_TOKEN = env_sas_token
        
        if self.STORAGE_ACCOUNT_NAME is None or self.STORAGE_ACCOUNT_SHARE is None or self.STORAGE_ACCOUNT_SAS_TOKEN is None:
            raise Exception("Environment variables necessary for running tests are not all set.")
        
        # use the azure files sdk to verify before starting our tests the share is empty.
        self.azure_fs = file.FileService(self.STORAGE_ACCOUNT_NAME, sas_token=self.STORAGE_ACCOUNT_SAS_TOKEN)

        self.delete_files_and_directories_from_share()

        # need to import this after mocking up some of its components, or else
        # this will need to create the fuse class to be tested.
        # we also want to mock out things.  though we may do that per test.
        self.fuse_driver = azfilesfuse.AzureFiles(
            self.STORAGE_ACCOUNT_NAME, self.STORAGE_ACCOUNT_SHARE, self.STORAGE_ACCOUNT_SAS_TOKEN)
    
    def delete_files_and_directories_from_share(self, dirpath=''):
        dirpath = dirpath.lstrip('/')
        for f in self.azure_fs.list_directories_and_files(self.STORAGE_ACCOUNT_SHARE, dirpath):
            if type(f) is models.File:
                    path = dirpath + "/" + f.name.lstrip('/')
                    path = path.lstrip('/')
                    directory = os.path.dirname(path)
                    filename = os.path.basename(path)
                    self.azure_fs.delete_file(self.STORAGE_ACCOUNT_SHARE, directory, filename)
                    
            if type(f) is models.Directory:
                self.delete_files_and_directories_from_share(dirpath + "/" + f.name)
        if dirpath != '':
            self.azure_fs.delete_directory(self.STORAGE_ACCOUNT_SHARE, dirpath)

    #region Tests
    def test_create(self):
        self.fuse_driver.create('file.txt', 644)
        self.assertEqual(
            self.fuse_driver.readdir('.', None),
            ['.', '..', 'file.txt'])
        
        # TODO: failing but works fine under debugger
        #self.fuse_driver.create('file2.txt', 644)
        #self.assertEqual(
        #    self.fuse_driver.readdir('.', None),
        #    ['.', '..', 'file.txt', 'file2.txt'])

    
    def test_getattr(self,):
        self.azure_fs.create_file_from_bytes(self.STORAGE_ACCOUNT_SHARE, '', 'file.txt', b'test file content')
        self.azure_fs.create_directory(self.STORAGE_ACCOUNT_SHARE, 'dir')

        # {'st_ctime': 1504203454.0, 'st_gid': 123, 'st_mode': 33188, 
        #  'st_mtime': 1504203454.0, 'st_nlink': 1, 'st_size': 17, 'st_uid': 123}
        import time
        t = time.time()
        
        file_attr = self.fuse_driver.getattr('file.txt')
        self.assertTrue(abs(file_attr['st_ctime']-t) < 1)
        self.assertTrue(abs(file_attr['st_mtime']-t) < 1)
        del(file_attr['st_ctime'])
        del(file_attr['st_mtime'])
        self.assertDictEqual(file_attr, {'st_gid': 123, 'st_mode': 33188, 'st_nlink': 1, 'st_size': 17, 'st_uid': 123})

        # {'st_ctime': 1504203454.0, 'st_gid': 123, 'st_mode': 16877, 
        #  'st_mtime': 1504203454.0, 'st_nlink': 2, 'st_uid': 123}
        dir_attr = self.fuse_driver.getattr('dir')
        self.assertTrue(abs(dir_attr['st_ctime']-t) < 1)
        self.assertTrue(abs(dir_attr['st_mtime']-t) < 1)
        del(dir_attr['st_ctime'])
        del(dir_attr['st_mtime'])
        self.assertDictEqual(dir_attr, {'st_gid': 123, 'st_mode': 16877, 'st_nlink': 2, 'st_uid': 123})

    def test_mkdir(self):
        # dir succeeds
        res = self.fuse_driver.mkdir('random_dir', None)
        self.assertEqual(0, res)

        # test different error codes and that they result in exceptions
        # verify a failure related to the specified directory existing already
        with self.assertRaisesRegex(Exception, '\[Errno 17\] File exists'):
            res = self.fuse_driver.mkdir('random_dir', None)

        # verify trying to make subdir in non existent
        with self.assertRaisesRegex(Exception, '\[Errno 2\] No such file or directory'):
            res = self.fuse_driver.mkdir('dir_not_exist/badrequest', None)

    def test_open(self):
        fd = self.fuse_driver.open('file.txt', 'r')
        self.assertEqual(fd, 0)

    def test_read(self):
        self.azure_fs.create_file_from_bytes(self.STORAGE_ACCOUNT_SHARE, '', 'file.txt', b'test file content')<commit_msg>expand range for times on test<commit_after>import unittest
from unittest import mock
from collections import namedtuple
import ctypes
import ctypes.util
import platform
import requests
import json
import vcr
import azure.storage.file as file
from azure.storage.file import models
import os


# USEFUL LINKS FOR THESE TESTS:
#     ERRNO http://man7.org/linux/man-pages/man3/errno.3.html
#     requests status codes:https://github.com/kennethreitz/requests/blob/5524472cc76ea00d64181505f1fbb7f93f11cc2b/requests/status_codes.py
#     http codes: https://en.wikipedia.org/wiki/List_of_HTTP_status_codes

# mock out finding the fuse library and loading it so we can load the driver.
# We don't actually use this bit anyway in testing.
# TODO: figure out how to have the mock only happen for find_library('fuse')
# mock should only occur for CDLL(_libfuse_path).  libfuse_path will be 'fake'
original_find_library = ctypes.util.find_library
original_cdll = ctypes.CDLL
def find_library_mock(value):
    if value == 'fuse':
        return 'fake'
    return original_find_library(value)

Contents = namedtuple('Contents', 'uid gid pid')

class fake_fuse_context():
    def __init__(self):
        pass
    contents = Contents(uid=123, gid=123, pid=123)

class fake_fuse_obj(object):
    def __init__(self):
        setattr(self, 'fuse_get_context', fake_fuse_context)
        setattr(self.fuse_get_context, 'restype', lambda: None)

def cdll_mock(*args, **kwargs):
    if args[0] is 'fake':
        return fake_fuse_obj()
    return original_cdll(*args, **kwargs)
with mock.patch.object(ctypes.util, 'find_library', side_effect=find_library_mock):
    with mock.patch('ctypes.CDLL',  side_effect=cdll_mock) as ma:
        with mock.patch.object(platform, 'system', side_effect=['Linux', 'Windows']):
            #cdll.side_effect = cdll_mock
            import azfilesfuse


class Test_azfilesfuse(unittest.TestCase):
    STORAGE_ACCOUNT_NAME='crwilcoxmsftplayground'
    STORAGE_ACCOUNT_SHARE='fusetests'
    STORAGE_ACCOUNT_SAS_TOKEN=None

    def setUp(self):
        # TODO: Verify Settings provided
        env_name = os.environ.get("azfilesfuse_test_accountname", None)
        if env_name is not None:
            self.STORAGE_ACCOUNT_NAME = env_name
        env_share = os.environ.get("azfilesfuse_test_accountshare", None)
        if env_share is not None:
            self.STORAGE_ACCOUNT_SHARE = env_share
        env_sas_token = os.environ.get("azfilesfuse_test_accountsastoken", None)
        if env_sas_token is not None:
            self.STORAGE_ACCOUNT_SAS_TOKEN = env_sas_token
        
        if self.STORAGE_ACCOUNT_NAME is None or self.STORAGE_ACCOUNT_SHARE is None or self.STORAGE_ACCOUNT_SAS_TOKEN is None:
            raise Exception("Environment variables necessary for running tests are not all set.")
        
        # use the azure files sdk to verify before starting our tests the share is empty.
        self.azure_fs = file.FileService(self.STORAGE_ACCOUNT_NAME, sas_token=self.STORAGE_ACCOUNT_SAS_TOKEN)

        self.delete_files_and_directories_from_share()

        # need to import this after mocking up some of its components, or else
        # this will need to create the fuse class to be tested.
        # we also want to mock out things.  though we may do that per test.
        self.fuse_driver = azfilesfuse.AzureFiles(
            self.STORAGE_ACCOUNT_NAME, self.STORAGE_ACCOUNT_SHARE, self.STORAGE_ACCOUNT_SAS_TOKEN)
    
    def delete_files_and_directories_from_share(self, dirpath=''):
        dirpath = dirpath.lstrip('/')
        for f in self.azure_fs.list_directories_and_files(self.STORAGE_ACCOUNT_SHARE, dirpath):
            if type(f) is models.File:
                    path = dirpath + "/" + f.name.lstrip('/')
                    path = path.lstrip('/')
                    directory = os.path.dirname(path)
                    filename = os.path.basename(path)
                    self.azure_fs.delete_file(self.STORAGE_ACCOUNT_SHARE, directory, filename)
                    
            if type(f) is models.Directory:
                self.delete_files_and_directories_from_share(dirpath + "/" + f.name)
        if dirpath != '':
            self.azure_fs.delete_directory(self.STORAGE_ACCOUNT_SHARE, dirpath)

    #region Tests
    def test_create(self):
        self.fuse_driver.create('file.txt', 644)
        self.assertEqual(
            self.fuse_driver.readdir('.', None),
            ['.', '..', 'file.txt'])
        
        # TODO: failing but works fine under debugger
        #self.fuse_driver.create('file2.txt', 644)
        #self.assertEqual(
        #    self.fuse_driver.readdir('.', None),
        #    ['.', '..', 'file.txt', 'file2.txt'])

    
    def test_getattr(self,):
        self.azure_fs.create_file_from_bytes(self.STORAGE_ACCOUNT_SHARE, '', 'file.txt', b'test file content')
        self.azure_fs.create_directory(self.STORAGE_ACCOUNT_SHARE, 'dir')

        # {'st_ctime': 1504203454.0, 'st_gid': 123, 'st_mode': 33188, 
        #  'st_mtime': 1504203454.0, 'st_nlink': 1, 'st_size': 17, 'st_uid': 123}
        import time
        t = time.time()
        
        file_attr = self.fuse_driver.getattr('file.txt')
        self.assertTrue(abs(file_attr['st_ctime']-t) < 5, "ctime:{} walltime:{}".format(file_attr['st_ctime'], t))
        self.assertTrue(abs(file_attr['st_mtime']-t) < 5, "mtime:{} walltime:{}".format(file_attr['st_mtime'], t))
        del(file_attr['st_ctime'])
        del(file_attr['st_mtime'])
        self.assertDictEqual(file_attr, {'st_gid': 123, 'st_mode': 33188, 'st_nlink': 1, 'st_size': 17, 'st_uid': 123})

        # {'st_ctime': 1504203454.0, 'st_gid': 123, 'st_mode': 16877, 
        #  'st_mtime': 1504203454.0, 'st_nlink': 2, 'st_uid': 123}
        dir_attr = self.fuse_driver.getattr('dir')
        self.assertTrue(abs(dir_attr['st_ctime']-t) < 5, "ctime:{} walltime:{}".format(dir_attr['st_ctime'], t))
        self.assertTrue(abs(dir_attr['st_mtime']-t) < 5, "mtime:{} walltime:{}".format(dir_attr['st_mtime'], t))
        del(dir_attr['st_ctime'])
        del(dir_attr['st_mtime'])
        self.assertDictEqual(dir_attr, {'st_gid': 123, 'st_mode': 16877, 'st_nlink': 2, 'st_uid': 123})

    def test_mkdir(self):
        # dir succeeds
        res = self.fuse_driver.mkdir('random_dir', None)
        self.assertEqual(0, res)

        # test different error codes and that they result in exceptions
        # verify a failure related to the specified directory existing already
        with self.assertRaisesRegex(Exception, '\[Errno 17\] File exists'):
            res = self.fuse_driver.mkdir('random_dir', None)

        # verify trying to make subdir in non existent
        with self.assertRaisesRegex(Exception, '\[Errno 2\] No such file or directory'):
            res = self.fuse_driver.mkdir('dir_not_exist/badrequest', None)

    def test_open(self):
        fd = self.fuse_driver.open('file.txt', 'r')
        self.assertEqual(fd, 0)

    def test_read(self):
        self.azure_fs.create_file_from_bytes(self.STORAGE_ACCOUNT_SHARE, '', 'file.txt', b'test file content')
{'fraction_non_alphanumeric': 0.11272649144166727, 'fraction_numerical': 0.0217718255389243, 'mean_word_length': 3.0806545879602574, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3582575', 'n_tokens_mistral': 4874, 'n_tokens_neox': 4603, 'n_words': 1000}