<commit_before>#!/usr/bin/python

# Copyright (c) 2016, <NAME>
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

import os
from os import urandom, R_OK, W_OK, X_OK
from unittest import defaultTestLoader, TestCase, TestSuite, TextTestRunner
from fuse import FuseOSError

from rencfs import RencFSEncrypt, RencFSDecrypt


class EncryptTest(TestCase):

    @classmethod
    def setUpClass(cls):
        cls.td = 'test1/'
        cls.tf = 'f2'
        cls.tff = cls.td + cls.tf
        cls.tl = 'l2'
        cls.tlf = cls.td + cls.tl
        cls.fs = RencFSEncrypt(cls.td, urandom(32))
        try:
            os.mkdir(cls.td)
        except OSError:
            pass
        with open(cls.tff, 'w') as f:
            f.write(' '*128)
        try:
            os.remove(cls.tlf)
        except OSError:
            pass
        os.symlink(cls.tff, cls.tlf)

    def test_access(self):
        self.assertFalse(self.fs.access(self.tf, R_OK))
        self.assertRaises(
            FuseOSError,
            self.fs.access,
            '__',
            R_OK
        )
        self.assertRaises(
            FuseOSError,
            self.fs.access,
            self.tf,
            W_OK
        )

    def test_getattr(self):
        self.assertGreater(
            self.fs.getattr(self.tf)['st_size'],
            os.lstat(self.tff).st_size
        )

    def test_readdir(self):
        self.assertGreaterEqual(
            len(list(self.fs.readdir('/', 0))),
            4
        )
        self.assertRaises(
            FuseOSError,
            lambda a, b: list(self.fs.readdir(a, b)),
            '__',
            0
        )
        self.assertRaises(
            FuseOSError,
            lambda a, b: list(self.fs.readdir(a, b)),
            self.tf,
            0
        )

    def test_readlink(self):
        self.assertEqual(
            self.fs.readlink(self.tl),
            self.tf
        )

    def test_statfs(self):
        self.assertGreaterEqual(
            self.fs.statfs('/')['f_files'],
            2
        )

    def test_utimens(self):
        self.assertRaises(
            FuseOSError,
            self.fs.utimens,
            self.tf,
            1
        )

    def test_create(self):
        self.assertRaises(
            FuseOSError,
            self.fs.create,
            '__',
            os.O_RDONLY
        )

    def test_read(self):
        fh = self.fs.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.assertGreater(len(self.fs.read(self.tf, 1024, 0, fh)), 128)
        self.assertEqual(len(self.fs.read(self.tf, 1, 1, fh)), 1)
        self.assertEqual(len(self.fs.read(self.tf, 1, 129, fh)), 1)

    def test_release(self):
        fh = self.fs.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.fs.read(self.tf, 128, 1, fh)
        self.assertTrue(fh in self.fs.keys)
        self.fs.release(self.tf, fh)
        self.assertFalse(fh in self.fs.keys)


class DecryptTest(TestCase):

    @classmethod
    def setUpClass(cls):
        cls.td = 'test1/'
        cls.tf = 'f2'
        cls.tff = cls.td + cls.tf
        cls.tl = 'l2'
        cls.tlf = cls.td + cls.tl
        with open(cls.tff, 'w') as f:
            f.write(' '*128)
        key = urandom(32)
        fs = RencFSEncrypt(cls.td, key)
        d = fs.read(cls.tf, 1024, 0, fs.open(cls.tf, os.O_RDONLY))
        f = os.open(cls.tff, os.O_WRONLY)
        os.write(f, d)
        os.close(f)
        cls.fs = RencFSDecrypt(cls.td, key)
        cls.fs2 = RencFSDecrypt(cls.td, key[16:] + key[:16])
        cls.fs3 = RencFSDecrypt(cls.td, key[16:] + key[:16], False)

    def test_getattr(self):
        self.assertLess(
            self.fs.getattr(self.tf)['st_size'],
            os.lstat(self.tff).st_size
        )
        self.assertEqual(
            self.fs.getattr(self.tf)['st_size'],
            128
        )

    def test_read(self):
        fh = self.fs.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.assertEqual(self.fs.read(self.tf, 1024, 0, fh), b' '*128)
        self.assertEqual(self.fs.read(self.tf, 1, 1, fh), b' ')
        self.assertEqual(len(self.fs.read(self.tf, 1, 129, fh)), 0)

    def test_failure(self):
        self.assertRaises(
            FuseOSError,
            self.fs2.open,
            self.tf,
            os.O_RDONLY
        )

    def test_noverify(self):
        fh = self.fs3.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.assertNotEqual(self.fs3.read(self.tf, 1024, 0, fh), b' '*128)


def main():
    tests = TestSuite()
    tests.addTest(defaultTestLoader.loadTestsFromTestCase(EncryptTest))
    tests.addTest(defaultTestLoader.loadTestsFromTestCase(DecryptTest))
    TextTestRunner().run(tests)


if __name__ == '__main__':
    main()<commit_msg>Make tests runnable in either order<commit_after>#!/usr/bin/python

# Copyright (c) 2016, <NAME>
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

import os
from os import urandom, R_OK, W_OK, X_OK
from unittest import defaultTestLoader, TestCase, TestSuite, TextTestRunner
from fuse import FuseOSError

from rencfs import RencFSEncrypt, RencFSDecrypt


class EncryptTest(TestCase):

    @classmethod
    def setUpClass(cls):
        testdir = 'test1/'
        cls.tf = 'f2'
        cls.tff = testdir + cls.tf
        cls.tl = 'l2'
        tlf = testdir + cls.tl
        cls.fs = RencFSEncrypt(testdir, urandom(32))
        try:
            os.mkdir(testdir)
        except OSError:
            pass
        with open(cls.tff, 'w') as f:
            f.write(' '*128)
        try:
            os.remove(tlf)
        except OSError:
            pass
        os.symlink(cls.tff, tlf)

    def test_access(self):
        self.assertFalse(self.fs.access(self.tf, R_OK))
        self.assertRaises(
            FuseOSError,
            self.fs.access,
            '__',
            R_OK
        )
        self.assertRaises(
            FuseOSError,
            self.fs.access,
            self.tf,
            W_OK
        )

    def test_getattr(self):
        self.assertGreater(
            self.fs.getattr(self.tf)['st_size'],
            os.lstat(self.tff).st_size
        )

    def test_readdir(self):
        self.assertGreaterEqual(
            len(list(self.fs.readdir('/', 0))),
            4
        )
        self.assertRaises(
            FuseOSError,
            lambda a, b: list(self.fs.readdir(a, b)),
            '__',
            0
        )
        self.assertRaises(
            FuseOSError,
            lambda a, b: list(self.fs.readdir(a, b)),
            self.tf,
            0
        )

    def test_readlink(self):
        self.assertEqual(
            self.fs.readlink(self.tl),
            self.tf
        )

    def test_statfs(self):
        self.assertGreaterEqual(
            self.fs.statfs('/')['f_files'],
            2
        )

    def test_utimens(self):
        self.assertRaises(
            FuseOSError,
            self.fs.utimens,
            self.tf,
            1
        )

    def test_create(self):
        self.assertRaises(
            FuseOSError,
            self.fs.create,
            '__',
            os.O_RDONLY
        )

    def test_read(self):
        fh = self.fs.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.assertGreater(len(self.fs.read(self.tf, 1024, 0, fh)), 128)
        self.assertEqual(len(self.fs.read(self.tf, 1, 1, fh)), 1)
        self.assertEqual(len(self.fs.read(self.tf, 1, 129, fh)), 1)

    def test_release(self):
        fh = self.fs.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.fs.read(self.tf, 128, 1, fh)
        self.assertTrue(fh in self.fs.keys)
        self.fs.release(self.tf, fh)
        self.assertFalse(fh in self.fs.keys)


class DecryptTest(TestCase):

    @classmethod
    def setUpClass(cls):
        testdir = 'test1/'
        cls.tf = 'f2'
        cls.tff = testdir + cls.tf
        try:
            os.mkdir(testdir)
        except OSError:
            pass
        with open(cls.tff, 'w') as f:
            f.write(' '*128)
        key = urandom(32)
        fs = RencFSEncrypt(testdir, key)
        d = fs.read(cls.tf, 1024, 0, fs.open(cls.tf, os.O_RDONLY))
        f = os.open(cls.tff, os.O_WRONLY)
        os.write(f, d)
        os.close(f)
        cls.fs = RencFSDecrypt(testdir, key)
        cls.fs2 = RencFSDecrypt(testdir, key[16:] + key[:16])
        cls.fs3 = RencFSDecrypt(testdir, key[16:] + key[:16], False)

    def test_getattr(self):
        self.assertLess(
            self.fs.getattr(self.tf)['st_size'],
            os.lstat(self.tff).st_size
        )
        self.assertEqual(
            self.fs.getattr(self.tf)['st_size'],
            128
        )

    def test_read(self):
        fh = self.fs.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.assertEqual(self.fs.read(self.tf, 1024, 0, fh), b' '*128)
        self.assertEqual(self.fs.read(self.tf, 1, 1, fh), b' ')
        self.assertEqual(len(self.fs.read(self.tf, 1, 129, fh)), 0)

    def test_failure(self):
        self.assertRaises(
            FuseOSError,
            self.fs2.open,
            self.tf,
            os.O_RDONLY
        )

    def test_noverify(self):
        fh = self.fs3.open(self.tf, os.O_RDONLY)
        self.assertTrue(fh)
        self.assertNotEqual(self.fs3.read(self.tf, 1024, 0, fh), b' '*128)


def main():
    tests = TestSuite()
    tests.addTest(defaultTestLoader.loadTestsFromTestCase(DecryptTest))
    tests.addTest(defaultTestLoader.loadTestsFromTestCase(EncryptTest))
    TextTestRunner().run(tests)


if __name__ == '__main__':
    main()
{'fraction_non_alphanumeric': 0.11714102681431317, 'fraction_numerical': 0.016381440468564108, 'mean_word_length': 1.8848996832101372, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, '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': '3253318', 'n_tokens_mistral': 3912, 'n_tokens_neox': 3649, 'n_words': 796}