Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_multiplexer_filter.py
| Show First 20 Lines • Show All 69 Lines • ▼ Show 20 Lines | class MixinTestReadFilter(unittest.TestCase): | ||||
| def test_can_check(self): | def test_can_check(self): | ||||
| with self.assertRaises(ObjNotFoundError): | with self.assertRaises(ObjNotFoundError): | ||||
| self.storage.check(self.absent_id) | self.storage.check(self.absent_id) | ||||
| with self.assertRaises(Error): | with self.assertRaises(Error): | ||||
| self.storage.check(self.invalid_id) | self.storage.check(self.invalid_id) | ||||
| self.storage.check(self.valid_id) | self.storage.check(self.valid_id) | ||||
| def test_can_get_random(self): | |||||
| self.assertEqual(1, len(list(self.storage.get_random(1)))) | |||||
| self.assertEqual( | |||||
| len(list(self.storage)), len(set(self.storage.get_random(1000))) | |||||
| ) | |||||
| def test_cannot_add(self): | def test_cannot_add(self): | ||||
| new_id = self.storage.add(b"New content") | new_id = self.storage.add(b"New content") | ||||
| result = self.storage.add(self.valid_content, self.valid_id) | result = self.storage.add(self.valid_content, self.valid_id) | ||||
| self.assertIsNone(new_id, self.storage) | self.assertIsNone(new_id, self.storage) | ||||
| self.assertIsNone(result) | self.assertIsNone(result) | ||||
| def test_cannot_restore(self): | def test_cannot_restore(self): | ||||
| result = self.storage.restore(self.valid_content, self.valid_id) | result = self.storage.restore(self.valid_content, self.valid_id) | ||||
| ▲ Show 20 Lines • Show All 159 Lines • ▼ Show 20 Lines | def test_check(self): | ||||
| self.storage.check(self.present_corrupted_valid_id) | self.storage.check(self.present_corrupted_valid_id) | ||||
| with self.assertRaises(ObjNotFoundError): | with self.assertRaises(ObjNotFoundError): | ||||
| self.storage.check(self.present_corrupted_invalid_id) | self.storage.check(self.present_corrupted_invalid_id) | ||||
| with self.assertRaises(ObjNotFoundError): | with self.assertRaises(ObjNotFoundError): | ||||
| self.storage.check(self.missing_corrupted_valid_id) | self.storage.check(self.missing_corrupted_valid_id) | ||||
| with self.assertRaises(ObjNotFoundError): | with self.assertRaises(ObjNotFoundError): | ||||
| self.storage.check(self.missing_corrupted_invalid_id) | self.storage.check(self.missing_corrupted_invalid_id) | ||||
| def test_get_random(self): | |||||
| self.assertEqual(0, len(list(self.storage.get_random(0)))) | |||||
| random_content = list(self.storage.get_random(1000)) | |||||
| self.assertIn(self.present_valid_id, random_content) | |||||
| self.assertNotIn(self.present_invalid_id, random_content) | |||||
| self.assertNotIn(self.missing_valid_id, random_content) | |||||
| self.assertNotIn(self.missing_invalid_id, random_content) | |||||
| self.assertIn(self.present_corrupted_valid_id, random_content) | |||||
| self.assertNotIn(self.present_corrupted_invalid_id, random_content) | |||||
| self.assertNotIn(self.missing_corrupted_valid_id, random_content) | |||||
| self.assertNotIn(self.missing_corrupted_invalid_id, random_content) | |||||
| def test_add(self): | def test_add(self): | ||||
| # Add valid and invalid contents to the storage and check their | # Add valid and invalid contents to the storage and check their | ||||
| # presence with the unfiltered storage. | # presence with the unfiltered storage. | ||||
| valid_content = self.ensure_valid(b"ulepsrjbgt") | valid_content = self.ensure_valid(b"ulepsrjbgt") | ||||
| valid_id = compute_hash(valid_content) | valid_id = compute_hash(valid_content) | ||||
| invalid_content = self.ensure_invalid(b"znvghkjked") | invalid_content = self.ensure_invalid(b"znvghkjked") | ||||
| invalid_id = compute_hash(invalid_content) | invalid_id = compute_hash(invalid_content) | ||||
| self.storage.add(valid_content, obj_id=valid_id) | self.storage.add(valid_content, obj_id=valid_id) | ||||
| ▲ Show 20 Lines • Show All 53 Lines • Show Last 20 Lines | |||||