19212723817

Committed 09 Nov 2025 06:34PM UTC coverage: 93.4% (-0.06%) from 93.462%

Build # 19212723817

Build Type

Pull #743

github

Committed by

dbrattli

Commit Message

Test fixes

Pull Request Pull Request #743: [WIP] RxPY v5

Run Details

1861 of 2033 new or added lines in 31 files covered. (91.54%)

153 existing lines in 13 files now uncovered.

25032 of 26801 relevant lines covered (93.4%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.38

/tests/test_core/test_priorityqueue.py

import unittest

from reactivex.internal import PriorityQueue


class TestItem:
    __test__ = False

    def __init__(self, value, label=None):
        self.value = value
        self.label = label

    def __str__(self):
        if self.label:
            return "%s (%s)" % (self.value, self.label)
        else:
            return "%s" % self.value

    def __repr__(self):
        return str(self)

    def __eq__(self, other):
        return self.value == other.value  # and self.label == other.label

    def __lt__(self, other):
        return self.value < other.value

    def __gt__(self, other):
        return self.value > other.value


class TestPriorityQueue(unittest.TestCase):
    def test_priorityqueue_count(self):
        assert PriorityQueue.MIN_COUNT < 0

    def test_priorityqueue_empty(self):
        """Must be empty on construction"""

        p = PriorityQueue()
        assert len(p) == 0
        assert p.items == []

        # Still empty after enqueue/dequeue
        p.enqueue(42)
        p.dequeue()
        assert len(p) == 0

    def test_priorityqueue_length(self):
        """Test that length is n after n invocations"""

        p = PriorityQueue()

        assert len(p) == 0
        for n in range(42):
            p.enqueue(n)
        assert len(p) == 42
        p.dequeue()
        assert len(p) == 41
        p.remove(10)
        assert len(p) == 40
        for n in range(len(p)):
            p.dequeue()
        assert len(p) == 0

    def test_priorityqueue_enqueue_dequeue(self):
        """Enqueue followed by dequeue should give the same result"""

        p = PriorityQueue()
        self.assertRaises(IndexError, p.dequeue)

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)

        assert [p.dequeue(), p.dequeue(), p.dequeue()] == [41, 42, 43]

    def test_priorityqueue_sort_stability(self):
        """Items with same value should be returned in the order they were
        added"""

        p = PriorityQueue()

        p.enqueue(TestItem(43, "high"))
        p.enqueue(TestItem(42, "first"))
        p.enqueue(TestItem(42, "second"))
        p.enqueue(TestItem(42, "last"))
        p.enqueue(TestItem(41, "low"))

        assert len(p) == 5

        assert p.dequeue() == TestItem(41, "low")
        assert p.dequeue() == TestItem(42, "first")
        assert p.dequeue() == TestItem(42, "second")
        assert p.dequeue() == TestItem(42, "last")
        assert p.dequeue() == TestItem(43, "high")

    def test_priorityqueue_remove(self):
        """Remove item from queue"""

        p = PriorityQueue()
        assert p.remove(42) == False

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)
        assert p.remove(42) == True
        assert [p.dequeue(), p.dequeue()] == [41, 43]

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)
        assert p.remove(41) == True
        assert [p.dequeue(), p.dequeue()] == [42, 43]

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)
        assert p.remove(43) == True
        assert [p.dequeue(), p.dequeue()] == [41, 42]

    def test_priorityqueue_peek(self):
        """Peek at first element in queue"""

        p = PriorityQueue()

        self.assertRaises(IndexError, p.peek)
        p.enqueue(42)
        assert p.peek() == 42
        p.enqueue(41)
        assert p.peek() == 41
        p.enqueue(43)
        assert p.peek() == 41

1	import unittest	1✔
2
3	from reactivex.internal import PriorityQueue	1✔
4
5
6	class TestItem:	1✔
7	__test__ = False	1✔
8
9	def __init__(self, value, label=None):	1✔
10	self.value = value	1✔
11	self.label = label	1✔
12
13	def __str__(self):	1✔
14	if self.label:	×
15	return "%s (%s)" % (self.value, self.label)	×
16	else:
17	return "%s" % self.value	×
18
19	def __repr__(self):	1✔
20	return str(self)	×
21
22	def __eq__(self, other):	1✔
23	return self.value == other.value # and self.label == other.label	1✔
24
25	def __lt__(self, other):	1✔
26	return self.value < other.value	1✔
27
28	def __gt__(self, other):	1✔
UNCOV 29	return self.value > other.value	×
30
31
32	class TestPriorityQueue(unittest.TestCase):	1✔
33	def test_priorityqueue_count(self):	1✔
34	assert PriorityQueue.MIN_COUNT < 0	1✔
35
36	def test_priorityqueue_empty(self):	1✔
37	"""Must be empty on construction"""
38
39	p = PriorityQueue()	1✔
40	assert len(p) == 0	1✔
41	assert p.items == []	1✔
42
43	# Still empty after enqueue/dequeue
44	p.enqueue(42)	1✔
45	p.dequeue()	1✔
46	assert len(p) == 0	1✔
47
48	def test_priorityqueue_length(self):	1✔
49	"""Test that length is n after n invocations"""
50
51	p = PriorityQueue()	1✔
52
53	assert len(p) == 0	1✔
54	for n in range(42):	1✔
55	p.enqueue(n)	1✔
56	assert len(p) == 42	1✔
57	p.dequeue()	1✔
58	assert len(p) == 41	1✔
59	p.remove(10)	1✔
60	assert len(p) == 40	1✔
61	for n in range(len(p)):	1✔
62	p.dequeue()	1✔
63	assert len(p) == 0	1✔
64
65	def test_priorityqueue_enqueue_dequeue(self):	1✔
66	"""Enqueue followed by dequeue should give the same result"""
67
68	p = PriorityQueue()	1✔
69	self.assertRaises(IndexError, p.dequeue)	1✔
70
71	p.enqueue(42)	1✔
72	p.enqueue(41)	1✔
73	p.enqueue(43)	1✔
74
75	assert [p.dequeue(), p.dequeue(), p.dequeue()] == [41, 42, 43]	1✔
76
77	def test_priorityqueue_sort_stability(self):	1✔
78	"""Items with same value should be returned in the order they were
79	added"""
80
81	p = PriorityQueue()	1✔
82
83	p.enqueue(TestItem(43, "high"))	1✔
84	p.enqueue(TestItem(42, "first"))	1✔
85	p.enqueue(TestItem(42, "second"))	1✔
86	p.enqueue(TestItem(42, "last"))	1✔
87	p.enqueue(TestItem(41, "low"))	1✔
88
89	assert len(p) == 5	1✔
90
91	assert p.dequeue() == TestItem(41, "low")	1✔
92	assert p.dequeue() == TestItem(42, "first")	1✔
93	assert p.dequeue() == TestItem(42, "second")	1✔
94	assert p.dequeue() == TestItem(42, "last")	1✔
95	assert p.dequeue() == TestItem(43, "high")	1✔
96
97	def test_priorityqueue_remove(self):	1✔
98	"""Remove item from queue"""
99
100	p = PriorityQueue()	1✔
101	assert p.remove(42) == False	1✔
102
103	p.enqueue(42)	1✔
104	p.enqueue(41)	1✔
105	p.enqueue(43)	1✔
106	assert p.remove(42) == True	1✔
107	assert [p.dequeue(), p.dequeue()] == [41, 43]	1✔
108
109	p.enqueue(42)	1✔
110	p.enqueue(41)	1✔
111	p.enqueue(43)	1✔
112	assert p.remove(41) == True	1✔
113	assert [p.dequeue(), p.dequeue()] == [42, 43]	1✔
114
115	p.enqueue(42)	1✔
116	p.enqueue(41)	1✔
117	p.enqueue(43)	1✔
118	assert p.remove(43) == True	1✔
119	assert [p.dequeue(), p.dequeue()] == [41, 42]	1✔
120
121	def test_priorityqueue_peek(self):	1✔
122	"""Peek at first element in queue"""
123
124	p = PriorityQueue()	1✔
125
126	self.assertRaises(IndexError, p.peek)	1✔
127	p.enqueue(42)	1✔
128	assert p.peek() == 42	1✔
129	p.enqueue(41)	1✔
130	assert p.peek() == 41	1✔
131	p.enqueue(43)	1✔
132	assert p.peek() == 41	1✔

ReactiveX / RxPY / 19212723817

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous