<commit_before>import os
import unittest

from remoteappmanager.db.csv_db import (
    CSVApplication, CSVApplicationPolicy, CSVUser, CSVAccounting)

from tests.db.abc_test_interfaces import ABCTestDatabaseInterface
from tests.temp_mixin import TempMixin


def write_csv_file(file_name):
    _HEADERS = ('user.name', 'application.image', 'policy.allow_home',
                'policy.allow_view', 'policy.allow_common',
                'policy.volume_source', 'policy.volume_target',
                'policy.volume_mode')

    #    name,     image, home, view, common, source, target, mode
    _RECORDS = [
        ('foo', 'image_1', '1', '1', '0', '', '', ''),
        ('foo', 'image_2', '1', '1', '1', '/src', '/target', 'ro'),
        ('bar', 'image_1', '0', '0', '0', '/src', '/target', 'ro')]

    with open(file_name, 'w') as csv_file:
        print(*_HEADERS, sep=',', file=csv_file)
        for record in _RECORDS:
            print(*record, sep=',', file=csv_file)


class TestCSVAccounting(TempMixin, ABCTestDatabaseInterface,
                        unittest.TestCase):
    def setUp(self):
        # Setup temporary directory
        super().setUp()

        self.csv_file = os.path.join(self.tempdir, 'testing.csv')

        write_csv_file(self.csv_file)

        self.addTypeEqualityFunc(CSVApplication, self.assertApplicationEqual)
        self.addTypeEqualityFunc(CSVApplicationPolicy,
                                 self.assertApplicationPolicyEqual)

    def create_expected_users(self):
        return (CSVUser('foo'), CSVUser('bar'))

    def create_expected_configs(self, user):
        mappings = {
            'foo': (
                (CSVApplication(image='image_1'),
                 CSVApplicationPolicy(allow_home=True,
                                      allow_view=True,
                                      allow_common=False,
                                      volume_source=None,
                                      volume_target=None,
                                      volume_mode=None)),
                (CSVApplication(image='image_2'),
                 CSVApplicationPolicy(allow_home=True,
                                      allow_view=True,
                                      allow_common=True,
                                      volume_source='/src',
                                      volume_target='/target',
                                      volume_mode='ro'))
                ),
            'bar': (
                (CSVApplication(image='image_1'),
                 CSVApplicationPolicy(allow_home=False,
                                      allow_view=False,
                                      allow_common=False,
                                      volume_source='/src',
                                      volume_target='/target',
                                      volume_mode='ro')),
                )}
        return mappings[user.name]

    def create_accounting(self):
        return CSVAccounting(self.csv_file)

    def test_get_user_by_name(self):
        accounting = self.create_accounting()

        user = accounting.get_user_by_name('foo')
        self.assertIsInstance(user, CSVUser)
        self.assertEqual(user.name, 'foo')

        user = accounting.get_user_by_name('unknown')
        self.assertIsNone(user)<commit_msg>Added tests for loading badly formatted csv and good csv with different header ordering<commit_after>import os
import unittest

from remoteappmanager.db.csv_db import (
    CSVApplication, CSVApplicationPolicy, CSVUser, CSVAccounting)

from tests.db.abc_test_interfaces import ABCTestDatabaseInterface
from tests.temp_mixin import TempMixin


class GoodTable:
    headers = ('user.name', 'application.image', 'policy.allow_home',
               'policy.allow_view', 'policy.allow_common',
               'policy.volume_source', 'policy.volume_target',
               'policy.volume_mode')

    #    name,     image, home, view, common, source, target, mode
    records = [
        ('foo', 'image_1', '1', '1', '0', '', '', ''),
        ('foo', 'image_2', '1', '1', '1', '/src', '/target', 'ro'),
        ('bar', 'image_1', '0', '0', '0', '/src', '/target', 'ro')]


class BadTableMissingHeaders:
    # policy.volume_source and policy.volume_target are missing
    headers = ('user.name', 'application.image', 'policy.allow_home',
               'policy.allow_view', 'policy.allow_common',
               'policy.volume_mode')

    #    name,     image, home, view, common, mode
    records = [
        ('foo', 'image_1', '1', '1', '0', ''),
        ('foo', 'image_2', '1', '1', '1', 'ro'),
        ('bar', 'image_1', '0', '0', '0', 'ro')]


class GoodTableWithDifferentHeaders:
    # There is an extra column and the columns order are different
    # but that's okay
    headers = ('policy.allow_view', 'policy.allow_common',
               'policy.volume_source', 'policy.volume_target',
               'policy.volume_mode',
               'user.name', 'application.image', 'policy.allow_home',
               'extra_column')

    #  view, common, source, target, mode, name, image, home, extra
    records = [
        ('1', '0', '', '', '', 'foo', 'image_1', '1', 'anything'),
        ('1', '1', '/src', '/target', 'ro', 'foo', 'image_2', '1', 'extra'),
        ('0', '0', '/src', '/target', 'ro', 'bar', 'image_1', '0', 'whatever')]


def write_csv_file(file_name, headers, records):
    with open(file_name, 'w') as csv_file:
        print(*headers, sep=',', file=csv_file)
        for record in records:
            print(*record, sep=',', file=csv_file)


class TestCSVAccounting(TempMixin, ABCTestDatabaseInterface,
                        unittest.TestCase):
    def setUp(self):
        # Setup temporary directory
        super().setUp()

        self.csv_file = os.path.join(self.tempdir, 'testing.csv')

        write_csv_file(self.csv_file, GoodTable.headers, GoodTable.records)

        self.addTypeEqualityFunc(CSVApplication, self.assertApplicationEqual)
        self.addTypeEqualityFunc(CSVApplicationPolicy,
                                 self.assertApplicationPolicyEqual)

    def create_expected_users(self):
        return (CSVUser('foo'), CSVUser('bar'))

    def create_expected_configs(self, user):
        mappings = {
            'foo': (
                (CSVApplication(image='image_1'),
                 CSVApplicationPolicy(allow_home=True,
                                      allow_view=True,
                                      allow_common=False,
                                      volume_source=None,
                                      volume_target=None,
                                      volume_mode=None)),
                (CSVApplication(image='image_2'),
                 CSVApplicationPolicy(allow_home=True,
                                      allow_view=True,
                                      allow_common=True,
                                      volume_source='/src',
                                      volume_target='/target',
                                      volume_mode='ro'))
                ),
            'bar': (
                (CSVApplication(image='image_1'),
                 CSVApplicationPolicy(allow_home=False,
                                      allow_view=False,
                                      allow_common=False,
                                      volume_source='/src',
                                      volume_target='/target',
                                      volume_mode='ro')),
                )}
        return mappings[user.name]

    def create_accounting(self):
        return CSVAccounting(self.csv_file)

    def test_get_user_by_name(self):
        accounting = self.create_accounting()

        user = accounting.get_user_by_name('foo')
        self.assertIsInstance(user, CSVUser)
        self.assertEqual(user.name, 'foo')

        user = accounting.get_user_by_name('unknown')
        self.assertIsNone(user)

    def test_error_with_missing_headers(self):
        write_csv_file(self.csv_file,
                       BadTableMissingHeaders.headers,
                       BadTableMissingHeaders.records)

        with self.assertRaisesRegex(ValueError, "Missing headers"):
            self.create_accounting()

    def test_load_csv_file_with_good_headers_different_order(self):
        """ Test loading a good table with headers of different orders """
        write_csv_file(self.csv_file,
                       GoodTableWithDifferentHeaders.headers,
                       GoodTableWithDifferentHeaders.records)
        self.create_accounting()
{'fraction_non_alphanumeric': 0.1177085754125029, 'fraction_numerical': 0.006274692075296305, 'mean_word_length': 1.7863386209129168, 'pattern_counts': {'":': 0, '<': 3, '<?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': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '480259', 'n_tokens_mistral': 2377, 'n_tokens_neox': 2268, 'n_words': 430}