1
# Copyright 2011 Canonical Ltd. This software is licensed under the
2
# GNU Affero General Public License version 3 (see the file LICENSE).
4
"""Tests for hwdbsubmissions script."""
9
from canonical.testing.layers import LaunchpadScriptLayer
10
from lp.hardwaredb.interfaces.hwdb import HWSubmissionProcessingStatus
11
from lp.hardwaredb.scripts.hwdbsubmissions import (
12
ProcessingLoopForPendingSubmissions,
13
ProcessingLoopForReprocessingBadSubmissions,
15
from lp.testing import TestCaseWithFactory
18
class TestProcessingLoops(TestCaseWithFactory):
19
layer = LaunchpadScriptLayer
21
def _makePendingSubmissionsLoop(self):
22
"""Parameters don't matter for these tests."""
23
return ProcessingLoopForPendingSubmissions(None, None, 0, False)
25
def test_PendingSubmissions_submitted_found(self):
26
# The PendingSubmissions loop finds submitted entries.
27
submission = self.factory.makeHWSubmission(
28
status=HWSubmissionProcessingStatus.SUBMITTED)
29
loop = self._makePendingSubmissionsLoop()
30
# The sample data already contains one submission which we ignore.
31
submissions = loop.getUnprocessedSubmissions(2)
32
self.assertEqual([submission], submissions[1:])
34
def test_PendingSubmissions_processed_not_found(self):
35
# The PendingSubmissions loop ignores processed entries.
36
submission = self.factory.makeHWSubmission(
37
status=HWSubmissionProcessingStatus.PROCESSED)
38
loop = self._makePendingSubmissionsLoop()
39
# The sample data already contains one submission which we ignore.
40
submissions = loop.getUnprocessedSubmissions(2)
41
self.assertEqual([], submissions[1:])
42
self.assertNotEqual([submission], submissions)
44
def test_PendingSubmissions_invalid_not_found(self):
45
# The PendingSubmissions loop ignores invalid entries.
46
submission = self.factory.makeHWSubmission(
47
status=HWSubmissionProcessingStatus.INVALID)
48
loop = self._makePendingSubmissionsLoop()
49
# The sample data already contains one submission which we ignore.
50
submissions = loop.getUnprocessedSubmissions(2)
51
self.assertEqual([], submissions[1:])
52
self.assertNotEqual([submission], submissions)
54
def test_PendingSubmissions_respects_chunk_size(self):
55
# Only the requested number of entries are returned.
56
self.factory.makeHWSubmission(
57
status=HWSubmissionProcessingStatus.SUBMITTED)
58
self.factory.makeHWSubmission(
59
status=HWSubmissionProcessingStatus.SUBMITTED)
60
loop = self._makePendingSubmissionsLoop()
61
# The sample data already contains one submission.
62
submissions = loop.getUnprocessedSubmissions(2)
63
self.assertEqual(2, len(submissions))
65
def _makeBadSubmissionsLoop(self, start=0):
66
"""Parameters don't matter for these tests."""
67
return ProcessingLoopForReprocessingBadSubmissions(
68
start, None, None, 0, False)
70
def test_BadSubmissions_invalid_found(self):
71
# The BadSubmissions loop finds invalid entries.
72
submission = self.factory.makeHWSubmission(
73
status=HWSubmissionProcessingStatus.INVALID)
74
loop = self._makeBadSubmissionsLoop()
75
submissions = loop.getUnprocessedSubmissions(2)
76
self.assertEqual([submission], submissions)
78
def test_BadSubmissions_processed_not_found(self):
79
# The BadSubmissions loop ignores processed entries.
80
self.factory.makeHWSubmission(
81
status=HWSubmissionProcessingStatus.PROCESSED)
82
loop = self._makeBadSubmissionsLoop()
83
submissions = loop.getUnprocessedSubmissions(2)
84
self.assertEqual([], submissions)
86
def test_BadSubmissions_submitted_not_found(self):
87
# The BadSubmissions loop ignores submitted entries.
88
self.factory.makeHWSubmission(
89
status=HWSubmissionProcessingStatus.SUBMITTED)
90
loop = self._makeBadSubmissionsLoop()
91
submissions = loop.getUnprocessedSubmissions(2)
92
self.assertEqual([], submissions)
94
def test_BadSubmissions_respects_chunk_size(self):
95
# Only the requested number of entries are returned.
96
self.factory.makeHWSubmission(
97
status=HWSubmissionProcessingStatus.INVALID)
98
self.factory.makeHWSubmission(
99
status=HWSubmissionProcessingStatus.INVALID)
100
loop = self._makeBadSubmissionsLoop()
101
# The sample data already contains one submission.
102
submissions = loop.getUnprocessedSubmissions(1)
103
self.assertEqual(1, len(submissions))
105
def test_BadSubmissions_respects_start(self):
106
# It is possible to request a start id. Previous entries are ignored.
107
submission1 = self.factory.makeHWSubmission(
108
status=HWSubmissionProcessingStatus.INVALID)
109
submission2 = self.factory.makeHWSubmission(
110
status=HWSubmissionProcessingStatus.INVALID)
111
self.assertTrue(submission1.id < submission2.id)
112
loop = self._makeBadSubmissionsLoop(submission2.id)
113
# The sample data already contains one submission.
114
submissions = loop.getUnprocessedSubmissions(2)
115
self.assertEqual([submission2], submissions)
added
removed
lib/lp/hardwaredb/scripts/tests/test_hwdbsubmissions.py
