Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@params(*serializer_params)
def test_partial_write(self, serializer):
"""Test recovery from previous crash w/ partial write"""
q = Queue(self.path, **serializer_params[serializer])
for i in range(100):
q.put('var%d' % i)
del q
with open(os.path.join(self.path, 'q00000'), 'ab') as f:
pickle.dump('文字化け', f)
q = Queue(self.path, **serializer_params[serializer])
self.assertEqual(100, q.qsize())
for i in range(100):
self.assertEqual('var%d' % i, q.get())
q.task_done()
with self.assertRaises(Empty):
q.get_nowait()
@params(*serializer_params)
def test_put_block_and_wait(self, serializer):
"""Test block until queue is not full."""
q = Queue(self.path, maxsize=10, **serializer_params[serializer])
def consumer():
for i in range(5):
q.get()
q.task_done()
def producer():
for j in range(16):
q.put('var%d' % j)
p = Thread(target=producer)
p.start()
c = Thread(target=consumer)
@params(
("1", [[1 * COMMIT_MULTIPLIER]]),
("01", [[0, 1 * COMMIT_MULTIPLIER]]),
("0\n1", [[0], [1 * COMMIT_MULTIPLIER]]),
("1\n0\n", [[1 * COMMIT_MULTIPLIER], [0]]),
("10\n01", [[1 * COMMIT_MULTIPLIER, 0], [0, 1 * COMMIT_MULTIPLIER]]),
("\n1", [[1 * COMMIT_MULTIPLIER]]),
("", []),
("\n", []),
("5", [[5 * COMMIT_MULTIPLIER]]),
("123", [[1 * COMMIT_MULTIPLIER, 2 * COMMIT_MULTIPLIER, 3 * COMMIT_MULTIPLIER]]),
("1\n2\n3", [[1 * COMMIT_MULTIPLIER], [2 * COMMIT_MULTIPLIER], [3 * COMMIT_MULTIPLIER]]),
("12\n3", [[1 * COMMIT_MULTIPLIER, 2 * COMMIT_MULTIPLIER], [3 * COMMIT_MULTIPLIER]]),
)
def test(self, content, expected_result):
with mock.patch("github_board.open", mock.mock_open(read_data=content), create=True):
self.assertListEqual(expected_result, load_template("fake_path"))
@params(*serializer_params)
def test_random_read_write(self, serializer):
"""Test random read/write"""
q = Queue(self.path, **serializer_params[serializer])
n = 0
for i in range(1000):
if random.random() < 0.5:
if n > 0:
q.get_nowait()
q.task_done()
n -= 1
else:
with self.assertRaises(Empty):
q.get_nowait()
else:
q.put('var%d' % random.getrandbits(16))
@params(*expected_vals)
def test_getrm(self, term, expected_value):
page = self.getrm_page
page.search(term)
page.result_arrow_should_exist()
page.go_to_results()
page.headers_should_match(expected_value)
@params(*optimal_rotations.SUPPORTED_BACKENDS)
def test_optimal_rotations(self, backend):
optimal_rotations.run(backend)
@params(
('PT04H30M21S', timedelta(hours=4, minutes=30, seconds=21)),
('PT00H12M01S', timedelta(minutes=12, seconds=1)),
)
def test_parse_duration(string, expected):
expect(parse_delta(string)) == expected
@params(*pca.SUPPORTED_BACKENDS)
def test_pca(self, backend):
pca.run(backend)
@params(*_data_types)
@unittest.skip('This function takes a long time to load the serialized object. Think '
'about improving its run time.')
def test_load_barcodes_local(self, dtype):
if dtype == 'drop_seq':
return # None should be passed, not a link that does not target a file obj.
barcode_file = _barcode_pattern % dtype
barcodes = seqc.core.check_and_load_barcodes(barcode_file)
barcodes = self.assertIsInstance(barcodes, seqc.barcodes.CellBarcodes)
@params(
("failedjob", FailedJob),
("queuedjob", QueuedJob),
("dequeuedjob", DequeuedJob),
("scheduledjob", ScheduledJob))
def test_changelist(self, modelname, Model):
url = reverse("admin:pq_%s_changelist" % modelname)
response = self.client.get(url, follow = True)
self.failUnlessEqual(response.status_code, 200,
"%s != %s -> %s, url: %s" % (response.status_code, 200, repr(Model), url))