KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_bulk_delete method in tempest

Best Python code snippet using tempest_python

test_object.py

Source:test_object.py

...95        ossfs.rm("nonexistent")96    ossfs.touch(nest_file)97    ossfs.rm(test_path + "/test_rm/nested", recursive=True)98    assert not ossfs.exists(nest_file)99def test_bulk_delete(ossfs, test_path):100    nest_file1 = test_path + "/test_bulk_delete/nested/nested2/file1"101    nest_file2 = test_path + "/test_bulk_delete/nested/nested2/file2"102    ossfs.touch(nest_file1)103    ossfs.touch(nest_file2)104    filelist = ossfs.find(test_path + "/test_bulk_delete")105    ossfs.rm(filelist)106    assert not ossfs.exists(107        test_path + "/test_bulk_delete/nested/nested2/file1"108    )109def test_ossfs_file_access(ossfs, number_file):110    data = b"1234567890\n"111    assert ossfs.cat(number_file) == data112    assert ossfs.head(number_file, 3) == data[:3]113    assert ossfs.tail(number_file, 3) == data[-3:]...

test_fsdb.py

Source:test_fsdb.py

...60        self.assertTrue(self.db.delete(info.identifier, bucket), 'delete failed')61        with self.assertRaises(mod.NotFound):62            self.db.get(info.identifier, bucket)63        return info, bucket64    def test_bulk_delete(self):65        blobs = [66            ('test.5', 'doc.5'),67            ('test.6', 'doc.6'),68        ]69        infos = [70            self.db.put(StringIO(b"content-{}".format(blob[0])), get_id(), bucket=blob[1])71            for blob in blobs72        ]73        blob_infos = zip(blobs, infos)74        paths = [self.db.get_path(info.identifier, blob[1]) for blob, info in blob_infos]75        self.assertTrue(self.db.bulk_delete(paths), 'delete failed')76        for blob, info in blob_infos:77            with self.assertRaises(mod.NotFound):78                self.db.get(info.identifier, blob[1])79        return paths80    def test_delete_bucket(self):81        bucket = join("doctype", "ys7v136b")82        info = self.db.put(StringIO(b"content"), get_id(), bucket=bucket)83        self.assertTrue(self.db.delete(bucket=bucket))84        self.assertTrue(info.identifier)85        with self.assertRaises(mod.NotFound):86            self.db.get(info.identifier, bucket=bucket)87    def test_delete_identifier_in_default_bucket(self):88        info = self.db.put(StringIO(b"content"), get_id())89        self.assertTrue(self.db.delete(info.identifier), 'delete failed')90        with self.assertRaises(mod.NotFound):91            self.db.get(info.identifier)92    def test_delete_no_args(self):93        info = self.db.put(StringIO(b"content"), get_id())94        with self.assertRaises(ArgumentError):95            self.db.delete()96        # blobs in default bucket should not be deleted97        with self.db.get(info.identifier) as fh:98            self.assertEqual(fh.read(), b"content")99        self.assertTrue(self.db.delete(bucket=mod.DEFAULT_BUCKET))100    def test_prevent_delete_bucket_by_mistake(self):101        info = self.db.put(StringIO(b"content"), get_id())102        id_mistake = None103        with self.assertRaises(ArgumentError):104            self.db.delete(id_mistake, mod.DEFAULT_BUCKET)105        # blobs in default bucket should not be deleted106        with self.db.get(info.identifier) as fh:107            self.assertEqual(fh.read(), b"content")108        self.assertTrue(self.db.delete(bucket=mod.DEFAULT_BUCKET))109    def test_empty_attachment_name(self):110        info = self.db.put(StringIO(b"content"), get_id())111        self.assertNotIn(".", info.identifier)112        return info113@generate_cases([114    ("test.1", "\u4500.1"),115    ("test.1", "/tmp/notallowed"),116    ("test.1", "."),117    ("test.1", ".."),118    ("test.1", "../notallowed"),119    ("test.1", "notallowed/.."),120    ("/test.1",),121    ("../test.1",),122], _BlobDBTests)123def test_bad_name(self, name, bucket=mod.DEFAULT_BUCKET):124    with self.assertRaises(mod.BadName):125        self.db.get(name, bucket)126class TestFilesystemBlobDB(TestCase, _BlobDBTests):127    @classmethod128    def setUpClass(cls):129        super(TestFilesystemBlobDB, cls).setUpClass()130        cls.rootdir = mkdtemp(prefix="blobdb")131        cls.db = mod.FilesystemBlobDB(cls.rootdir)132    @classmethod133    def tearDownClass(cls):134        cls.db = None135        rmtree(cls.rootdir)136        cls.rootdir = None137        super(TestFilesystemBlobDB, cls).tearDownClass()138    def test_put_with_colliding_blob_id(self):139        # unfortunately can't do this on S3 because there is no way to140        # reliably check if an object exists before putting it.141        ident = get_id()142        self.db.put(StringIO(b"bing"), ident)143        with self.assertRaises(mod.FileExists):144            self.db.put(StringIO(b"bang"), ident)145    def test_delete(self):146        info, bucket = super(TestFilesystemBlobDB, self).test_delete()147        self.assertFalse(self.db.delete(info.identifier, bucket), 'delete should fail')148    def test_bulk_delete(self):149        paths = super(TestFilesystemBlobDB, self).test_bulk_delete()150        self.assertFalse(self.db.bulk_delete(paths), 'delete should fail')151    def test_delete_bucket(self):152        super(TestFilesystemBlobDB, self).test_delete_bucket()153        names = os.listdir(self.db.get_path(bucket="doctype"))154        self.assertNotIn("ys7v136b", names, "bucket not deleted")155    def test_bucket_path(self):156        bucket = join("doctype", "8cd98f0")157        self.db.put(StringIO(b"content"), get_id(), bucket=bucket)158        path = self.db.get_path(bucket=bucket)159        self.assertTrue(isdir(path), path)160        self.assertTrue(os.listdir(path))161    def test_empty_attachment_name(self):162        info = super(TestFilesystemBlobDB, self).test_empty_attachment_name()163        path = self.db.get_path(info.identifier)...

tests.py

Source:tests.py

...17        for index_name in self.index_names:18            index, is_created = Index.objects.get_or_create(name=index_name)19            self.ec.es.indices.create(index=index_name, ignore=[400, 404])20            self.ids.append(index.pk)21    def test_bulk_delete(self):22        url = reverse("v2:index-bulk-delete")23        response = self.client.post(url, data={"ids": self.ids}, format="json")24        self.assertTrue(response.status_code == status.HTTP_200_OK)25        for index_name in self.index_names:26            self.assertFalse(self.ec.es.indices.exists(index_name))27        print_output("test_bulk_delete:response.data", response.data)28    def tearDown(self) -> None:29        indices = Index.objects.filter(pk__in=self.ids)30        names = [index.name for index in indices]31        if names:32            indices.delete()33            for index in names:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.