<commit_before>import unittest
import os
import re
import shutil
import time
from contextlib import contextmanager
from packaging.version import Version
from utils import create_sparse_tempfile, create_lio_device, delete_lio_device, wipe_all, fake_path, read_file, TestTags, tag_test
from bytesize import bytesize
import overrides_hack

from gi.repository import BlockDev, GLib


def _can_load_zram():
    """Test if we can load the zram module"""

    if os.system("lsmod|grep zram >/dev/null") != 0:
        # not loaded
        return True
    elif os.system("rmmod zram") == 0:
        # successfully unloaded
        return True
    else:
        # loaded and failed to unload
        return False

@contextmanager
def _track_module_load(test_case, mod_name, loaded_attr):
    setattr(test_case, loaded_attr, os.system("lsmod|grep %s > /dev/null" % mod_name) == 0)
    try:
        yield
    finally:
        setattr(test_case, loaded_attr, os.system("lsmod|grep %s > /dev/null" % mod_name) == 0)

def _wait_for_bcache_setup(bcache_dev):
    i = 0
    cache_dir = "/sys/block/%s/bcache/cache" % bcache_dev
    while not os.access(cache_dir, os.R_OK):
        time.sleep(1)
        i += 1
        if i >= 30:
            print("WARNING: Giving up waiting for bcache setup!!!")
            break

class KbdZRAMTestCase(unittest.TestCase):

    requested_plugins = BlockDev.plugin_specs_from_names(("kbd", "swap"))

    @classmethod
    def setUpClass(cls):
        if not BlockDev.is_initialized():
            BlockDev.init(cls.requested_plugins, None)
        else:
            BlockDev.reinit(cls.requested_plugins, True, None)

    def setUp(self):
        self.addCleanup(self._clean_up)
        self._loaded_zram_module = False

    def _clean_up(self):
        # make sure we unload the module if we loaded it
        if self._loaded_zram_module:
            os.system("rmmod zram")

class KbdZRAMDevicesTestCase(KbdZRAMTestCase):
    @unittest.skipUnless(_can_load_zram(), "cannot load the 'zram' module")
    @tag_test(TestTags.SLOW)
    def test_create_destroy_devices(self):
        # the easiest case
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 2]))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

        # no nstreams specified
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], None))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

        # with module pre-loaded, but unsed
        self.assertEqual(os.system("modprobe zram num_devices=2"), 0)
        time.sleep(1)
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 1]))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

        # with module pre-loaded, and devices used (as active swaps)
        self.assertEqual(os.system("modprobe zram num_devices=2"), 0)
        self.assertEqual(os.system("echo 10M > /sys/class/block/zram0/disksize"), 0)
        self.assertEqual(os.system("echo 10M > /sys/class/block/zram1/disksize"), 0)
        time.sleep(1)
        for zram_dev in ("/dev/zram0", "/dev/zram1"):
            self.assertTrue(BlockDev.swap_mkswap(zram_dev, None))
            self.assertTrue(BlockDev.swap_swapon(zram_dev, -1))
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            with self.assertRaises(GLib.GError):
                self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 1]))
            for zram_dev in ("/dev/zram0", "/dev/zram1"):
                self.assertTrue(BlockDev.swap_swapoff(zram_dev))
            self.assertEqual(os.system("rmmod zram"), 0)

            # should work just fine now
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 1]))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

    @unittest.skipUnless(_can_load_zram(), "cannot load the 'zram' module")
    @tag_test(TestTags.SLOW)
    def test_zram_add_remove_device(self):
        """Verify that it is possible to add and remove a zram device"""

        # the easiest case
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))
            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_remove_device(device))

        # no nstreams specified
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 0)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))
            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_remove_device(device))

        # create two devices
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))

            succ, device2 = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device2.startswith("/dev/zram"))

            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_remove_device(device))
            self.assertTrue(BlockDev.kbd_zram_remove_device(device2))

        # mixture of multiple devices and a single device
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 2]))
            time.sleep(5)
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))
            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(5)


class KbdZRAMStatsTestCase(KbdZRAMTestCase):
    @unittest.skipUnless(_can_load_zram(), "cannot load the 'zram' module")
    def test_zram_get_stats(self):
        """Verify that it is possible to get stats for a zram device"""

        # location of some sysfs files we use is different since linux 4.11
        kernel_version = os.uname()[2]
        if Version(kernel_version) >= Version("4.11"):
            self._zram_get_stats_new()
        else:
            self._zram_get_stats_old()

    def _zram_get_stats_new(self):
        with _track_module_load(self, "zram", "_loaded_zram_module"):<commit_msg>tests: Use BlockDev.utils_get_linux_version instead of os.uname<commit_after>import unittest
import os
import re
import shutil
import time
from contextlib import contextmanager
from packaging.version import Version
from utils import create_sparse_tempfile, create_lio_device, delete_lio_device, wipe_all, fake_path, read_file, TestTags, tag_test
from bytesize import bytesize
import overrides_hack

from gi.repository import BlockDev, GLib


def _can_load_zram():
    """Test if we can load the zram module"""

    if os.system("lsmod|grep zram >/dev/null") != 0:
        # not loaded
        return True
    elif os.system("rmmod zram") == 0:
        # successfully unloaded
        return True
    else:
        # loaded and failed to unload
        return False

@contextmanager
def _track_module_load(test_case, mod_name, loaded_attr):
    setattr(test_case, loaded_attr, os.system("lsmod|grep %s > /dev/null" % mod_name) == 0)
    try:
        yield
    finally:
        setattr(test_case, loaded_attr, os.system("lsmod|grep %s > /dev/null" % mod_name) == 0)

def _wait_for_bcache_setup(bcache_dev):
    i = 0
    cache_dir = "/sys/block/%s/bcache/cache" % bcache_dev
    while not os.access(cache_dir, os.R_OK):
        time.sleep(1)
        i += 1
        if i >= 30:
            print("WARNING: Giving up waiting for bcache setup!!!")
            break

class KbdZRAMTestCase(unittest.TestCase):

    requested_plugins = BlockDev.plugin_specs_from_names(("kbd", "swap"))

    @classmethod
    def setUpClass(cls):
        if not BlockDev.is_initialized():
            BlockDev.init(cls.requested_plugins, None)
        else:
            BlockDev.reinit(cls.requested_plugins, True, None)

    def setUp(self):
        self.addCleanup(self._clean_up)
        self._loaded_zram_module = False

    def _clean_up(self):
        # make sure we unload the module if we loaded it
        if self._loaded_zram_module:
            os.system("rmmod zram")

class KbdZRAMDevicesTestCase(KbdZRAMTestCase):
    @unittest.skipUnless(_can_load_zram(), "cannot load the 'zram' module")
    @tag_test(TestTags.SLOW)
    def test_create_destroy_devices(self):
        # the easiest case
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 2]))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

        # no nstreams specified
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], None))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

        # with module pre-loaded, but unsed
        self.assertEqual(os.system("modprobe zram num_devices=2"), 0)
        time.sleep(1)
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 1]))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

        # with module pre-loaded, and devices used (as active swaps)
        self.assertEqual(os.system("modprobe zram num_devices=2"), 0)
        self.assertEqual(os.system("echo 10M > /sys/class/block/zram0/disksize"), 0)
        self.assertEqual(os.system("echo 10M > /sys/class/block/zram1/disksize"), 0)
        time.sleep(1)
        for zram_dev in ("/dev/zram0", "/dev/zram1"):
            self.assertTrue(BlockDev.swap_mkswap(zram_dev, None))
            self.assertTrue(BlockDev.swap_swapon(zram_dev, -1))
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            with self.assertRaises(GLib.GError):
                self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 1]))
            for zram_dev in ("/dev/zram0", "/dev/zram1"):
                self.assertTrue(BlockDev.swap_swapoff(zram_dev))
            self.assertEqual(os.system("rmmod zram"), 0)

            # should work just fine now
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 1]))
            time.sleep(1)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(1)

    @unittest.skipUnless(_can_load_zram(), "cannot load the 'zram' module")
    @tag_test(TestTags.SLOW)
    def test_zram_add_remove_device(self):
        """Verify that it is possible to add and remove a zram device"""

        # the easiest case
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))
            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_remove_device(device))

        # no nstreams specified
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 0)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))
            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_remove_device(device))

        # create two devices
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))

            succ, device2 = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device2.startswith("/dev/zram"))

            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_remove_device(device))
            self.assertTrue(BlockDev.kbd_zram_remove_device(device2))

        # mixture of multiple devices and a single device
        with _track_module_load(self, "zram", "_loaded_zram_module"):
            self.assertTrue(BlockDev.kbd_zram_create_devices(2, [10 * 1024**2, 10 * 1024**2], [1, 2]))
            time.sleep(5)
            succ, device = BlockDev.kbd_zram_add_device (10 * 1024**2, 4)
            self.assertTrue(succ)
            self.assertTrue(device.startswith("/dev/zram"))
            time.sleep(5)
            self.assertTrue(BlockDev.kbd_zram_destroy_devices())
            time.sleep(5)


class KbdZRAMStatsTestCase(KbdZRAMTestCase):
    @unittest.skipUnless(_can_load_zram(), "cannot load the 'zram' module")
    def test_zram_get_stats(self):
        """Verify that it is possible to get stats for a zram device"""

        # location of some sysfs files we use is different since linux 4.11
        ver = BlockDev.utils_get_linux_version()
        if Version("%d.%d.%d" % (ver.major, ver.minor, ver.micro)) >= Version("4.11"):
            self._zram_get_stats_new()
        else:
            self._zram_get_stats_old()

    def _zram_get_stats_new(self):
        with _track_module_load(self, "zram", "_loaded_zram_module"):
{'fraction_non_alphanumeric': 0.1152327221438646, 'fraction_numerical': 0.027150916784203102, 'mean_word_length': 2.7745541655576256, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, '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': '2310889', 'n_tokens_mistral': 4871, 'n_tokens_neox': 4646, 'n_words': 925}