<commit_before>import os
import unittest

from pykit import fsutil
from pykit import humannum
from pykit import proc
from pykit import ututil

dd = ututil.dd

# xx/pykit/fsutil/test/
this_base = os.path.dirname(__file__)

pyt = 'python2'


class TestFSUtil(unittest.TestCase):

    def test_exceptions(self):

        dd('present: ', fsutil.FSUtilError)
        dd('present: ', fsutil.NotMountPoint)

    def test_get_all_mountpoint(self):
        mps = fsutil.get_all_mountpoint()
        dd('mount points:', mps)
        self.assertIn('/', mps)

        mpsall = fsutil.get_all_mountpoint(all=True)
        dd('all mount points:', mps)

        self.assertTrue(len(mpsall) > len(mps))
        self.assertEqual(set([]), set(mps) - set(mpsall))

    def test_mountpoint(self):

        cases = (
                ('/',               '/',),
                ('/bin/ls',         '/',),
                ('/dev',            '/dev',),
                ('/dev/',           '/dev',),
                ('/dev/random',     '/dev',),
                ('/dev/inexistent', '/dev',),
        )

        for path, expected in cases:

            dd(path, ' --> ', expected)

            rst = fsutil.get_mountpoint(path)
            dd('rst: ', rst)

            self.assertEqual(expected, rst)

            # test assert_mount_point

            fsutil.assert_mountpoint(rst)
            self.assertRaises(fsutil.NotMountPoint,
                              fsutil.assert_mountpoint, path + '/aaa')

    def test_get_disk_partitions(self):

        rst = fsutil.get_disk_partitions()

        self.assertIn('/', rst)

        root = rst['/']
        for k in ('device', 'mountpoint', 'fstype', 'opts'):
            self.assertIn(k, root)

    def test_get_device(self):

        rst = fsutil.get_device('/inexistent')

        root_dev = proc.shell_script(
            'mount | grep " on / " | grep -o "[^ ]*"')[1].strip()

        self.assertEqual(root_dev, rst)

    def test_get_device_fs(self):

        rc, out, err = proc.shell_script(
            'mount | grep " on / " | grep -o "[^ ]*"')
        dd('find device on /', rc, out, err)

        dev = out.strip()
        dd('device: ', dev)
        rst = fsutil.get_device_fs(dev)
        self.assertIn(rst, ('hfs', 'xfs', 'ext2', 'ext3', 'ext4'))

    def test_get_path_fs(self):

        rst = fsutil.get_path_fs('/dev')
        dd('fs of /dev: ', rst)
        self.assertNotEqual('unknown', rst)
        self.assertTrue('dev' in rst)

        rst = fsutil.get_path_fs('/blabla')
        dd('fs of /blabla: ', rst)
        self.assertIn(rst, ('hfs', 'xfs', 'ext2', 'ext3', 'ext4'))

    def test_get_path_usage(self):

        path = '/'
        rst = fsutil.get_path_usage(path)
        dd(humannum.humannum(rst))

        # check against os.statvfs.

        st = os.statvfs(path)

        self.assertEqual(st.f_frsize * st.f_bavail, rst['available'])
        self.assertEqual(st.f_frsize * st.f_blocks, rst['total'])
        self.assertEqual(st.f_frsize * (st.f_blocks - st.f_bavail), rst['used'])
        self.assertEqual((st.f_blocks - st.f_bavail) * 100 / st.f_blocks, int(rst['percent'] * 100))

        if st.f_bfree > st.f_bavail:
            self.assertLess(rst['available'], st.f_frsize * st.f_bfree)
        else:
            dd('st.f_bfree == st.f_bavail')

        # check against df

        rc, out, err = proc.shell_script('df -m / | tail -n1')
        # Filesystem 1M-blocks   Used Available Capacity  iused    ifree %iused  Mounted on
        # /dev/disk1    475828 328021    147556    69% 84037441 37774557   69%   /
        dd('space of "/" from df')
        total_mb = out.strip().split()[1]

        # "123.8M", keep int only
        rst_total_mb = humannum.humannum(rst['total'], unit=humannum.M)
        rst_total_mb = rst_total_mb.split('.')[0].split('M')[0]

        dd('total MB from df:', total_mb, 'result total MB:', rst_total_mb)

        self.assertEqual(total_mb, rst_total_mb)
<commit_msg>fsutil: use assertAlmostEqual to tolerate disk space changing during test<commit_after>import os
import unittest

from pykit import fsutil
from pykit import humannum
from pykit import proc
from pykit import ututil

dd = ututil.dd

# xx/pykit/fsutil/test/
this_base = os.path.dirname(__file__)

pyt = 'python2'


class TestFSUtil(unittest.TestCase):

    def test_exceptions(self):

        dd('present: ', fsutil.FSUtilError)
        dd('present: ', fsutil.NotMountPoint)

    def test_get_all_mountpoint(self):
        mps = fsutil.get_all_mountpoint()
        dd('mount points:', mps)
        self.assertIn('/', mps)

        mpsall = fsutil.get_all_mountpoint(all=True)
        dd('all mount points:', mps)

        self.assertTrue(len(mpsall) > len(mps))
        self.assertEqual(set([]), set(mps) - set(mpsall))

    def test_mountpoint(self):

        cases = (
                ('/',               '/',),
                ('/bin/ls',         '/',),
                ('/dev',            '/dev',),
                ('/dev/',           '/dev',),
                ('/dev/random',     '/dev',),
                ('/dev/inexistent', '/dev',),
        )

        for path, expected in cases:

            dd(path, ' --> ', expected)

            rst = fsutil.get_mountpoint(path)
            dd('rst: ', rst)

            self.assertEqual(expected, rst)

            # test assert_mount_point

            fsutil.assert_mountpoint(rst)
            self.assertRaises(fsutil.NotMountPoint,
                              fsutil.assert_mountpoint, path + '/aaa')

    def test_get_disk_partitions(self):

        rst = fsutil.get_disk_partitions()

        self.assertIn('/', rst)

        root = rst['/']
        for k in ('device', 'mountpoint', 'fstype', 'opts'):
            self.assertIn(k, root)

    def test_get_device(self):

        rst = fsutil.get_device('/inexistent')

        # GNU "grep -o" split items into several lines
        root_dev = proc.shell_script(
            'mount | grep " on / " | grep -o "[^ ]*" | head -n1')[1].strip()

        self.assertEqual(root_dev, rst)

    def test_get_device_fs(self):

        # GNU "grep -o" split items into several lines
        rc, out, err = proc.shell_script(
            'mount | grep " on / " | grep -o "[^ ]*" | head -n1')
        dd('find device on /', rc, out, err)

        dev = out.strip()
        dd('device: ', dev)
        rst = fsutil.get_device_fs(dev)
        self.assertIn(rst, ('hfs', 'xfs', 'ext2', 'ext3', 'ext4'))

    def test_get_path_fs(self):

        rst = fsutil.get_path_fs('/dev')
        dd('fs of /dev: ', rst)
        self.assertNotEqual('unknown', rst)
        self.assertTrue('dev' in rst)

        rst = fsutil.get_path_fs('/blabla')
        dd('fs of /blabla: ', rst)
        self.assertIn(rst, ('hfs', 'xfs', 'ext2', 'ext3', 'ext4'))

    def test_get_path_usage(self):

        path = '/'
        rst = fsutil.get_path_usage(path)

        # check against os.statvfs.
        st = os.statvfs(path)

        dd(humannum.humannum(rst))

        # space is changing..

        self.assertAlmostEqual(st.f_frsize * st.f_bavail, rst['available'], delta=4*1024**2)
        self.assertAlmostEqual(st.f_frsize * st.f_blocks, rst['total'], delta=4*1024**2)
        self.assertAlmostEqual(st.f_frsize * (st.f_blocks - st.f_bavail), rst['used'], delta=4*1024**2)
        self.assertAlmostEqual((st.f_blocks - st.f_bavail) * 100 / st.f_blocks, int(rst['percent'] * 100), delta=4*1024**2)

        if st.f_bfree > st.f_bavail:
            self.assertLess(rst['available'], st.f_frsize * st.f_bfree)
        else:
            dd('st.f_bfree == st.f_bavail')

        # check against df

        rc, out, err = proc.shell_script('df -m / | tail -n1')
        # Filesystem 1M-blocks   Used Available Capacity  iused    ifree %iused  Mounted on
        # /dev/disk1    475828 328021    147556    69% 84037441 37774557   69%   /
        dd('space of "/" from df')
        total_mb = out.strip().split()[1]

        # "123.8M", keep int only
        rst_total_mb = humannum.humannum(rst['total'], unit=humannum.M)
        rst_total_mb = rst_total_mb.split('.')[0].split('M')[0]

        dd('total MB from df:', total_mb, 'result total MB:', rst_total_mb)

        self.assertAlmostEqual(int(total_mb), int(rst_total_mb), delta=4)

{'fraction_non_alphanumeric': 0.14949201741654572, 'fraction_numerical': 0.01826318335752298, 'mean_word_length': 2.0877520537714713, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '6688003', 'n_tokens_mistral': 3047, 'n_tokens_neox': 2791, 'n_words': 560}