<commit_before>import argparse
import logging
import random
import testtools
import yaml

from mock import patch
from subiquity.models.blockdev import (Blockdev,
                                       blockdev_align_up,
                                       FIRST_PARTITION_OFFSET,
                                       GPT_END_RESERVE)
from subiquity.models.filesystem import FilesystemModel
from subiquity.prober import Prober
from subiquity.tests import fakes


GB = 1 << 40

class TestFilesystemModel(testtools.TestCase):
    def setUp(self):
        super(TestFilesystemModel, self).setUp()
        # don't show logging messages while testing
        logging.disable(logging.CRITICAL)
        self.make_fsm()

    # mocking the reading of the fake data saves on IO 
    @patch.object(Prober, 'get_storage')
    def make_fsm(self, _get_storage):
        _get_storage.return_value = fakes.FAKE_MACHINE_STORAGE_DATA 
        self.opts = argparse.Namespace()
        self.opts.machine_config = fakes.FAKE_MACHINE_JSON
        self.opts.dry_run = True
        self.prober = Prober(self.opts)
        self.storage = fakes.FAKE_MACHINE_STORAGE_DATA
        self.fsm = FilesystemModel(self.prober, self.opts)

    def test_filesystemmodel_init(self):
        self.assertNotEqual(self.fsm, None)
        self.assertEqual(self.fsm.info, {})
        self.assertEqual(self.fsm.devices, {})
        self.assertEqual(self.fsm.raid_devices, {})
        self.assertEqual(self.fsm.storage, {})

    def test_filesystemmodel_get_signals(self):
        self.assertEqual(sorted(self.fsm.get_signals()),
                         sorted(self.fsm.signals + self.fsm.fs_menu))

    def test_filesystemmodel_get_signal_by_name(self):
        for (name, signal, method) in self.fsm.get_signals():
            self.assertEqual(self.fsm.get_signal_by_name(name), signal)

    def test_filesystemmodel_get_menu(self):
        self.assertEqual(sorted(self.fsm.get_menu()),
                         sorted(self.fsm.fs_menu))

    def test_filesystemmodel_probe_storage(self):
        '''sd[b..i]'''
        disks = [d for d in self.storage.keys()
                 if self.storage[d]['DEVTYPE'] == 'disk' and 
                    self.storage[d]['MAJOR'] in ['8', '253']]
        self.fsm.probe_storage()
        self.assertNotEqual(self.fsm.storage, {})
        self.assertEqual(sorted(self.fsm.info.keys()),
                         sorted(disks))

    def test_filesystemmodel_get_disk(self):
        self.fsm.probe_storage()
        diskname = random.choice(list(self.fsm.info.keys()))
        disk = Blockdev(diskname,
                        self.fsm.info[diskname].serial,
                        self.fsm.info[diskname].model,
                        size=self.fsm.info[diskname].size)

        test_disk = self.fsm.get_disk(diskname)
        print(disk)
        print(test_disk)
        self.assertEqual(test_disk, disk)


class TestBlockdev(testtools.TestCase):
    def setUp(self):
        super(TestBlockdev, self).setUp()
        self.devpath = '/dev/foobar'
        self.serial = 'serial'
        self.model = 'model'
        self.parttype = 'gpt'
        self.size = 128 * GB
        self.bd = Blockdev(self.devpath, self.serial, self.model,
                           self.parttype, self.size)

    def test_blockdev_init(self):
        # verify
        self.assertNotEqual(self.bd, None)
        self.assertEqual(self.bd.available, True)
        self.assertEqual(self.bd.blocktype, 'disk')
        self.assertEqual(self.bd.devpath, self.devpath)
        self.assertEqual(self.bd.freespace, self.size)
        self.assertEqual(self.bd.model, self.model)
        self.assertEqual(self.bd.path, self.devpath)
        self.assertEqual(self.bd.percent_free, 100)
        self.assertEqual(self.bd.size, self.size)
        self.assertEqual(self.bd.usedspace, 0)
        self.assertEqual(list(self.bd.available_partitions), [])
        self.assertEqual(list(self.bd.filesystems), [])
        self.assertEqual(list(self.bd.mounts), [])<commit_msg>unittest: filesystemmodel<commit_after>import argparse
import logging
import random
import testtools
import yaml

from mock import patch
from subiquity.models.blockdev import (Blockdev,
                                       blockdev_align_up,
                                       FIRST_PARTITION_OFFSET,
                                       GPT_END_RESERVE)
from subiquity.models.filesystem import FilesystemModel
from subiquity.prober import Prober
from subiquity.tests import fakes


GB = 1 << 40

class TestFilesystemModel(testtools.TestCase):
    def setUp(self):
        super(TestFilesystemModel, self).setUp()
        # don't show logging messages while testing
        logging.disable(logging.CRITICAL)
        self.make_fsm()

    # mocking the reading of the fake data saves on IO 
    @patch.object(Prober, 'get_storage')
    def make_fsm(self, _get_storage):
        _get_storage.return_value = fakes.FAKE_MACHINE_STORAGE_DATA 
        self.opts = argparse.Namespace()
        self.opts.machine_config = fakes.FAKE_MACHINE_JSON
        self.opts.dry_run = True
        self.prober = Prober(self.opts)
        self.storage = fakes.FAKE_MACHINE_STORAGE_DATA
        self.fsm = FilesystemModel(self.prober, self.opts)

    def test_filesystemmodel_init(self):
        self.assertNotEqual(self.fsm, None)
        self.assertEqual(self.fsm.info, {})
        self.assertEqual(self.fsm.devices, {})
        self.assertEqual(self.fsm.raid_devices, {})
        self.assertEqual(self.fsm.storage, {})

    def test_filesystemmodel_get_signals(self):
        self.assertEqual(sorted(self.fsm.get_signals()),
                         sorted(self.fsm.signals + self.fsm.fs_menu))

    def test_filesystemmodel_get_signal_by_name(self):
        for (name, signal, method) in self.fsm.get_signals():
            self.assertEqual(self.fsm.get_signal_by_name(name), signal)

    def test_filesystemmodel_get_menu(self):
        self.assertEqual(sorted(self.fsm.get_menu()),
                         sorted(self.fsm.fs_menu))

    def test_filesystemmodel_probe_storage(self):
        '''sd[b..i]'''
        disks = [d for d in self.storage.keys()
                 if self.storage[d]['DEVTYPE'] == 'disk' and 
                    self.storage[d]['MAJOR'] in ['8', '253']]
        self.fsm.probe_storage()
        self.assertNotEqual(self.fsm.storage, {})
        self.assertEqual(sorted(self.fsm.info.keys()),
                         sorted(disks))

    def test_filesystemmodel_get_disk(self):
        self.fsm.probe_storage()
        diskname = random.choice(list(self.fsm.info.keys()))
        disk = Blockdev(diskname,
                        self.fsm.info[diskname].serial,
                        self.fsm.info[diskname].model,
                        size=self.fsm.info[diskname].size)

        test_disk = self.fsm.get_disk(diskname)
        print(disk)
        print(test_disk)
        self.assertEqual(test_disk, disk)

    def test_filesystemmodel_get_disk_from_partition(self):
        self.fsm.probe_storage()
        diskname = random.choice(list(self.fsm.info.keys()))
        disk = self.fsm.get_disk(diskname)
        disk.add_partition(1, disk.freespace, None, None, flag='raid')

        partpath = '{}{}'.format(disk.path, 1)
        print(partpath)
        self.assertTrue(partpath[-1], 1)
        test_disk = self.fsm.get_disk(partpath)
        print(disk)
        print(test_disk)
        self.assertEqual(test_disk, disk)

    def test_filesystemmodel_get_all_disks(self):
        self.fsm.probe_storage()
        all_disks = self.fsm.get_all_disks()
        for disk in all_disks:
            self.assertTrue(disk in self.fsm.devices.values())

    def test_filesystemmodel_get_available_disks(self):
        ''' occupy one of the probed disks and ensure
            that it's not included in the available disks
            result since it's not actually avaialable
        '''
        self.fsm.probe_storage()
        diskname = random.choice(list(self.fsm.info.keys()))
        disk = self.fsm.get_disk(diskname)
        disk.add_partition(1, disk.freespace, None, None, flag='raid')

        avail_disks = self.fsm.get_available_disks()
        self.assertLess(len(avail_disks), len(self.fsm.devices.values()))
        self.assertTrue(disk not in avail_disks)


class TestBlockdev(testtools.TestCase):
    def setUp(self):
        super(TestBlockdev, self).setUp()
        self.devpath = '/dev/foobar'
        self.serial = 'serial'
        self.model = 'model'
        self.parttype = 'gpt'
        self.size = 128 * GB
        self.bd = Blockdev(self.devpath, self.serial, self.model,
                           self.parttype, self.size)

    def test_blockdev_init(self):
        # verify
        self.assertNotEqual(self.bd, None)
        self.assertEqual(self.bd.available, True)
        self.assertEqual(self.bd.blocktype, 'disk')
        self.assertEqual(self.bd.devpath, self.devpath)
        self.assertEqual(self.bd.freespace, self.size)
        self.assertEqual(self.bd.model, self.model)
        self.assertEqual(self.bd.path, self.devpath)
        self.assertEqual(self.bd.percent_free, 100)
        self.assertEqual(self.bd.size, self.size)
        self.assertEqual(self.bd.usedspace, 0)
        self.assertEqual(list(self.bd.available_partitions), [])
        self.assertEqual(list(self.bd.filesystems), [])
        self.assertEqual(list(self.bd.mounts), [])
{'fraction_non_alphanumeric': 0.11121713316369805, 'fraction_numerical': 0.003498727735368957, 'mean_word_length': 2.4140427072023165, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '401972', 'n_tokens_mistral': 2988, 'n_tokens_neox': 2889, 'n_words': 465}