import string import unittest from random import randint from cpl_core.utils import String from cpl_query.exceptions import InvalidTypeException, ArgumentNoneException from cpl_query.extension.list import List from cpl_query.iterable import Iterable from unittests_query.models import User, Address class IterableQueryTestCase(unittest.TestCase): def setUp(self) -> None: self._tests = List(User) self._t_user = User( 'Test user', Address( 'teststr.', 15 ) ) self._t_user2 = User( 'Test user', Address( 'teststr.', 14 ) ) self._generate_test_data() def _generate_test_data(self): for i in range(0, 100): user = User( String.random_string(string.ascii_letters, 8).lower(), Address( String.random_string(string.ascii_letters, 10).lower(), randint(1, 10) ) ) self._tests.append(user) self._tests.append(self._t_user) self._tests.append(self._t_user2) def test_any(self): results = [] for user in self._tests: if user.address.nr == 10: results.append(user) res = self._tests.any(lambda u: u.address.nr == 10) n_res = self._tests.any(lambda u: u.address.nr == 100) self.assertTrue(res) self.assertFalse(n_res) def test_all(self): results = [] for user in self._tests: if user.address.nr == 10: results.append(user) res = self._tests.all(lambda u: u.address is not None) n_res = self._tests.all(lambda u: u.address.nr == 100) self.assertTrue(res) self.assertFalse(n_res) def test_avg(self): avg = 0 for user in self._tests: avg += user.address.nr avg = avg / len(self._tests) res = self._tests.average(lambda u: u.address.nr) self.assertEqual(avg, res) def invalid(): tests = List(str, ['hello', 'world']) e_res = tests.average() self.assertRaises(InvalidTypeException, invalid) tests = List(int, list(range(0, 100))) self.assertEqual(sum(tests) / len(tests), tests.average()) def wrong2(): tests2 = List(int, values=list(range(0, 100))) e_res = tests2.average(lambda u: u.address.nr) self.assertRaises(AttributeError, wrong2) def test_contains(self): self.assertTrue(self._tests.contains(self._t_user)) self.assertFalse(self._tests.contains(User("Test", None))) def test_count(self): self.assertEqual(len(self._tests), self._tests.count()) self.assertEqual(1, self._tests.count(lambda u: u == self._t_user)) def test_distinct(self): res = self._tests.select(lambda u: u.address.nr).where(lambda a: a == 5).distinct() self.assertEqual(1, res.count()) addresses = [] for u in self._tests: if u.address.nr in addresses: continue addresses.append(u.address.nr) res2 = self._tests.distinct(lambda x: x.address.nr).select(lambda x: x.address.nr).to_list() self.assertEqual(addresses, res2) def test_element_at(self): index = randint(0, len(self._tests) - 1) self.assertEqual(self._tests[index], self._tests.element_at(index)) def test_element_at_or_default(self): index = randint(0, len(self._tests) - 1) self.assertEqual(self._tests[index], self._tests.element_at_or_default(index)) self.assertIsNone(self._tests.element_at_or_default(len(self._tests))) def test_last(self): results = [] for user in self._tests: if user.address.nr == 10: results.append(user) res = self._tests.where(lambda u: u.address.nr == 10) s_res = self._tests.where(lambda u: u.address.nr == 10).last() self.assertEqual(len(res), len(results)) self.assertEqual(res[len(res) - 1], s_res) def test_last_or_default(self): results = [] for user in self._tests: if user.address.nr == 10: results.append(user) res = self._tests.where(lambda u: u.address.nr == 10) s_res = self._tests.where(lambda u: u.address.nr == 10).last_or_default() sn_res = self._tests.where(lambda u: u.address.nr == 11).last_or_default() self.assertEqual(len(res), len(results)) self.assertEqual(res[len(res) - 1], s_res) self.assertIsNone(sn_res) def test_first(self): results = [] for user in self._tests: if user.address.nr == 10: results.append(user) res = self._tests.where(lambda u: u.address.nr == 10) s_res = self._tests.where(lambda u: u.address.nr == 10).first() self.assertEqual(len(res), len(results)) self.assertEqual(res[0], s_res) self.assertEqual(res[0], res.first()) self.assertEqual(res.first(), res.first()) def test_first_or_default(self): results = [] for user in self._tests: if user.address.nr == 10: results.append(user) res = self._tests.where(lambda u: u.address.nr == 10) s_res = self._tests.where(lambda u: u.address.nr == 10).first_or_default() sn_res = self._tests.where(lambda u: u.address.nr == 11).first_or_default() self.assertEqual(len(res), len(results)) self.assertEqual(res[0], s_res) self.assertIsNone(sn_res) def test_group_by(self): def by_adr(u): return u.address.nr t = self._tests.select(by_adr).group_by() res = self._tests.group_by(by_adr) self.assertTrue(isinstance(res.first_or_default(), Iterable)) self.assertNotEqual(self._tests.count(), res.count()) self.assertEqual(self._tests.distinct(by_adr).count(), res.count()) elements = List(int) groups = {} for x in range(0, 1000): v = randint(1, 100) if v not in groups: groups[v] = [] groups[v].append(v) elements.append(v) r1, r2 = list(groups.values()), elements.group_by().select(lambda l: l.to_list()).to_list() self.assertEqual(r1, r2) def test_for_each(self): users = [] self._tests.for_each(lambda user: ( users.append(user) ) ) self.assertEqual(len(users), len(self._tests)) def test_max(self): res = self._tests.max(lambda u: u.address.nr) self.assertEqual(res, self._t_user.address.nr) tests = List(int, list(range(0, 100))) self.assertEqual(99, tests.max()) def invalid(): tests = List(str, ['hello', 'world']) e_res = tests.average() self.assertRaises(InvalidTypeException, invalid) def test_min(self): res = self._tests.min(lambda u: u.address.nr) self.assertEqual(1, res) tests = List(int, list(range(0, 100))) self.assertEqual(0, tests.min()) def invalid(): tests = List(str, ['hello', 'world']) e_res = tests.average() self.assertRaises(InvalidTypeException, invalid) def test_order_by(self): res = self._tests.order_by(lambda user: user.address.street) res2 = self._tests.order_by(lambda user: user.address.nr).to_list() s_res = self._tests.to_list() s_res.sort(key=lambda user: user.address.street) self.assertEqual(res.to_list(), s_res) s_res = self._tests.to_list() s_res.sort(key=lambda user: user.address.nr) self.assertEqual(res2, s_res) self.assertEqual(self._t_user, res.where(lambda u: u.address.nr == self._t_user.address.nr).single()) def test_order_by_descending(self): res = self._tests.order_by_descending(lambda user: user.address.street).to_list() res2 = self._tests.order_by_descending(lambda user: user.address.nr).to_list() s_res = self._tests.to_list() s_res.sort(key=lambda user: user.address.street, reverse=True) self.assertEqual(res, s_res) s_res = self._tests.to_list() s_res.sort(key=lambda user: user.address.nr, reverse=True) self.assertEqual(res2, s_res) def test_then_by(self): res = self._tests.order_by(lambda user: user.address.street).then_by(lambda user: user.address.nr).to_list() s_res = self._tests.to_list() s_res.sort(key=lambda user: (user.address.street, user.address.nr)) self.assertEqual(res, s_res) def test_then_by_descending(self): res = self._tests.order_by_descending(lambda user: user.address.street).then_by_descending(lambda user: user.address.nr).to_list() s_res = self._tests.to_list() s_res.sort(key=lambda user: (user.address.street, user.address.nr), reverse=True) self.assertEqual(res, s_res) def test_reverse(self): res = self._tests.reverse() l_res = self._tests.to_list() l_res.reverse() self.assertEqual(res.to_list(), l_res) def test_select(self): range_list = List(int, range(0, 100)) selected_range = range_list.select(lambda x: x + 1) modulo_range = [] for x in range(0, 100): if x % 2 == 0: modulo_range.append(x) self.assertEqual(selected_range.to_list(), list(range(1, 101))) self.assertEqual(range_list.where(lambda x: x % 2 == 0).to_list(), modulo_range) def test_select_many(self): range_list = List(int, list(range(0, 100))) selected_range = range_list.select(lambda x: [x, x]) self.assertEqual(selected_range.to_list(), [[x, x] for x in range(0, 100)]) self.assertEqual(selected_range.select_many(lambda x: x).to_list(), [_x for _l in [2 * [x] for x in range(0, 100)] for _x in _l]) class TestClass: def __init__(self, i, is_sub=False): self.i = i if is_sub: return self.elements = [TestClass(x, True) for x in range(0, 10)] elements = List(TestClass, [TestClass(i) for i in range(0, 100)]) selected_elements = elements.select_many(lambda x: x.elements).select(lambda x: x.i) self.assertEqual(selected_elements.where(lambda x: x == 0).count(), 100) def test_single(self): res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr) s_res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr).single() self.assertEqual(len(res), 1) self.assertEqual(self._t_user, s_res) def test_single_or_default(self): res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr) s_res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr).single_or_default() sn_res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr + 1).single_or_default() self.assertEqual(len(res), 1) self.assertEqual(self._t_user, s_res) self.assertIsNone(sn_res) def test_skip(self): skipped = self._tests.skip(5).to_list() self.assertEqual(len(skipped), len(self._tests) - 5) self.assertEqual(skipped, self._tests[5:]) def test_skip_last(self): skipped = self._tests.skip_last(5) self.assertEqual(skipped.count(), len(self._tests) - 5) self.assertEqual(skipped.to_list(), self._tests[:-5]) self.assertEqual(skipped.last(), self._tests[:-5][len(self._tests[:-5]) - 1]) def test_sum(self) -> List['int']: res = self._tests.sum(lambda u: u.address.nr) s_res = 0 for user in self._tests: s_res += user.address.nr self.assertEqual(s_res, res) tests = List(int, list(range(0, 100))) self.assertEqual(0, tests.min()) def invalid(): tests2 = List(str, ['hello', 'world']) e_res = tests2.average() self.assertRaises(InvalidTypeException, invalid) def test_take(self): skipped = self._tests.take(5) self.assertEqual(skipped.count(), 5) self.assertEqual(skipped.to_list(), self._tests[:5]) def test_take_last(self): skipped = self._tests.take_last(5) self.assertEqual(skipped.count(), 5) self.assertEqual(skipped.to_list(), self._tests[-5:]) self.assertEqual(skipped.last(), self._tests[len(self._tests) - 1]) def test_where(self): results = [] for user in self._tests: if user.address.nr == 5: results.append(user) res = self._tests.where(lambda u: u.address.nr == 5) self.assertEqual(len(results), len(res)) def ex(): e_res = self._tests.where(None) self.assertRaises(ArgumentNoneException, ex)